Merge pull request #89 from ArcanaFramework/touch-ups

Handle Self in extras signature matching

Author: tclose

extras/fileformats/extras/generic/__init__.py

@@ -0,0 +1,2 @@
+if __name__.startswith("fileformats."):
+    from . import converters  # noqa: F401

extras/fileformats/extras/generic/converters.py

@@ -0,0 +1,34 @@
+import tempfile
+from pathlib import Path
+import typing as ty
+import pydra.mark
+from fileformats.core import converter, FileSet
+from fileformats.generic import DirectoryOf, SetOf, TypedDirectory, TypedSet
+
+T = FileSet.type_var("T")
+
+
+@converter(target_format=SetOf[T], source_format=DirectoryOf[T])  # type: ignore[misc]
+@pydra.mark.task  # type: ignore[misc]
+@pydra.mark.annotate({"return": {"out_file": TypedSet}})  # type: ignore[misc]
+def list_dir_contents(in_file: TypedDirectory) -> TypedSet:
+    classified_set: ty.Type[TypedSet] = SetOf.__class_getitem__(*in_file.content_types)  # type: ignore[assignment, arg-type]
+    return classified_set(in_file.contents)
+
+
+@converter(target_format=DirectoryOf[T], source_format=SetOf[T])  # type: ignore[misc]
+@pydra.mark.annotate({"return": {"out_file": TypedDirectory}})  # type: ignore[misc]
+@pydra.mark.task  # type: ignore[misc]
+def put_contents_in_dir(
+    in_file: TypedSet,
+    out_dir: ty.Optional[Path] = None,
+    copy_mode: FileSet.CopyMode = FileSet.CopyMode.copy,
+) -> TypedDirectory:
+    if out_dir is None:
+        out_dir = Path(tempfile.mkdtemp())
+    for fset in in_file.contents:
+        fset.copy(out_dir, mode=copy_mode)
+    classified_dir: ty.Type[TypedDirectory] = DirectoryOf.__class_getitem__(  # type: ignore[assignment]
+        *in_file.content_types  # type: ignore[arg-type]
+    )
+    return classified_dir(out_dir)

extras/fileformats/extras/generic/tests/test_converters.py

@@ -0,0 +1,36 @@
+import pytest
+from pathlib import Path
+import typing as ty
+from fileformats.generic import DirectoryOf, SetOf
+from fileformats.text import TextFile
+
+
+@pytest.fixture
+def file_names() -> ty.List[str]:
+    return ["file1.txt", "file2.txt", "file3.txt"]
+
+
+@pytest.fixture
+def files_dir(file_names: ty.List[str], tmp_path: Path) -> Path:
+    out_dir = tmp_path / "out"
+    out_dir.mkdir()
+    for fname in file_names:
+        (out_dir / fname).write_text(fname)
+    return out_dir
+
+
+@pytest.fixture
+def test_files(files_dir: Path, file_names: ty.List[str]) -> ty.List[Path]:
+    return [files_dir / n for n in file_names]
+
+
+def test_list_dir_contents(files_dir: Path, test_files: ty.List[Path]) -> None:
+    text_set = SetOf[TextFile].convert(DirectoryOf[TextFile](files_dir))  # type: ignore[misc]
+    assert sorted(t.fspath for t in text_set.contents) == test_files
+
+
+def test_put_contents_in_dir(
+    file_names: ty.List[str], test_files: ty.List[Path]
+) -> None:
+    text_dir = DirectoryOf[TextFile].convert(SetOf[TextFile](test_files))  # type: ignore[misc]
+    assert sorted(t.name for t in text_dir.contents) == file_names

fileformats/application/__init__.py

@@ -515,7 +515,7 @@
     Zlib,
     Zstd,
 )
-from ..text import Javascript
+from fileformats.text import Javascript
 
 __all__ = [
     "__version__",

fileformats/application/serialization.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 from fileformats.core import DataType, FileSet, extra_implementation
 from fileformats.core.mixin import WithClassifier
-from ..generic import UnicodeFile
+from fileformats.generic import UnicodeFile
 from fileformats.core.exceptions import FormatMismatchError
 from fileformats.core import SampleFileGenerator
 

fileformats/core/collection.py

@@ -67,3 +67,6 @@ def unconstrained(cls) -> bool:
         """Whether the file-format is unconstrained by extension, magic number or another
         constraint"""
         return super().unconstrained and not cls.content_types
+
+    def __len__(self) -> int:
+        return len(self.contents)

fileformats/core/datatype.py

@@ -253,6 +253,10 @@ def get_format(mime_name: str) -> ty.Type[DataType]:
                         f"Did not find '{class_name}' class in fileformats.{namespace} "
                         f"corresponding to MIME, or MIME-like, type {mime_string}"
                     ) from None
+        if not issubclass(klass, cls):
+            raise FormatRecognitionError(
+                f"Class '{klass}' does not inherit from '{cls}'"
+            )
         return klass
 
     @classproperty

fileformats/core/extras.py

@@ -1,3 +1,4 @@
+import sys
 import importlib
 import typing as ty
 import inspect
@@ -11,6 +12,11 @@
 from .exceptions import FormatConversionError, FileFormatsExtrasError
 from .utils import import_extras_module, check_package_exists_on_pypi, add_exc_note
 
+if sys.version_info < (3, 11):
+    from typing_extensions import Self
+else:
+    from typing import Self
+
 if ty.TYPE_CHECKING:
     from pydra.engine.core import TaskBase
 
@@ -74,15 +80,28 @@ def decorator(implementation: ExtraImplementation) -> ExtraImplementation:
         fsig = inspect.signature(implementation)
         msig_args = list(msig.parameters.values())[1:]
         fsig_args = list(fsig.parameters.values())[1:]
+        dispatched_type = list(fsig.parameters.values())[0].annotation
         differences = []
 
-        def type_match(a: ty.Union[str, type], b: ty.Union[str, type]) -> bool:
+        def type_match(mtype: ty.Union[str, type], ftype: ty.Union[str, type]) -> bool:
+            """Check if the types match between the method and function annotations,
+            allowing for string annotations and `Self`
+
+            Parameters
+            ----------
+            mtype : Union[str, type]
+                the type of the method argument
+            ftype : Union[str, type]
+                the type of the function argument
+            """
             return (
-                a is ty.Any  # type: ignore[comparison-overlap]
-                or a == b
-                or inspect.isclass(a)
-                and inspect.isclass(b)
-                and issubclass(b, a)
+                mtype is ty.Any  # type: ignore[comparison-overlap]
+                or mtype == ftype
+                or inspect.isclass(ftype)
+                and (
+                    (inspect.isclass(mtype) and issubclass(ftype, mtype))
+                    or (mtype is Self and issubclass(ftype, dispatched_type))  # type: ignore[comparison-overlap, arg-type]
+                )
             )
 
         mhas_kwargs = msig_args and msig_args[-1].kind == inspect.Parameter.VAR_KEYWORD

fileformats/core/fileset.py

@@ -1610,7 +1610,8 @@ def move(
                     )
             shutil.move(str(fspath), new_path)
             new_paths.append(new_path)
-        return type(self)(new_paths)
+        self.fspaths = frozenset(new_paths)
+        return self
 
     def _fspaths_to_copy(
         self,