diff --git a/nwb_linkml/src/nwb_linkml/io/schema.py b/nwb_linkml/src/nwb_linkml/io/schema.py index 954fb3a..3fd8aec 100644 --- a/nwb_linkml/src/nwb_linkml/io/schema.py +++ b/nwb_linkml/src/nwb_linkml/io/schema.py @@ -5,6 +5,7 @@ Loading/saving NWB Schema yaml files from pathlib import Path from pprint import pprint from typing import Optional +import warnings from linkml_runtime.loaders import yaml_loader @@ -82,6 +83,8 @@ def load_namespace_adapter( version (str): Optional: tag or commit to check out namespace is a :class:`.NamespaceRepo`. If ``None``, use ``HEAD`` if not already checked out, or otherwise use whatever version is already checked out. + imported (list[:class:`.NamespacesAdapter`]): Optional: override discovered imports + with already-loaded namespaces adapters Returns: :class:`.NamespacesAdapter` @@ -111,10 +114,17 @@ def load_namespace_adapter( for ns in namespaces.namespaces: for schema in ns.schema_: if schema.source is None: - # this is normal, we'll resolve later - continue - yml_file = (path / schema.source).resolve() - sch.append(load_schema_file(yml_file)) + if imported is None and schema.namespace == "hdmf-common": + # special case - hdmf-common is imported by name without location or version, + # so to get the correct version we have to handle it separately + imported = _resolve_hdmf(namespace, path) + if imported is not None: + imported = [imported] + else: + continue + else: + yml_file = (path / schema.source).resolve() + sch.append(load_schema_file(yml_file)) if imported is not None: adapter = NamespacesAdapter(namespaces=namespaces, schemas=sch, imported=imported) @@ -124,6 +134,31 @@ def load_namespace_adapter( return adapter +def _resolve_hdmf( + namespace: Path | NamespaceRepo | Namespaces, path: Optional[Path] = None +) -> Optional[NamespacesAdapter]: + if path is None and isinstance(namespace, Namespaces): + # cant get any more information from already-loaded namespaces without a path + return None + + if isinstance(namespace, NamespaceRepo): + # easiest route is if we got a NamespaceRepo + if namespace.name == "core": + hdmf_path = (path / namespace.imports["hdmf-common"]).resolve() + return load_namespace_adapter(namespace=hdmf_path) + # otherwise the hdmf-common adapter itself, and it loads common + else: + return None + elif path is not None: + # otherwise try and get it from relative paths + # pretty much a hack, but hey we are compensating for absence of versioning system here + maybe_repo_root = path / NWB_CORE_REPO.imports["hdmf-common"] + if maybe_repo_root.exists(): + return load_namespace_adapter(namespace=maybe_repo_root) + warnings.warn(f"Could not locate hdmf-common from namespace {namespace} and path {path}") + return None + + def load_nwb_core( core_version: str = "2.7.0", hdmf_version: str = "1.8.0", hdmf_only: bool = False ) -> NamespacesAdapter: diff --git a/nwb_linkml/src/nwb_linkml/providers/git.py b/nwb_linkml/src/nwb_linkml/providers/git.py index 05ba68b..8219aaf 100644 --- a/nwb_linkml/src/nwb_linkml/providers/git.py +++ b/nwb_linkml/src/nwb_linkml/providers/git.py @@ -36,6 +36,14 @@ class NamespaceRepo(BaseModel): ), default_factory=list, ) + imports: Optional[dict[str, Path]] = Field( + None, + description=( + "Any named imports that are included eg. as submodules within their repository. Dict" + " mapping schema name (used in the namespace field) to the namespace file relative to" + " the directory containing the **namespace.yaml file** (not the repo root)" + ), + ) def provide_from_git(self, commit: str | None = None) -> Path: """Provide a namespace file from a git repo""" @@ -61,6 +69,7 @@ NWB_CORE_REPO = NamespaceRepo( "2.6.0", "2.7.0", ], + imports={"hdmf-common": Path("../hdmf-common-schema") / "common" / "namespace.yaml"}, ) HDMF_COMMON_REPO = NamespaceRepo( @@ -86,7 +95,7 @@ HDMF_COMMON_REPO = NamespaceRepo( DEFAULT_REPOS = { repo.name: repo for repo in [NWB_CORE_REPO, HDMF_COMMON_REPO] -} # type: Dict[str, NamespaceRepo] +} # type: dict[str, NamespaceRepo] class GitError(OSError): @@ -112,7 +121,7 @@ class GitRepo: self.namespace = namespace self._commit = commit - def _git_call(self, *args: List[str]) -> subprocess.CompletedProcess: + def _git_call(self, *args: str) -> subprocess.CompletedProcess: res = subprocess.run(["git", "-C", self.temp_directory, *args], capture_output=True) if res.returncode != 0: raise GitError( @@ -138,8 +147,11 @@ class GitRepo: """ URL for "origin" remote """ - res = self._git_call("remote", "get-url", "origin") - return res.stdout.decode("utf-8").strip() + try: + res = self._git_call("remote", "get-url", "origin") + return res.stdout.decode("utf-8").strip() + except GitError: + return "" @property def active_commit(self) -> str: @@ -157,6 +169,16 @@ class GitRepo: """ return self.temp_directory / self.namespace.path + @property + def import_namespaces(self) -> dict[str, Path]: + """ + Absolute location of each of the imported namespaces specified in + :attr:`.NamespaceRepo.imports` + """ + if self.namespace.imports is None: + return {} + return {k: (self.namespace_file / v).resolve() for k, v in self.namespace.imports.items()} + @property def commit(self) -> Optional[str]: """