eager resolution of hdmf namespace when loaded from yaml

This commit is contained in:
sneakers-the-rat 2024-08-13 19:08:36 -07:00
parent 0a93195726
commit 9bd36340d7
Signed by untrusted user who does not match committer: jonny
GPG key ID: 6DCB96EF1E4D232D
2 changed files with 65 additions and 8 deletions

View file

@ -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:

View file

@ -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]:
"""