Merge pull request #67 from DSD-DBS/library-support

#66: Support for models which use libraries

capellambse/__init__.py

@@ -60,6 +60,7 @@ def migrate_cache_dir():
     check_plugin_version,
     yield_key_and_version_from_namespaces_by_plugin,
 )
+from .loader.filehandler import get_filehandler
 from .model import MelodyModel
 
 

capellambse/loader/core.py

@@ -84,6 +84,23 @@ def _find_refs(root: etree._Element) -> cabc.Iterable[str]:
     )
 
 
+def _unquote_ref(ref: str) -> str:
+    ref = urllib.parse.unquote(ref)
+    prefix = "platform:/resource/"
+    if ref.startswith(prefix):
+        ref = ref.replace(prefix, "../")
+    return ref
+
+
+class MissingResourceLocationError(KeyError):
+    """Raised when a model needs an additional resource location."""
+
+
+class ResourceLocationManager(dict):
+    def __missing__(self, key: str) -> t.NoReturn:
+        raise MissingResourceLocationError(key)
+
+
 class ModelFile:
     """Represents a single file in the model (i.e. a fragment)."""
 
@@ -204,18 +221,51 @@ class MelodyLoader:
 
     def __init__(
         self,
-        path: str | os.PathLike,
+        path: str | os.PathLike | filehandler.FileHandler,
         entrypoint: str | pathlib.PurePosixPath | None = None,
+        *,
+        resources: cabc.Mapping[str, filehandler.FileHandler] | None = None,
         **kwargs: t.Any,
     ) -> None:
-        self.filehandler = filehandler.get_filehandler(path, **kwargs)
+        """Construct a MelodyLoader.
+
+        Parameters
+        ----------
+        path
+            The ``path`` argument to the primary file handler, or the
+            primary file handler itself.
+        entrypoint
+            The entry point into the model, i.e. the top-level ``.aird``
+            file. This must be located within the primary file handler.
+        resources
+            Additional file handler instances that provide library
+            resources that are referenced from the model.
+        kwargs
+            Additional arguments to the primary file handler, if
+            necessary.
+        """
+        if isinstance(path, filehandler.FileHandler):
+            handler = path
+        else:
+            handler = filehandler.get_filehandler(path, **kwargs)
+        self.resources = ResourceLocationManager({"\0": handler})
+        for resource_name, resource_handler in (resources or {}).items():
+            if not resource_name:
+                raise ValueError("Empty resource name")
+            if "/" in resource_name or "\0" in resource_name:
+                raise ValueError(f"Invalid resource name: {resource_name!r}")
+            self.resources[resource_name] = resource_handler
         self.entrypoint = self.__derive_entrypoint(entrypoint)
 
         self.trees: dict[pathlib.PurePosixPath, ModelFile] = {}
         self.__load_referenced_files(
-            helpers.normalize_pure_path(pathlib.PurePosixPath(self.entrypoint))
+            pathlib.PurePosixPath("\0", self.entrypoint)
         )
 
+    @property
+    def filehandler(self) -> filehandler.FileHandler:
+        return self.resources["\0"]
+
     def __derive_entrypoint(
         self, entrypoint: str | pathlib.PurePosixPath | None
     ) -> pathlib.PurePosixPath:
@@ -226,22 +276,23 @@ def __derive_entrypoint(
             basedir = self.filehandler.path
             assert isinstance(basedir, pathlib.Path)
             self.filehandler.path = basedir.parent
-            return pathlib.PurePosixPath(basedir.name)
+            return helpers.normalize_pure_path(basedir.name)
 
         raise ValueError("This type of file handler needs an ``entrypoint``")
 
     def __load_referenced_files(
-        self,
-        filename: pathlib.PurePosixPath,
+        self, resource_path: pathlib.PurePosixPath
     ) -> None:
-        if filename in self.trees:
+        if resource_path in self.trees:
             return
 
-        frag = ModelFile(filename, self.filehandler)
-        self.trees[filename] = frag
+        handler = self.resources[resource_path.parts[0]]
+        filename = pathlib.PurePosixPath(*resource_path.parts[1:])
+        frag = ModelFile(filename, handler)
+        self.trees[resource_path] = frag
         for ref in _find_refs(frag.root):
             ref_name = helpers.normalize_pure_path(
-                urllib.parse.unquote(ref), base=filename.parent
+                _unquote_ref(ref), base=resource_path.parent
             )
             self.__load_referenced_files(ref_name)
 
@@ -269,6 +320,11 @@ def save(self, **kw: t.Any) -> None:
                     ", ".join(repr(k) for k in unsupported_kws),
                 )
             for fname, tree in self.trees.items():
