added module bytes_repr serializer

Author: tclose

pydra/design/base.py

@@ -23,6 +23,7 @@
     MultiOutputObj,
     MultiOutputFile,
 )
+from pydra.utils.hash import hash_function
 
 
 if ty.TYPE_CHECKING:
@@ -179,6 +180,8 @@ class Field:
         The converter for the field passed through to the attrs.field, by default it is None
     validator: callable | iterable[callable], optional
         The validator(s) for the field passed through to the attrs.field, by default it is None
+    hash_eq: bool, optional
+        Whether to use the hash of the value for equality comparison, by default it is False
     """
 
     name: str | None = None
@@ -192,8 +195,9 @@ class Field:
     requires: list[RequirementSet] = attrs.field(
         factory=list, converter=requires_converter
     )
-    converter: ty.Callable | None = None
-    validator: ty.Callable | None = None
+    converter: ty.Callable[..., ty.Any] | None = None
+    validator: ty.Callable[..., bool] | None = None
+    hash_eq: bool = False
 
     def requirements_satisfied(self, inputs: "TaskDef") -> bool:
         """Check if all the requirements are satisfied by the inputs"""
@@ -408,6 +412,7 @@ def make_task_def(
     klass : type
         The class created using the attrs package
     """
+
     spec_type._check_arg_refs(inputs, outputs)
 
     for inpt in inputs.values():
@@ -448,15 +453,15 @@ def make_task_def(
     # Now that we have saved the attributes in lists to be
     for arg in inputs.values():
         # If an outarg input then the field type should be Path not a FileSet
-        default_kwargs = _get_default(arg)
+        attrs_kwargs = _get_attrs_kwargs(arg)
         if isinstance(arg, Out) and is_fileset_or_union(arg.type):
             if getattr(arg, "path_template", False):
                 if is_optional(arg.type):
                     field_type = Path | bool | None
                     # Will default to None and not be inserted into the command
                 else:
                     field_type = Path | bool
-                    default_kwargs = {"default": True}
+                    attrs_kwargs = {"default": True}
             elif is_optional(arg.type):
                 field_type = Path | None
             else:
@@ -471,14 +476,14 @@ def make_task_def(
                 validator=make_validator(arg, klass.__name__),
                 metadata={PYDRA_ATTR_METADATA: arg},
                 on_setattr=attrs.setters.convert,
-                **default_kwargs,
+                **attrs_kwargs,
             ),
         )
         klass.__annotations__[arg.name] = field_type
 
     # Create class using attrs package, will create attributes for all columns and
     # parameters
-    attrs_klass = attrs.define(auto_attribs=False, kw_only=True)(klass)
+    attrs_klass = attrs.define(auto_attribs=False, kw_only=True, eq=False)(klass)
 
     return attrs_klass
 
@@ -541,13 +546,15 @@ def make_outputs_spec(
             n: attrs.field(
                 converter=make_converter(o, f"{spec_name}.Outputs"),
                 metadata={PYDRA_ATTR_METADATA: o},
-                **_get_default(o),
+                **_get_attrs_kwargs(o),
             )
             for n, o in outputs.items()
         },
     )
     outputs_klass.__annotations__.update((o.name, o.type) for o in outputs.values())
-    outputs_klass = attrs.define(auto_attribs=False, kw_only=True)(outputs_klass)
+    outputs_klass = attrs.define(auto_attribs=False, kw_only=True, eq=False)(
+        outputs_klass
+    )
     return outputs_klass
 
 
@@ -972,14 +979,19 @@ def check_explicit_fields_are_none(klass, inputs, outputs):
         )
 
 
-def _get_default(field: Field) -> dict[str, ty.Any]:
+def _get_attrs_kwargs(field: Field) -> dict[str, ty.Any]:
+    kwargs = {}
     if not hasattr(field, "default"):
-        return {"factory": nothing_factory}
-    if field.default is not EMPTY:
-        return {"default": field.default}
-    if is_optional(field.type):
-        return {"default": None}
-    return {"factory": nothing_factory}
+        kwargs["factory"] = nothing_factory
+    elif field.default is not EMPTY:
+        kwargs["default"] = field.default
+    elif is_optional(field.type):
+        kwargs["default"] = None
+    else:
+        kwargs["factory"] = nothing_factory
+    if field.hash_eq:
+        kwargs["eq"] = hash_function
+    return kwargs
 
 
 def nothing_factory():

