Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 20 additions & 9 deletions scripts/from_bioregistry.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from tqdm.contrib.logging import logging_redirect_tqdm

import bioontologies
from bioontologies.robot import ROBOTError


@click.command()
Expand All @@ -27,19 +28,29 @@ def main() -> None:
resource.prefix
for resource in bioregistry.resources()
if resource.get_download_owl() or resource.get_download_obograph()
][282:]
]
for prefix in tqdm(prefixes, desc="Parsing OWL ontologies", unit="ontology"):
tqdm.write(click.style("\n" + prefix, fg="green"))
tqdm.write(click.style(f"\n[{prefix}]", fg="green"))
with logging_redirect_tqdm():
document = bioontologies.get_obograph_by_prefix(prefix, cache=False)
try:
graph = document.squeeze(standardize=True, prefix=prefix)
except ValueError:
tqdm.write(f"[{prefix}] failed to parse")
else:
tqdm.write(
click.style(f"[{prefix}] parsed {graph.title} - v{graph.version}", fg="red")
document = bioontologies.get_obograph_by_prefix(prefix, cache=False, reason=False)
except ROBOTError as e:
tqdm.write(click.style(f"[{prefix}] {e}", fg="red"))
continue

try:
graph = document.squeeze(
standardize=True, prefix=prefix, tqdm_kwargs={"leave": False}
)
except ValueError as e:
tqdm.write(click.style(f"[{prefix}] failed to parse\n\n{e}", fg="red"))
else:
if graph.version:
msg = f"[{prefix}] parsed {graph.title} - v{graph.version}"
else:
msg = f"[{prefix}] parsed {graph.title}"

tqdm.write(click.style(msg, fg="green"))


if __name__ == "__main__":
Expand Down
61 changes: 55 additions & 6 deletions src/bioontologies/robot.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import os
import subprocess
import tempfile
import textwrap
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
Expand All @@ -24,6 +25,7 @@

__all__ = [
"ParseResults",
"ROBOTError",
"convert",
"convert_to_obograph",
"convert_to_obograph_local",
Expand Down Expand Up @@ -59,7 +61,7 @@ def is_available() -> bool:
return False

try:
check_output(["java", "--help"]) # noqa:S607,S603
check_output(["java", "--help"]) # noqa: S607
except Exception:
logger.error(
"java --help failed - this means the java runtime environment (JRE) "
Expand All @@ -85,11 +87,19 @@ def is_available() -> bool:
def call_robot(args: list[str]) -> str:
"""Run a robot command and return the output as a string."""
rr = ["java", "-jar", str(get_robot_jar_path()), *args]
logger.debug("Running shell command: %s", args)
ret = check_output( # noqa:S603
rr,
cwd=os.path.dirname(__file__),
)
logger.debug("Running shell command: %s", rr)
try:
ret = check_output( # noqa:S603
rr,
cwd=os.path.dirname(__file__),
)
except subprocess.CalledProcessError as e:
raise ROBOTError(
command=e.cmd,
return_code=e.returncode,
output=e.output.decode() if e.output is not None else None,
) from None

return ret.decode()


Expand Down Expand Up @@ -440,6 +450,45 @@ def _path_context(path: None | str | Path, name: str = "output.json"):
yield Path(directory).joinpath(name)


class ROBOTError(Exception):
"""Custom error for ROBOT command failures that includes output preview."""

def __init__(
self,
command: list[str],
return_code: int,
output: str | None = None,
preview_length: int = 500,
) -> None:
"""Initialize a wrapper around a ROBOT exception.

:param command: The command that was executed and failed
:param return_code: The exit code returned by the command
:param output: The stdout/stderr output from the command execution
:param preview_length:
Maximum number of characters to include in the
error message preview. Default is 500 characters.

The error message will contain the command, return code, and a preview
of the output truncated to preview_length characters.
"""
self.command = command
self.return_code = return_code
self.output = output or "<no output>"
self.preview_length = preview_length

# Create the error message
command_str = " ".join(command)
output_preview = textwrap.indent(textwrap.shorten(self.output, preview_length), " ")

message = (
f"Command `{command_str}` returned non-zero exit status {return_code}.\n\n"
f"Output:\n\n{output_preview}"
)

super().__init__(message)


def convert(
input_path: str | Path,
output_path: str | Path,
Expand Down
Loading