+                resname = fname.parts[0]
+                fname = pathlib.PurePosixPath(*fname.parts[1:])
+                if resname != "\0":
+                    continue
+
                 tree.write_xml(fname)
 
     def idcache_index(self, subtree: etree._Element) -> None:

capellambse/loader/filehandler/__init__.py

@@ -76,9 +76,16 @@ def get_filehandler(path: str | os.PathLike, **kwargs: t.Any) -> FileHandler:
 class FileHandler(metaclass=abc.ABCMeta):
     path: str | os.PathLike
 
-    def __init__(self, path: str | os.PathLike, **kw: t.Any) -> None:
+    def __init__(
+        self,
+        path: str | os.PathLike,
+        *,
+        subdir: str | pathlib.PurePosixPath = "/",
+        **kw: t.Any,
+    ) -> None:
         super().__init__(**kw)  # type: ignore[call-arg]
         self.path = path
+        self.subdir = subdir
 
     @abc.abstractmethod
     def get_model_info(self) -> ModelInfo:

capellambse/loader/filehandler/gitfilehandler.py

@@ -433,8 +433,10 @@ def __init__(
         known_hosts_file: str = "",
         disable_cache: bool = False,
         update_cache: bool = True,
+        *,
+        subdir: str | pathlib.PurePosixPath = "/",
     ) -> None:
-        super().__init__(path)
+        super().__init__(path, subdir=subdir)
         self.revision = revision
         self.disable_cache = disable_cache
         self.username = username
@@ -461,7 +463,9 @@ def open(
         filename: str | pathlib.PurePosixPath,
         mode: t.Literal["r", "rb", "w", "wb"] = "rb",
     ) -> t.BinaryIO:
-        path = capellambse.helpers.normalize_pure_path(filename)
+        path = capellambse.helpers.normalize_pure_path(
+            filename, base=self.subdir
+        )
         if "w" in mode:
             if self._transaction is None:
                 raise TransactionClosedError(

capellambse/loader/filehandler/http.py

@@ -94,6 +94,8 @@ def __init__(
         path: str | os.PathLike,
         username: str | None = None,
         password: str | None = None,
+        *,
+        subdir: str | pathlib.PurePosixPath = "/",
     ) -> None:
         if not isinstance(path, str):
             raise TypeError(
@@ -103,6 +105,8 @@ def __init__(
             raise ValueError(
                 "Either both username and password must be given, or neither"
             )
+        if subdir != "/":
+            raise ValueError("`subdir=` is not supported in HTTP(S)")
         super().__init__(path)
 
         self.session = requests.Session()

capellambse/loader/filehandler/localfilehandler.py

@@ -20,7 +20,7 @@
 import subprocess
 import typing as t
 
-import capellambse.helpers
+from capellambse import helpers
 from capellambse.loader.modelinfo import ModelInfo
 
 from . import FileHandler
@@ -29,8 +29,13 @@
 
 
 class LocalFileHandler(FileHandler):
-    def __init__(self, path: str | os.PathLike) -> None:
-        path = pathlib.Path(path)
+    def __init__(
+        self,
+        path: str | os.PathLike,
+        *,
+        subdir: str | pathlib.PurePosixPath = "/",
+    ) -> None:
+        path = pathlib.Path(path, helpers.normalize_pure_path(subdir))
 
         super().__init__(path)
         self.__transaction: set[pathlib.PurePosixPath] | None = None
@@ -42,7 +47,7 @@ def open(
         mode: t.Literal["r", "rb", "w", "wb"] = "rb",
     ) -> t.BinaryIO:
         assert isinstance(self.path, pathlib.Path)
-        normpath = capellambse.helpers.normalize_pure_path(filename)
+        normpath = helpers.normalize_pure_path(filename)
         if "w" not in mode or self.__transaction is None:
             path = self.path / normpath
             return t.cast(t.BinaryIO, path.open(mode))