pydra/design/python.py

@@ -158,7 +158,7 @@ def make(wrapped: ty.Callable | type) -> PythonDef:
             )
 
         parsed_inputs["function"] = arg(
-            name="function", type=ty.Callable, default=function
+            name="function", type=ty.Callable, default=function, hash_eq=True
         )
 
         defn = make_task_def(

pydra/engine/specs.py

@@ -54,7 +54,7 @@ def is_set(value: ty.Any) -> bool:
     return value not in (attrs.NOTHING, EMPTY)
 
 
-@attrs.define
+@attrs.define(kw_only=True, auto_attribs=False, eq=False)
 class TaskOutputs:
     """Base class for all output definitions"""
 
@@ -113,11 +113,31 @@ def __getitem__(self, name_or_index: str | int) -> ty.Any:
                 f"{self} doesn't have an attribute {name_or_index}"
             ) from None
 
+    def __eq__(self, other: ty.Any) -> bool:
+        """Check if two task definitions are equal"""
+        values = attrs.asdict(self)
+        fields = list_fields(self)
+        try:
+            other_values = attrs.asdict(other)
+        except AttributeError:
+            return False
+        try:
+            other_fields = list_fields(other)
+        except AttributeError:
+            return False
+        if fields != other_fields:
+            return False
+        for field in list_fields(self):
+            if field.hash_eq:
+                values[field.name] = hash_function(values[field.name])
+                other_values[field.name] = hash_function(other_values[field.name])
+        return values == other_values
+
 
 OutputsType = ty.TypeVar("OutputType", bound=TaskOutputs)
 
 
-@attrs.define(kw_only=True, auto_attribs=False)
+@attrs.define(kw_only=True, auto_attribs=False, eq=False)
 class TaskDef(ty.Generic[OutputsType]):
     """Base class for all task definitions"""
 
@@ -341,6 +361,34 @@ def __iter__(self) -> ty.Generator[str, None, None]:
             if not (f.name.startswith("_") or f.name in self.RESERVED_FIELD_NAMES)
         )
 
