diff --git a/scripts/download_schema.py b/scripts/download_schema.py index ec00b9252..423899b63 100755 --- a/scripts/download_schema.py +++ b/scripts/download_schema.py @@ -2,15 +2,40 @@ import click from allotropy.schema_gen.fetcher import SchemaFetcher +from allotropy.schema_gen.technique_resolver import ( + is_shorthand, + resolve_shorthand_to_urls, +) @click.command() -@click.argument("schema_url") -def _download_schema(schema_url: str) -> None: - print(f"Downloading schema at {schema_url}...") +@click.argument("schema_input") +def _download_schema(schema_input: str) -> None: + """Download an Allotrope schema and its dependencies. + + SCHEMA_INPUT can be a full purl URL or a shorthand like "plate-reader 2026/03". + + \b + Examples: + plate-reader 2026/03 + pcr WD/2025/06 + http://purl.allotrope.org/json-schemas/adm/pcr/REC/2024/09/qpcr.schema + """ + if is_shorthand(schema_input): + urls = resolve_shorthand_to_urls(schema_input) + click.echo(f"Resolved to {len(urls)} schema(s):") + for url in urls: + click.echo(f" {url}") + else: + urls = [schema_input] + fetcher = SchemaFetcher() - schemas = fetcher.fetch_with_dependencies(schema_url) - print(f"Downloaded {len(schemas)} schema(s) to cache") + total = 0 + for url in urls: + click.echo(f"Downloading schema at {url}...") + schemas = fetcher.fetch_with_dependencies(url) + total += len(schemas) + click.echo(f"Downloaded {total} schema(s) to cache") if __name__ == "__main__": diff --git a/scripts/generate_schemas.py b/scripts/generate_schemas.py index 956023510..c6e0dee1f 100755 --- a/scripts/generate_schemas.py +++ b/scripts/generate_schemas.py @@ -5,6 +5,10 @@ _discover_cached_technique_urls, generate_models, ) +from allotropy.schema_gen.technique_resolver import ( + is_shorthand, + resolve_shorthand_to_urls, +) @click.command() @@ -20,15 +24,32 @@ def _generate_schemas( ) -> None: """Generate Python models from one or more Allotrope schema URLs. + Each argument can be a full purl URL or a shorthand like "plate-reader 2026/03". Use --all to regenerate every cached technique schema. + + \b + Examples: + plate-reader 2026/03 + pcr WD/2025/06 + http://purl.allotrope.org/json-schemas/adm/pcr/REC/2024/09/qpcr.schema """ + urls: list[str] if regenerate_all: urls = _discover_cached_technique_urls() click.echo(f"Discovered {len(urls)} cached technique schema(s)") elif schema_urls: - urls = list(schema_urls) + urls = [] + for arg in schema_urls: + if is_shorthand(arg): + resolved = resolve_shorthand_to_urls(arg) + click.echo(f"Resolved '{arg}' to {len(resolved)} schema(s):") + for u in resolved: + click.echo(f" {u}") + urls.extend(resolved) + else: + urls.append(arg) else: - msg = "Provide schema URLs or use --all." + msg = "Provide schema URLs/shorthands or use --all." raise click.UsageError(msg) generate_models(urls) diff --git a/src/allotropy/schema_gen/technique_resolver.py b/src/allotropy/schema_gen/technique_resolver.py new file mode 100644 index 000000000..d30790c9e --- /dev/null +++ b/src/allotropy/schema_gen/technique_resolver.py @@ -0,0 +1,267 @@ +"""Resolve technique shorthand (e.g. "plate-reader 2026/03") to purl URLs. + +Supports fuzzy matching of technique names against the GitLab directory listing +(with local schema cache fallback) and automatic discovery of all schema files +within a technique+version directory. +""" + +from __future__ import annotations + +import difflib +import json +from pathlib import Path +import re +import ssl +from urllib.error import HTTPError, URLError +from urllib.request import urlopen + +import click + +from allotropy.schema_gen.naming import ALLOTROPE_URL_PREFIX, DEFAULT_SCHEMA_CACHE_DIR + +# GitLab API base for the Allotrope public ASM repository +_GITLAB_API_BASE = ( + "https://gitlab.com/api/v4/projects/allotrope-public%2Fasm/repository/tree" +) + +_VALID_STATUSES = {"REC", "WD", "BENCHLING"} + +# Matches: "plate-reader 2026/03" or "pcr REC/2026/03" or "pcr benchling/2024/11" +_RE_SHORTHAND = re.compile( + r"^(?P\S+)" + r"\s+" + r"(?:(?P[a-zA-Z]+)/)?" + r"(?P\d{4})/(?P\d{2})$", +) + + +def is_shorthand(input_str: str) -> bool: + """Return True if the input is a technique shorthand, not a full URL.""" + return not input_str.startswith(("http://", "https://")) + + +def parse_shorthand(input_str: str) -> tuple[str, str, str, str]: + """Parse shorthand into (technique, status, year, month). + + Status defaults to "REC" if not provided. + + Raises: + click.UsageError: If the input doesn't match the expected format. + """ + match = _RE_SHORTHAND.match(input_str.strip()) + if not match: + msg = ( + f"Invalid shorthand format: '{input_str}'\n" + "Expected: [/]/\n" + "Examples: plate-reader 2026/03, pcr WD/2025/06" + ) + raise click.UsageError(msg) + + technique = match.group("technique").lower() + status = (match.group("status") or "REC").upper() + year = match.group("year") + month = match.group("month") + + if status not in _VALID_STATUSES: + msg = f"Invalid status '{status}'. Must be one of: {', '.join(sorted(_VALID_STATUSES))}" + raise click.UsageError(msg) + + return technique, status, year, month + + +def _get_ssl_context() -> ssl.SSLContext: + """Build an SSL context that works with corporate proxies. + + The hatch virtualenv Python may not trust the system certificate store + (e.g. on macOS the framework Python uses its own OpenSSL cert bundle). + We try the default context first, then fall back to known system cert + locations before giving up. + """ + # macOS system cert bundle, Linux common locations + system_ca_files = ( + "/private/etc/ssl/cert.pem", # macOS + "/etc/ssl/certs/ca-certificates.crt", # Debian/Ubuntu + "/etc/pki/tls/certs/ca-bundle.crt", # RHEL/CentOS + ) + ctx = ssl.create_default_context() + for ca_file in system_ca_files: + if Path(ca_file).exists(): + try: + ctx.load_verify_locations(ca_file) + return ctx + except ssl.SSLError: + continue + return ctx + + +def _gitlab_tree(path: str) -> list[dict[str, str]] | None: + """Fetch a directory listing from the GitLab API. + + Returns None on any network/HTTP error instead of raising. + """ + url = f"{_GITLAB_API_BASE}?path={path}&per_page=100&ref=main" + try: + ctx = _get_ssl_context() + with urlopen(url, timeout=30, context=ctx) as response: # noqa: S310 + data = json.loads(response.read().decode("utf-8")) + except (HTTPError, URLError, OSError): + return None + return data # type: ignore[no-any-return] + + +def _list_cached_techniques( + cache_dir: Path = DEFAULT_SCHEMA_CACHE_DIR, +) -> list[str]: + """List technique names from the local schema cache.""" + adm_dir = cache_dir / "adm" + if not adm_dir.is_dir(): + return [] + return sorted( + d.name + for d in adm_dir.iterdir() + if d.is_dir() and d.name not in ("core", "qudt") + ) + + +def list_gitlab_techniques() -> list[str]: + """List technique names, preferring GitLab API with local cache fallback.""" + entries = _gitlab_tree("json-schemas/adm") + if entries is not None: + return sorted(entry["name"] for entry in entries if entry.get("type") == "tree") + # Fallback to local cache + click.echo( + "Warning: Could not reach GitLab API, using local schema cache for technique names." + ) + techniques = _list_cached_techniques() + if not techniques: + msg = ( + "Could not reach GitLab API and no local schema cache found.\n" + "Check your network connection and try again." + ) + raise click.ClickException(msg) + return techniques + + +def resolve_technique_name(input_name: str, techniques: list[str]) -> str: + """Resolve a technique name with exact, normalized, or fuzzy matching. + + Returns: + The matched technique name. + + Raises: + click.Abort: If no match is found or the user declines the suggestion. + """ + # Exact match + if input_name in techniques: + return input_name + + # Normalized match: underscores -> hyphens, lowercase + normalized = input_name.replace("_", "-").lower() + if normalized in techniques: + return normalized + + # Fuzzy match + matches = difflib.get_close_matches(normalized, techniques, n=5, cutoff=0.6) + + if not matches: + click.echo(f"Technique '{input_name}' not found. No similar techniques.") + click.echo(f"Available techniques: {', '.join(techniques)}") + raise click.Abort() + + if len(matches) == 1: + if click.confirm( + f"Technique '{input_name}' not found. Did you mean '{matches[0]}'?" + ): + return matches[0] + raise click.Abort() + + click.echo(f"Technique '{input_name}' not found. Similar techniques:") + for i, name in enumerate(matches, 1): + click.echo(f" {i}. {name}") + choice: int = click.prompt( + "Select a technique", + type=click.IntRange(1, len(matches)), + ) + return matches[choice - 1] + + +def _list_cached_schemas( + technique: str, + status: str, + year: str, + month: str, + cache_dir: Path = DEFAULT_SCHEMA_CACHE_DIR, +) -> list[str]: + """List schema filenames from the local cache directory.""" + cache_path = cache_dir / "adm" / technique / status / year / month + if not cache_path.is_dir(): + return [] + return sorted( + f.name + for f in cache_path.iterdir() + if f.name.endswith(".schema.json") and not f.name.endswith(".embed.schema.json") + ) + + +def list_schemas_in_directory( + technique: str, status: str, year: str, month: str +) -> list[str]: + """List schema filenames in a technique+version directory. + + Tries GitLab API first, falls back to local cache. + Returns only *.schema.json files, excluding *.embed.schema.json. + """ + path = f"json-schemas/adm/{technique}/{status}/{year}/{month}" + entries = _gitlab_tree(path) + + if entries is not None: + return sorted( + entry["name"] + for entry in entries + if entry.get("type") == "blob" + and entry["name"].endswith(".schema.json") + and not entry["name"].endswith(".embed.schema.json") + ) + + # Fallback to local cache + cached = _list_cached_schemas(technique, status, year, month) + if cached: + click.echo( + f"Warning: Could not reach GitLab API, using local cache for {technique}/{status}/{year}/{month}." + ) + return cached + + msg = ( + f"No schemas found at {technique}/{status}/{year}/{month}.\n" + "Could not reach GitLab API and no local cache exists for this version.\n" + "Check your network connection or download the schema manually first." + ) + raise click.ClickException(msg) + + +def resolve_shorthand_to_urls(input_str: str) -> list[str]: + """Resolve a technique shorthand string to a list of purl URLs. + + This is the main entry point. It parses the shorthand, resolves the + technique name (with fuzzy matching if needed), discovers all schema + files in the directory, and constructs purl URLs. + """ + technique, status, year, month = parse_shorthand(input_str) + + click.echo(f"Looking up technique '{technique}'...") + techniques = list_gitlab_techniques() + technique = resolve_technique_name(technique, techniques) + + schema_files = list_schemas_in_directory(technique, status, year, month) + if not schema_files: + msg = f"No schema files found in {technique}/{status}/{year}/{month}" + raise click.UsageError(msg) + + urls = [] + for filename in schema_files: + # Remove .json suffix to get the purl-style path (e.g. plate-reader.schema) + schema_name = filename.removesuffix(".json") + url = f"{ALLOTROPE_URL_PREFIX}adm/{technique}/{status}/{year}/{month}/{schema_name}" + urls.append(url) + + return urls diff --git a/tests/schema_gen/test_technique_resolver.py b/tests/schema_gen/test_technique_resolver.py new file mode 100644 index 000000000..20beb7cfc --- /dev/null +++ b/tests/schema_gen/test_technique_resolver.py @@ -0,0 +1,362 @@ +"""Tests for allotropy.schema_gen.technique_resolver.""" + +from __future__ import annotations + +from io import BytesIO +import json +from pathlib import Path +from typing import Any, ClassVar +from unittest.mock import patch + +import click +import pytest + +from allotropy.schema_gen.technique_resolver import ( + _list_cached_schemas, + _list_cached_techniques, + is_shorthand, + list_gitlab_techniques, + list_schemas_in_directory, + parse_shorthand, + resolve_shorthand_to_urls, + resolve_technique_name, +) + +ALLOTROPE_PREFIX = "http://purl.allotrope.org/json-schemas/" + + +class TestIsShorthand: + def test_full_http_url(self) -> None: + assert not is_shorthand( + "http://purl.allotrope.org/json-schemas/adm/pcr/REC/2024/09/qpcr.schema" + ) + + def test_full_https_url(self) -> None: + assert not is_shorthand( + "https://gitlab.com/allotrope-public/asm/-/raw/main/json-schemas/adm/pcr/REC/2024/09/qpcr.schema.json" + ) + + def test_shorthand_simple(self) -> None: + assert is_shorthand("plate-reader 2026/03") + + def test_shorthand_with_status(self) -> None: + assert is_shorthand("pcr WD/2025/06") + + def test_shorthand_with_underscores(self) -> None: + assert is_shorthand("plate_reader 2026/03") + + +class TestParseShorthand: + def test_simple(self) -> None: + assert parse_shorthand("plate-reader 2026/03") == ( + "plate-reader", + "REC", + "2026", + "03", + ) + + def test_with_status_rec(self) -> None: + assert parse_shorthand("pcr REC/2024/09") == ("pcr", "REC", "2024", "09") + + def test_with_status_wd(self) -> None: + assert parse_shorthand("pcr WD/2025/06") == ("pcr", "WD", "2025", "06") + + def test_with_status_benchling(self) -> None: + assert parse_shorthand("chromatography BENCHLING/2024/11") == ( + "chromatography", + "BENCHLING", + "2024", + "11", + ) + + def test_case_insensitive_status(self) -> None: + assert parse_shorthand("pcr rec/2024/09") == ("pcr", "REC", "2024", "09") + + def test_case_insensitive_technique(self) -> None: + technique, _, _, _ = parse_shorthand("Plate-Reader 2026/03") + assert technique == "plate-reader" + + def test_underscores_in_technique(self) -> None: + technique, _, _, _ = parse_shorthand("plate_reader 2026/03") + assert technique == "plate_reader" + + def test_strips_whitespace(self) -> None: + assert parse_shorthand(" plate-reader 2026/03 ") == ( + "plate-reader", + "REC", + "2026", + "03", + ) + + def test_invalid_format_no_version(self) -> None: + with pytest.raises(click.UsageError, match="Invalid shorthand format"): + parse_shorthand("plate-reader") + + def test_invalid_format_garbage(self) -> None: + with pytest.raises(click.UsageError, match="Invalid shorthand format"): + parse_shorthand("just some random text") + + def test_invalid_status(self) -> None: + with pytest.raises(click.UsageError, match="Invalid status"): + parse_shorthand("pcr FAKE/2024/09") + + +def _mock_urlopen(entries: list[dict[str, Any]]) -> BytesIO: + """Create a mock urlopen response with the given JSON entries.""" + data = json.dumps(entries).encode("utf-8") + return BytesIO(data) + + +def _tree_entry(name: str, entry_type: str = "tree") -> dict[str, str]: + """Create a minimal GitLab API tree entry.""" + return {"name": name, "type": entry_type, "path": f"json-schemas/adm/{name}"} + + +def _blob_entry(name: str, path: str = "") -> dict[str, str]: + return {"name": name, "type": "blob", "path": path or name} + + +class TestListGitlabTechniques: + def test_returns_sorted_tree_entries(self) -> None: + entries = [ + _tree_entry("pcr"), + _tree_entry("absorbance"), + _tree_entry("plate-reader"), + ] + with patch( + "allotropy.schema_gen.technique_resolver.urlopen", + return_value=_mock_urlopen(entries), + ): + result = list_gitlab_techniques() + assert result == ["absorbance", "pcr", "plate-reader"] + + def test_filters_non_tree_entries(self) -> None: + entries = [ + _tree_entry("pcr"), + _blob_entry("README.md"), + _tree_entry("absorbance"), + ] + with patch( + "allotropy.schema_gen.technique_resolver.urlopen", + return_value=_mock_urlopen(entries), + ): + result = list_gitlab_techniques() + assert result == ["absorbance", "pcr"] + + +class TestResolveTechniqueName: + TECHNIQUES: ClassVar[list[str]] = [ + "absorbance", + "cell-counting", + "cell-culture-analyzer", + "pcr", + "plate-reader", + "solution-analyzer", + ] + + def test_exact_match(self) -> None: + assert resolve_technique_name("plate-reader", self.TECHNIQUES) == "plate-reader" + + def test_normalized_match_underscores(self) -> None: + assert resolve_technique_name("plate_reader", self.TECHNIQUES) == "plate-reader" + + def test_normalized_match_uppercase(self) -> None: + assert resolve_technique_name("PCR", self.TECHNIQUES) == "pcr" + + def test_fuzzy_single_match_accepted(self) -> None: + with patch("click.confirm", return_value=True): + result = resolve_technique_name("platereader", self.TECHNIQUES) + assert result == "plate-reader" + + def test_fuzzy_single_match_declined(self) -> None: + with patch("click.confirm", return_value=False): + with pytest.raises(click.Abort): + resolve_technique_name("platereader", self.TECHNIQUES) + + def test_no_matches_aborts(self) -> None: + with pytest.raises(click.Abort): + resolve_technique_name("xyzzy", self.TECHNIQUES) + + +class TestListSchemasInDirectory: + def test_returns_schema_files(self) -> None: + entries = [ + _blob_entry("plate-reader.schema.json"), + _blob_entry("plate-reader.embed.schema.json"), + ] + with patch( + "allotropy.schema_gen.technique_resolver.urlopen", + return_value=_mock_urlopen(entries), + ): + result = list_schemas_in_directory("plate-reader", "REC", "2026", "03") + assert result == ["plate-reader.schema.json"] + + def test_multiple_schemas(self) -> None: + entries = [ + _blob_entry("dpcr.schema.json"), + _blob_entry("dpcr.embed.schema.json"), + _blob_entry("qpcr.schema.json"), + _blob_entry("qpcr.embed.schema.json"), + ] + with patch( + "allotropy.schema_gen.technique_resolver.urlopen", + return_value=_mock_urlopen(entries), + ): + result = list_schemas_in_directory("pcr", "REC", "2026", "03") + assert result == ["dpcr.schema.json", "qpcr.schema.json"] + + def test_filters_directories(self) -> None: + entries = [ + _tree_entry("subdir"), + _blob_entry("test.schema.json"), + ] + with patch( + "allotropy.schema_gen.technique_resolver.urlopen", + return_value=_mock_urlopen(entries), + ): + result = list_schemas_in_directory("test", "REC", "2026", "03") + assert result == ["test.schema.json"] + + +class TestResolveShorthandToUrls: + def _mock_techniques(self) -> list[dict[str, str]]: + return [ + _tree_entry("pcr"), + _tree_entry("plate-reader"), + ] + + def _mock_schemas(self) -> list[dict[str, str]]: + return [ + _blob_entry("plate-reader.schema.json"), + _blob_entry("plate-reader.embed.schema.json"), + ] + + def test_resolves_to_purl_urls(self) -> None: + techniques = self._mock_techniques() + schemas = self._mock_schemas() + + def mock_urlopen(url: str, **_kwargs: Any) -> BytesIO: + if "path=json-schemas/adm&" in url: + return _mock_urlopen(techniques) + return _mock_urlopen(schemas) + + with patch( + "allotropy.schema_gen.technique_resolver.urlopen", + side_effect=mock_urlopen, + ): + result = resolve_shorthand_to_urls("plate-reader 2026/03") + + assert result == [ + f"{ALLOTROPE_PREFIX}adm/plate-reader/REC/2026/03/plate-reader.schema" + ] + + def test_multi_schema_technique(self) -> None: + techniques = self._mock_techniques() + pcr_schemas = [ + _blob_entry("dpcr.schema.json"), + _blob_entry("dpcr.embed.schema.json"), + _blob_entry("qpcr.schema.json"), + _blob_entry("qpcr.embed.schema.json"), + ] + + def mock_urlopen(url: str, **_kwargs: Any) -> BytesIO: + if "path=json-schemas/adm&" in url: + return _mock_urlopen(techniques) + return _mock_urlopen(pcr_schemas) + + with patch( + "allotropy.schema_gen.technique_resolver.urlopen", + side_effect=mock_urlopen, + ): + result = resolve_shorthand_to_urls("pcr 2026/03") + + assert result == [ + f"{ALLOTROPE_PREFIX}adm/pcr/REC/2026/03/dpcr.schema", + f"{ALLOTROPE_PREFIX}adm/pcr/REC/2026/03/qpcr.schema", + ] + + +class TestCacheFallback: + def test_list_cached_techniques(self, tmp_path: Path) -> None: + adm = tmp_path / "adm" + (adm / "pcr").mkdir(parents=True) + (adm / "plate-reader").mkdir() + (adm / "core").mkdir() # should be excluded + (adm / "qudt").mkdir() # should be excluded + result = _list_cached_techniques(tmp_path) + assert result == ["pcr", "plate-reader"] + + def test_list_cached_techniques_empty(self, tmp_path: Path) -> None: + assert _list_cached_techniques(tmp_path) == [] + + def test_list_cached_schemas(self, tmp_path: Path) -> None: + schema_dir = tmp_path / "adm" / "pcr" / "REC" / "2026" / "03" + schema_dir.mkdir(parents=True) + (schema_dir / "qpcr.schema.json").write_text("{}") + (schema_dir / "qpcr.embed.schema.json").write_text("{}") + (schema_dir / "dpcr.schema.json").write_text("{}") + result = _list_cached_schemas("pcr", "REC", "2026", "03", tmp_path) + assert result == ["dpcr.schema.json", "qpcr.schema.json"] + + def test_list_cached_schemas_missing_dir(self, tmp_path: Path) -> None: + assert _list_cached_schemas("pcr", "REC", "2026", "03", tmp_path) == [] + + def test_list_gitlab_techniques_falls_back_to_cache(self) -> None: + """When GitLab API fails, falls back to local cache.""" + with ( + patch( + "allotropy.schema_gen.technique_resolver._gitlab_tree", + return_value=None, + ), + patch( + "allotropy.schema_gen.technique_resolver._list_cached_techniques", + return_value=["pcr", "plate-reader"], + ), + ): + result = list_gitlab_techniques() + assert result == ["pcr", "plate-reader"] + + def test_list_gitlab_techniques_no_cache_raises(self) -> None: + """When GitLab API fails and no cache, raises ClickException.""" + with ( + patch( + "allotropy.schema_gen.technique_resolver._gitlab_tree", + return_value=None, + ), + patch( + "allotropy.schema_gen.technique_resolver._list_cached_techniques", + return_value=[], + ), + pytest.raises(click.ClickException), + ): + list_gitlab_techniques() + + def test_list_schemas_falls_back_to_cache(self) -> None: + """When GitLab API fails for schema listing, falls back to local cache.""" + with ( + patch( + "allotropy.schema_gen.technique_resolver._gitlab_tree", + return_value=None, + ), + patch( + "allotropy.schema_gen.technique_resolver._list_cached_schemas", + return_value=["plate-reader.schema.json"], + ), + ): + result = list_schemas_in_directory("plate-reader", "REC", "2026", "03") + assert result == ["plate-reader.schema.json"] + + def test_list_schemas_no_cache_raises(self) -> None: + """When GitLab API fails and no cache, raises ClickException.""" + with ( + patch( + "allotropy.schema_gen.technique_resolver._gitlab_tree", + return_value=None, + ), + patch( + "allotropy.schema_gen.technique_resolver._list_cached_schemas", + return_value=[], + ), + pytest.raises(click.ClickException), + ): + list_schemas_in_directory("plate-reader", "REC", "2026", "03")