+    def __eq__(self, other: ty.Any) -> bool:
+        """Check if two task definitions are equal"""
+        values = attrs.asdict(self)
+        try:
+            other_values = attrs.asdict(other)
+        except AttributeError:
+            return False
+        if set(values) != set(other_values):
+            return False  # Return if attribute keys don't match
+        for field in list_fields(self):
+            if field.hash_eq:
+                values[field.name] = hash_function(values[field.name])
+                other_values[field.name] = hash_function(other_values[field.name])
+        if values != other_values:
+            return False
+        hash_cache = Cache()
+        if hash_function(type(self), cache=hash_cache) != hash_function(
+            type(other), cache=hash_cache
+        ):
+            return False
+        try:
+            other_outputs = other.Outputs
+        except AttributeError:
+            return False
+        return hash_function(self.Outputs, cache=hash_cache) == hash_function(
+            other_outputs, cache=hash_cache
+        )
+
     def __getitem__(self, name: str) -> ty.Any:
         """Return the value for the given attribute, resolving any templates
 
@@ -595,7 +643,7 @@ class RuntimeSpec:
     network: bool = False
 
 
-@attrs.define(kw_only=True, auto_attribs=False)
+@attrs.define(kw_only=True, auto_attribs=False, eq=False)
 class PythonOutputs(TaskOutputs):
 
     @classmethod
@@ -624,7 +672,7 @@ def _from_task(cls, task: "Task[PythonDef]") -> Self:
 PythonOutputsType = ty.TypeVar("OutputType", bound=PythonOutputs)
 
 
-@attrs.define(kw_only=True, auto_attribs=False)
+@attrs.define(kw_only=True, auto_attribs=False, eq=False)
 class PythonDef(TaskDef[PythonOutputsType]):
 
     _task_type = "python"
@@ -653,7 +701,7 @@ def _run(self, task: "Task[PythonDef]") -> None:
             )
 
 
-@attrs.define(kw_only=True, auto_attribs=False)
+@attrs.define(kw_only=True, auto_attribs=False, eq=False)
 class WorkflowOutputs(TaskOutputs):
 
     @classmethod
@@ -707,7 +755,7 @@ def _from_task(cls, task: "Task[WorkflowDef]") -> Self:
 WorkflowOutputsType = ty.TypeVar("OutputType", bound=WorkflowOutputs)
 
 
-@attrs.define(kw_only=True, auto_attribs=False)
+@attrs.define(kw_only=True, auto_attribs=False, eq=False)
 class WorkflowDef(TaskDef[WorkflowOutputsType]):
 
     _task_type = "workflow"
@@ -738,7 +786,7 @@ def construct(self) -> "Workflow":
 STDERR_HELP = """The standard error stream produced by the command."""
 
 
-@attrs.define(kw_only=True, auto_attribs=False)
+@attrs.define(kw_only=True, auto_attribs=False, eq=False)
 class ShellOutputs(TaskOutputs):
     """Output definition of a generic shell process."""
 
@@ -899,7 +947,7 @@ def _resolve_value(
 ShellOutputsType = ty.TypeVar("OutputType", bound=ShellOutputs)
 
 
-@attrs.define(kw_only=True, auto_attribs=False)
+@attrs.define(kw_only=True, auto_attribs=False, eq=False)
 class ShellDef(TaskDef[ShellOutputsType]):
 
     _task_type = "shell"

pydra/engine/submitter.py

@@ -87,7 +87,7 @@ class Submitter:
     def __init__(
         self,
         cache_dir: os.PathLike | None = None,
-        worker: str | ty.Type[Worker] | Worker = "debug",
+        worker: str | ty.Type[Worker] | Worker | None = "debug",
         environment: "Environment | None" = None,
         rerun: bool = False,
         cache_locations: list[os.PathLike] | None = None,
@@ -98,6 +98,9 @@ def __init__(
         **kwargs,
     ):
 
+        if worker is None:
+            worker = "debug"
+
         from . import check_latest_version
 
         if Task._etelemetry_version_data is None:

pydra/utils/hash.py

@@ -24,6 +24,7 @@
 from filelock import SoftFileLock
 import attrs.exceptions
 from fileformats.core.fileset import FileSet, MockMixin
+from fileformats.generic import FsObject
 import fileformats.core.exceptions
 from . import user_cache_dir, add_exc_note
 from .misc import in_stdlib
@@ -332,17 +333,16 @@ def bytes_repr(obj: object, cache: Cache) -> Iterator[bytes]:
     elif hasattr(obj, "__slots__") and obj.__slots__ is not None:
         dct = {attr: getattr(obj, attr) for attr in obj.__slots__}
     else:
+
+        def is_special_or_method(n: str):
+            return (n.startswith("__") and n.endswith("__")) or inspect.ismethod(
+                getattr(obj, n)
+            )
+
         try:
-            dct = obj.__dict__
+            dct = {n: v for n, v in obj.__dict__.items() if not is_special_or_method(n)}
         except AttributeError:
-            dct = {
-                n: getattr(obj, n)
-                for n in dir(obj)
-                if not (
-                    (n.startswith("__") and n.endswith("__"))
-                    or inspect.ismethod(getattr(obj, n))
-                )
-            }
+            dct = {n: getattr(obj, n) for n in dir(obj) if not is_special_or_method(n)}
     yield from bytes_repr_mapping_contents(dct, cache)
     yield b"}"
 
@@ -456,6 +456,13 @@ def bytes_repr_dict(obj: dict, cache: Cache) -> Iterator[bytes]:
     yield b"}"
 
 
+@register_serializer
+def bytes_repr_module(obj: types.ModuleType, cache: Cache) -> Iterator[bytes]:
+    yield b"module:("
+    yield hash_single(FsObject(obj.__file__), cache=cache)
+    yield b")"
+
+
 @register_serializer(ty._GenericAlias)
 @register_serializer(ty._SpecialForm)
 @register_serializer(type)