diff --git a/docs/refactor/consolidation-20261009/L6-codegen.md b/docs/refactor/consolidation-20261009/L6-codegen.md new file mode 100644 index 0000000000..e1eed906f7 --- /dev/null +++ b/docs/refactor/consolidation-20261009/L6-codegen.md @@ -0,0 +1,77 @@ +# L6: pycodify and ObjectState own Python source generation + +**Head audited:** `openhcs` `main` at `7f41c419c` (#1197); `pycodify` `main` at `108a8ed` (0.1.3 plus the unreleased nominal-subtype fix); `ObjectState` `main` at `19aafdc` (2.0.0). **Rules:** [00-RULES.md](00-RULES.md). **Step 4.** +**Shared abstractions** ([02-SHARED-ABSTRACTIONS.md](02-SHARED-ABSTRACTIONS.md)). *Builds:* none. *Uses:* none. + +Pipelines are saved as `.py` files generated by pycodify (pickle `.pipeline` files are deprecated), so code generation is the persistence format: a formatter bug is a data-loss bug. + +## What is wrong + +**pycodify writes values it cannot read back, and OpenHCS fixes this with private copies of generic mechanisms instead of fixing the library.** + +pycodify `formatters.py` (verified with a script at this head): +- `EnumFormatter` (`:77-87`) imports `cls.__name__`: a member of `Outer.Mode` renders `from m import Mode` / `Mode.B`, an `ImportError` on load. +- `_format_dataclass_expression` (`:285-287`) and `CallableFormatter` (`:129-130`) make the same `__name__` mistake for nested dataclasses and callables. +- Classes have no formatter: `CallableFormatter` excludes types, so a class value falls to `FallbackFormatter` and renders as ``, a `SyntaxError`. `FallbackFormatter` (`:352-359`) emits `repr` for any value, so every unknown type is silent garbage. +- Non-clean mode replaces a `None` field whose annotation is a dataclass with a placeholder instance of all-`None` fields (`:310-314`, `_create_placeholder_dataclass_instance`): `Optional[Inner] = None` saves as `Inner(x=None)`, a different value. +- `init=False` fields are emitted as constructor keywords (`:300`): a `TypeError` on load. +- ObjectState's lazy semantics are duck-typed into the generic formatter: `hasattr(instance, "_resolve_field_value")`, `hasattr(cls, "_is_lazy_dataclass")` (`:251, 258, 262, 303`). + +OpenHCS copies (`openhcs/serialization/`, 747 lines; `openhcs/core/python_source_literal.py`, 25): +- `EnumMemberFormatter` is the nested-enum fix, written in OpenHCS (`pycodify_formatters.py:131-149`); `AxisDeclarationFormatter` (`:69-82`) is the same qualified-name rule for axis classes; `OpenHCSImportableTypeFormatter` (`:54-66`) supplies the missing class formatter through a `FunctionReference`; `NameMappingLookup` (`:25-34`) restates `Mapping.get`. The qualified-name rule is written three times. +- `PythonSourceLiteral` (`core/python_source_literal.py`) and its formatter (`pycodify_formatters.py:85-97`): a generic "value owns its source expression" hook, with three methods (`source_literal`, `source_literal_imports`, `source_literal_with_names`) for one fact. +- `source_path_factoring.py` (195 lines) and `OpenHCSPathFormatter` (`:100-128`): document-level path factoring, generic, plugged in through pycodify's type-keyed `FormatContext.extensions` bag, whose only user it is. `OpenHCSPythonSourceDocument` restates `generate_python_source`; six production callers construct it. +- The LazyDataclass formatter (`pycodify_formatters.py:313-481`, 170 lines: `LazyDataclassFieldEmissionState`, `LazyDataclassSerializedField`, `LazyDataclassSerializationPlan`, `LazyDataclassFormatEligibility`, `LazyDataclassFormatter`, and `FunctionStepCleanModeFieldPolicy`): ObjectState's raw-value and explicitly-set-field semantics, read from `__dict__["_explicitly_set_fields"]` outside ObjectState. +- `OpenHCSCallableFormatter.format` (`:46-51`) catches `RuntimeError` and emits `repr(function)`: an unreferenceable callable saves as ``. + +ObjectState: field-name-gated debug logging names OpenHCS's old `well_filter` field (and `num_workers`, `sub_dir`, `output_dir_suffix`) in `dual_axis_resolver.py`, `object_state.py`, `object_state_registry.py`, `context_manager.py` (G7 leftover); docstrings use `WellFilterConfig` examples. + +## Target + +pycodify 0.2.0: +- `imported_name(module, qualname, context) -> SourceFragment`: the one qualified-name rule (import the root of `__qualname__`, spell the rest as attributes, honour name mappings, no import for builtins, `TypeError` for ``). Enum members, classes (new `TypeFormatter`), callables and dataclass constructors all use it. +- `PythonSourceLiteral(ABC)` with one method, `source_fragment(context) -> SourceFragment`, and its formatter. +- `paths.py`: a `PathRendering` family on `FormatContext.paths` (plain, collecting, factored). `generate_python_source` always factors absolute paths (discovery pass, plan, render). `FormatContext.extensions`/`extension()` are deleted. +- Deleted: `FallbackFormatter` (unformattable values raise `TypeError`), the placeholder-instance substitution, the lazy `hasattr` probes; `init=False` fields are skipped. `frozenset`, `bytes` and `complex` get exact formatters. + +ObjectState 2.1.0 (adds a module and a `pycodify` dependency; removes only private debug branches): +- `objectstate.codegen`: `LazyResolutionFormatter` for every `LazyResolutionDataclass` (raw values; clean mode writes fields whose raw value differs from the field default), refined by `LazyDataclassFormatter` (constructor-set fields), and `field_needs_source(current, default, context)` for owners such as `FunctionStep` that format lazy-config fields. +- The field-name-gated debug branches are deleted; docstring examples renamed to `partition_filter`. + +OpenHCS: +- Deleted: `openhcs/core/python_source_literal.py`, `openhcs/serialization/source_path_factoring.py`, `NameMappingLookup`, `EnumMemberFormatter`, `AxisDeclarationFormatter`, `OpenHCSImportableTypeFormatter`, `PythonSourceLiteralFormatter`, `OpenHCSPathFormatter`, the six LazyDataclass classes and `FunctionStepCleanModeFieldPolicy`, the `repr` catch in `OpenHCSCallableFormatter`. +- `FunctionReference` subclasses pycodify's `PythonSourceLiteral`; the six document callers call `generate_python_source`. + +**Found by the round-trip test.** Base configs that inherit through `None` (`StepPartitionFilterConfig`, `PathPlanningConfig`, the streaming configs and 5 others) are `LazyResolutionDataclass` but not `LazyDataclass`; pycodify's generic dataclass formatter wrote their *resolved* values in full mode, so a saved `GlobalPipelineConfig` stopped inheriting (for example `step_partition_filter_config.partition_filter_mode` was baked to `IncludePartitions`). Fixed by `LazyResolutionFormatter`. + +## Persisted state + +Saved `.py` pipelines and configs are durable. Output that loaded before still loads: the qualified-name, class and path spellings are unchanged for every value that round-tripped; values that rendered unloadable source now render loadable source or raise at save time. No cutover tool. + +## Guards + +`tests/unit/test_l6_codegen_guards.py`: +- `openhcs/core/python_source_literal.py` and `openhcs/serialization/source_path_factoring.py` do not exist. +- No `SourceFormatter` subclass under `openhcs/` handles `Enum`, `Path`, `type`, `PythonSourceLiteral` or `LazyDataclass` (AST: no `isinstance(value, Enum|Path|LazyDataclass|PythonSourceLiteral)` and no `inspect.isclass` in a `can_format`). +- No `__qualname__.partition` and no `_explicitly_set_fields` under `openhcs/`. +- No string `well_filter` in `external/ObjectState/src`. + +## Tests + +- OpenHCS: one family-level round-trip test (`tests/unit/test_source_round_trip.py`): every registered config dataclass (`LazyDataclassFactory.registered_type_pairs()`, both lazy and base), each with every field set to a non-default value derived from its annotation (nested enums, partition-filter mode family members, axis classes, paths), plus every function-pattern shape the step editor produces (callable, `(callable, kwargs)`, chain, dict by group-by value, dict of chains) inside `FunctionStep`s inside a pipeline document, in clean and full mode, re-imported and compared for equality. The narrow formatter tests it covers are deleted; tests of deleted modules move to the libraries. +- pycodify: the generic tests (qualified names, classes, path factoring, `PythonSourceLiteral`, unformattable values raise). +- ObjectState: lazy formatter tests (clean omission, explicit empty configs, full mode keeps `None`, round trip). + +## New-case experiments + +- A new nested enum or axis class in a config: today it needs an OpenHCS formatter at a priority above pycodify's broken one. After: nothing. +- A new value type that owns its source spelling: today it subclasses an OpenHCS ABC with three methods. After: one `source_fragment` method on pycodify's ABC. +- A second application persisting ObjectState configs as source: today it must copy 170 lines out of OpenHCS. After: `import objectstate.codegen`. + +## Release order + +pycodify 0.2.0, then ObjectState 2.1.0 (depends on `pycodify>=0.2.0,<0.3`), then the OpenHCS repin (`pycodify>=0.2.0,<0.3`, `objectstate>=2.1,<3`). + +## Done when + +The OpenHCS copies are gone, the guards and the round-trip test pass, the pycodify and ObjectState PRs are open with their tests green, and the OpenHCS submodules point at those branch heads. diff --git a/external/ObjectState b/external/ObjectState index 4d74eacf98..4eb8de1398 160000 --- a/external/ObjectState +++ b/external/ObjectState @@ -1 +1 @@ -Subproject commit 4d74eacf98fddc94f4d4cc3aaeef4b998a2ad03b +Subproject commit 4eb8de1398428c539e1e14360c847ccea2fe4b34 diff --git a/external/pycodify b/external/pycodify index 108a8edbf2..e3e3d993e9 160000 --- a/external/pycodify +++ b/external/pycodify @@ -1 +1 @@ -Subproject commit 108a8edbf25de258168756bac22c7572f018fa90 +Subproject commit e3e3d993e9727cacd8aaf6fd7f720b764f40e5e8 diff --git a/openhcs/core/config_document.py b/openhcs/core/config_document.py index d8bb11b90b..9b174d3d8f 100644 --- a/openhcs/core/config_document.py +++ b/openhcs/core/config_document.py @@ -162,19 +162,16 @@ def render( ) -> str: """Render a validated config object as reviewable Python source.""" - from pycodify import Assignment + from pycodify import Assignment, generate_python_source import openhcs.serialization.pycodify_formatters # noqa: F401 - from openhcs.serialization.source_path_factoring import ( - OpenHCSPythonSourceDocument, - ) cls._require_config_type(config, expected_config_type) - return OpenHCSPythonSourceDocument( + return generate_python_source( Assignment(ConfigDocumentField.CONFIG.value, config), header=cls.HEADER, clean_mode=clean_mode, - ).render() + ) @staticmethod def _require_config_type( diff --git a/openhcs/core/function_reference.py b/openhcs/core/function_reference.py index 6fb96799c2..0702107406 100644 --- a/openhcs/core/function_reference.py +++ b/openhcs/core/function_reference.py @@ -2,17 +2,17 @@ from __future__ import annotations -import builtins import dataclasses import importlib import inspect import logging from abc import ABC, abstractmethod -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Sequence from dataclasses import dataclass from types import ModuleType from typing import TYPE_CHECKING, ClassVar +from pycodify import FormatContext, PythonSourceLiteral, SourceFragment, imported_name from python_introspect import Enableable from openhcs.core.callable_contract import ( @@ -21,7 +21,6 @@ FunctionStepExecutionScope, ) from openhcs.core.function_contract_metadata import FunctionContractAttribute -from openhcs.core.python_source_literal import PythonSourceLiteral if TYPE_CHECKING: from openhcs.core.steps.abstract import AbstractStep @@ -75,24 +74,10 @@ def source_import_identity(self) -> CallableImportIdentity: def source_expression(self, imported_name: str) -> str: """Render through the source authority, including pycodify name aliases.""" - def source_literal(self) -> str: - return self.source_expression(self.source_import_identity.function_name) - - def source_literal_with_names( - self, name_mappings: Mapping[tuple[str, str], str], - ) -> str: - identity = self.source_import_identity - imported_name = name_mappings.get( - (identity.module_name, identity.function_name), identity.function_name, - ) - return self.source_expression(imported_name) - - def source_literal_imports(self) -> frozenset[tuple[str, str]]: - imports = super().source_literal_imports() + def source_fragment(self, context: FormatContext) -> SourceFragment: identity = self.source_import_identity - if identity.module_name == builtins.__name__: - return imports - return imports | frozenset({(identity.module_name, identity.function_name)}) + name = imported_name(identity.module_name, identity.function_name, context) + return SourceFragment(self.source_expression(name.code), name.imports) @abstractmethod def resolve(self) -> Callable: diff --git a/openhcs/core/function_step_document.py b/openhcs/core/function_step_document.py index 7c8041f11c..cd4133928c 100644 --- a/openhcs/core/function_step_document.py +++ b/openhcs/core/function_step_document.py @@ -68,17 +68,14 @@ def render( """Render a validated FunctionStep document as reviewable Python source.""" import openhcs.serialization.pycodify_formatters # noqa: F401 - from pycodify import Assignment - from openhcs.serialization.source_path_factoring import ( - OpenHCSPythonSourceDocument, - ) + from pycodify import Assignment, generate_python_source normalized = cls.from_value(document.step) - return OpenHCSPythonSourceDocument( + return generate_python_source( Assignment(FunctionStepDocumentField.STEP.value, normalized.step), header=cls.HEADER, clean_mode=clean_mode, - ).render() + ) @staticmethod def _require_step(value: object) -> None: diff --git a/openhcs/core/function_step_transport.py b/openhcs/core/function_step_transport.py index 0e9ac8f433..4fe7cf2a38 100644 --- a/openhcs/core/function_step_transport.py +++ b/openhcs/core/function_step_transport.py @@ -36,19 +36,16 @@ def source_from_pipeline( ) -> str: """Normalize and pycodify one public FunctionStep list.""" import openhcs.serialization.pycodify_formatters # noqa: F401 - from pycodify import Assignment - from openhcs.serialization.source_path_factoring import ( - OpenHCSPythonSourceDocument, - ) + from pycodify import Assignment, generate_python_source steps = list(pipeline_steps) cls._require_function_steps(steps) normalized = cls.normalize_pipeline(steps) - return OpenHCSPythonSourceDocument( + return generate_python_source( Assignment(PipelineDocumentField.PIPELINE_STEPS.value, normalized), header="# Edit this pipeline and save to apply changes", clean_mode=clean_mode, - ).render() + ) @classmethod def pipeline_steps_from_namespace( diff --git a/openhcs/core/pipeline_document.py b/openhcs/core/pipeline_document.py index 47197d2178..bae002c41f 100644 --- a/openhcs/core/pipeline_document.py +++ b/openhcs/core/pipeline_document.py @@ -98,10 +98,7 @@ def render( """Render a validated pipeline document as reviewable Python source.""" import openhcs.serialization.pycodify_formatters # noqa: F401 - from pycodify import Assignment, BlankLine, CodeBlock - from openhcs.serialization.source_path_factoring import ( - OpenHCSPythonSourceDocument, - ) + from pycodify import Assignment, BlankLine, CodeBlock, generate_python_source normalized = cls.from_values( pipeline_config=document.pipeline_config, @@ -120,11 +117,11 @@ def render( ), ) ) - return OpenHCSPythonSourceDocument( + return generate_python_source( body, header=cls.HEADER, clean_mode=clean_mode, - ).render() + ) @staticmethod def _require_pipeline_config(value: object) -> None: diff --git a/openhcs/core/python_source_literal.py b/openhcs/core/python_source_literal.py deleted file mode 100644 index b9d4cc39f8..0000000000 --- a/openhcs/core/python_source_literal.py +++ /dev/null @@ -1,25 +0,0 @@ -"""Nominal source-literal hooks for objects with no structural constructor.""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from collections.abc import Mapping - - -class PythonSourceLiteral(ABC): - """Object that owns its canonical importable Python source expression.""" - - @abstractmethod - def source_literal(self) -> str: - """Return the Python expression that reconstructs this value.""" - - def source_literal_imports(self) -> frozenset[tuple[str, str]]: - """Return imports required by :meth:`source_literal`.""" - return frozenset() - - def source_literal_with_names( - self, name_mappings: Mapping[tuple[str, str], str], - ) -> str: - """Render context-free literals unchanged; import owners can bind aliases.""" - - return self.source_literal() diff --git a/openhcs/pyqt_gui/services/reactor_providers.py b/openhcs/pyqt_gui/services/reactor_providers.py index aa55b034dd..17922c59d0 100644 --- a/openhcs/pyqt_gui/services/reactor_providers.py +++ b/openhcs/pyqt_gui/services/reactor_providers.py @@ -69,17 +69,13 @@ def render_assignment( header: str, clean_mode: bool, ) -> str: - from pycodify import Assignment + from pycodify import Assignment, generate_python_source - from openhcs.serialization.source_path_factoring import ( - OpenHCSPythonSourceDocument, - ) - - return OpenHCSPythonSourceDocument( + return generate_python_source( Assignment(assignment_name, value), header=header, clean_mode=clean_mode, - ).render() + ) def normalize_source( self, diff --git a/openhcs/serialization/pycodify_formatters.py b/openhcs/serialization/pycodify_formatters.py index e131093065..4aa795791b 100644 --- a/openhcs/serialization/pycodify_formatters.py +++ b/openhcs/serialization/pycodify_formatters.py @@ -2,38 +2,29 @@ import inspect from collections.abc import Callable -from dataclasses import dataclass, fields -from enum import Enum from pathlib import Path from objectstate import semantic_values_equal -from objectstate.lazy_factory import LazyDataclass -from pycodify import FormatContext, SourceFormatter, SourceFragment, to_source +from objectstate.codegen import field_needs_source +from pycodify import ( + FormatContext, + SourceFormatter, + SourceFragment, + imported_name, + to_source, +) from pyqt_reactive.pattern_metadata import PatternScopeToken from python_introspect import callable_declaration_kwargs, parameter_exclusions from openhcs.core.axes import is_grouping_declaration -from openhcs.core.callable_contract import CallableContract, CallableImportIdentity +from openhcs.core.callable_contract import CallableContract from openhcs.core.function_reference import ( FunctionReference, FunctionReferenceTransportAuthority, - ImportableFunctionReference, ) from openhcs.core.steps.function_step import FunctionStep -class NameMappingLookup: - @staticmethod - def resolve( - context: FormatContext, - import_pair: tuple[str, str], - default_name: str, - ) -> str: - if import_pair in context.name_mappings: - return context.name_mappings[import_pair] - return default_name - - class OpenHCSCallableFormatter(SourceFormatter): priority = 75 @@ -44,109 +35,7 @@ def reference_for(self, value) -> FunctionReference: return FunctionReferenceTransportAuthority.function_reference(value) def format(self, value, context: FormatContext) -> SourceFragment: - try: - reference = self.reference_for(value) - except RuntimeError: - return SourceFragment(repr(value), frozenset()) - return to_source(reference, context) - - -class OpenHCSImportableTypeFormatter(OpenHCSCallableFormatter): - """Classes own direct declaration imports, not processing-registry lookup.""" - - priority = 76 - - def can_format(self, value) -> bool: - return inspect.isclass(value) - - def reference_for(self, value) -> FunctionReference: - identity = CallableImportIdentity.from_callable(value) - return ImportableFunctionReference( - import_identity=identity, composite_key=identity.import_path, - ) - - -class AxisDeclarationFormatter(SourceFormatter): - """Axes and grouping declarations are spelled by their qualified class path.""" - - priority = 77 - - def can_format(self, value) -> bool: - return is_grouping_declaration(value) - - def format(self, value, context: FormatContext) -> SourceFragment: - root_name, _, nested_path = value.__qualname__.partition(".") - import_pair = (value.__module__, root_name) - mapped_root = NameMappingLookup.resolve(context, import_pair, root_name) - reference = f"{mapped_root}.{nested_path}" if nested_path else mapped_root - return SourceFragment(reference, frozenset([import_pair])) - - -class PythonSourceLiteralFormatter(SourceFormatter): - priority = 120 - - def can_format(self, value) -> bool: - from openhcs.core.python_source_literal import PythonSourceLiteral - - return isinstance(value, PythonSourceLiteral) - - def format(self, value, context: FormatContext) -> SourceFragment: - return SourceFragment( - value.source_literal_with_names(context.name_mappings), - value.source_literal_imports(), - ) - - -class OpenHCSPathFormatter(SourceFormatter): - """Collect and substitute document-scoped path expressions.""" - - priority = 118 - - def can_format(self, value) -> bool: - return isinstance(value, Path) - - def format(self, value: Path, context: FormatContext) -> SourceFragment: - from openhcs.serialization.source_path_factoring import ( - SourcePathFactoringPlan, - SourcePathOccurrenceCollector, - ) - - collector = context.extension(SourcePathOccurrenceCollector) - if collector is not None: - collector.record(value) - plan = context.extension(SourcePathFactoringPlan) - if plan is not None: - expression = plan.expression_for(value) - if expression is not None: - return to_source(expression, context) - - import_pair = ("pathlib", "Path") - name = NameMappingLookup.resolve(context, import_pair, "Path") - return SourceFragment( - f"{name}({str(value)!r})", - frozenset((import_pair,)), - ) - - -class EnumMemberFormatter(SourceFormatter): - priority = 115 - - def can_format(self, value) -> bool: - return isinstance(value, Enum) - - def format(self, value, context: FormatContext) -> SourceFragment: - enum_type = type(value) - if "" in enum_type.__qualname__: - return SourceFragment(repr(value), frozenset()) - - root_name, _, nested_path = enum_type.__qualname__.partition(".") - import_pair = (enum_type.__module__, root_name) - mapped_root = NameMappingLookup.resolve(context, import_pair, root_name) - enum_reference = f"{mapped_root}.{nested_path}" if nested_path else mapped_root - return SourceFragment( - f"{enum_reference}.{value.name}", - frozenset([import_pair]), - ) + return to_source(self.reference_for(value), context) class MaterializationSpecFormatter(SourceFormatter): @@ -158,17 +47,12 @@ def can_format(self, value) -> bool: return isinstance(value, MaterializationSpec) def format(self, value, context: FormatContext) -> SourceFragment: - materialization_type = type(value) - import_pair = ( - materialization_type.__module__, - materialization_type.__name__, - ) - class_name = NameMappingLookup.resolve( - context, import_pair, materialization_type.__name__ + class_ref = imported_name( + type(value).__module__, type(value).__qualname__, context ) item_ctx = context.indented() output_frags = [to_source(output, item_ctx) for output in value.outputs] - imports = {import_pair} + imports = set(class_ref.imports) for frag in output_frags: imports |= set(frag.imports) @@ -188,7 +72,7 @@ def format(self, value, context: FormatContext) -> SourceFragment: inner = f",\n{item_ctx.indent_str}".join(args) return SourceFragment( - f"{class_name}(\n{item_ctx.indent_str}{inner}\n{context.indent_str})", + f"{class_ref.code}(\n{item_ctx.indent_str}{inner}\n{context.indent_str})", frozenset(imports), ) @@ -310,177 +194,6 @@ def format(self, value: list, context: FormatContext) -> SourceFragment: return SourceFragment(code, imports) -@dataclass(frozen=True) -class LazyDataclassFieldEmissionState: - explicit_field_names: frozenset[str] - has_concrete_field_values: bool - - @classmethod - def from_instance(cls, value: LazyDataclass) -> "LazyDataclassFieldEmissionState": - raw_values = value.__dict__ - explicit_field_names = frozenset(raw_values["_explicitly_set_fields"]) - has_concrete_field_values = any( - raw_values[field.name] is not None for field in fields(value) - ) - return cls( - explicit_field_names=explicit_field_names, - has_concrete_field_values=has_concrete_field_values, - ) - - @property - def requires_serialization(self) -> bool: - return bool(self.explicit_field_names) or self.has_concrete_field_values - - @staticmethod - def value_requires_clean_serialization(value) -> bool: - if value is None: - return False - if isinstance(value, LazyDataclass): - return LazyDataclassFieldEmissionState.from_instance( - value - ).requires_serialization - return True - - @staticmethod - def raw_field_values_match(left: LazyDataclass, right: LazyDataclass) -> bool: - if type(left) is not type(right): - return False - left_values = left.__dict__ - right_values = right.__dict__ - for field in fields(left): - left_value = left_values[field.name] - right_value = right_values[field.name] - if isinstance(left_value, LazyDataclass) and isinstance( - right_value, - LazyDataclass, - ): - if not LazyDataclassFieldEmissionState.raw_field_values_match( - left_value, - right_value, - ): - return False - elif left_value != right_value: - return False - return True - - -@dataclass(frozen=True) -class LazyDataclassSerializedField: - name: str - fragment: SourceFragment - - -@dataclass(frozen=True) -class LazyDataclassSerializationPlan: - class_name: str - import_pair: tuple[str, str] - fields: tuple[LazyDataclassSerializedField, ...] - - @classmethod - def from_instance( - cls, - value: LazyDataclass, - context: FormatContext, - ) -> "LazyDataclassSerializationPlan": - lazy_class = type(value) - import_pair = (lazy_class.__module__, lazy_class.__name__) - if import_pair in context.name_mappings: - class_name = context.name_mappings[import_pair] - else: - class_name = lazy_class.__name__ - - raw_values = value.__dict__ - explicit_field_names = raw_values["_explicitly_set_fields"] - field_ctx = context.indented() - serialized_fields = [] - - for field in fields(value): - current_value = raw_values[field.name] - if context.clean_mode and current_value is None: - continue - if context.clean_mode and field.name not in explicit_field_names: - if not LazyDataclassFieldEmissionState.value_requires_clean_serialization( - current_value - ): - continue - - serialized_fields.append( - LazyDataclassSerializedField( - name=field.name, - fragment=to_source(current_value, field_ctx), - ) - ) - - return cls( - class_name=class_name, - import_pair=import_pair, - fields=tuple(serialized_fields), - ) - - -class LazyDataclassFormatEligibility: - @staticmethod - def accepts(candidate) -> bool: - return isinstance(candidate, LazyDataclass) - - -class FunctionStepCleanModeFieldPolicy: - def should_emit( - self, - field_name, - current_value, - default_value, - context: FormatContext, - ) -> bool: - if not context.clean_mode: - return True - - if isinstance(current_value, LazyDataclass): - emission_state = LazyDataclassFieldEmissionState.from_instance( - current_value - ) - if not emission_state.requires_serialization: - return False - if isinstance( - default_value, LazyDataclass - ) and LazyDataclassFieldEmissionState.raw_field_values_match( - current_value, - default_value, - ): - return False - return True - - return current_value != default_value - - -class LazyDataclassFormatter(SourceFormatter): - priority = 36 - - def can_format(self, value) -> bool: - return LazyDataclassFormatEligibility.accepts(value) - - def format(self, value: LazyDataclass, context: FormatContext) -> SourceFragment: - plan = LazyDataclassSerializationPlan.from_instance(value, context) - imports = {plan.import_pair} - field_ctx = context.indented() - lines = [] - - for field in plan.fields: - imports |= field.fragment.imports - lines.append(f"{field.name}={field.fragment.code}") - - if not lines: - return SourceFragment(f"{plan.class_name}()", frozenset(imports)) - - inner = f",\n{field_ctx.indent_str}".join(lines) - code = ( - f"{plan.class_name}(\n" - f"{field_ctx.indent_str}{inner}\n" - f"{context.indent_str})" - ) - return SourceFragment(code, frozenset(imports)) - - class FunctionStepFormatter(SourceFormatter): priority = 80 @@ -512,7 +225,6 @@ def format(self, value: FunctionStep, context: FormatContext) -> SourceFragment: field_ctx = context.indented() params = [] imports = set() - field_policy = FunctionStepCleanModeFieldPolicy() for name, param in signatures: if name in step_values: @@ -525,27 +237,21 @@ def format(self, value: FunctionStep, context: FormatContext) -> SourceFragment: else: default_value = param.default - if not field_policy.should_emit( - name, - current_value, - default_value, - context, - ): + if not field_needs_source(current_value, default_value, context): continue frag = to_source(current_value, field_ctx) imports |= frag.imports params.append(f"{name}={frag.code}") - import_pair = (FunctionStep.__module__, FunctionStep.__name__) - class_name = NameMappingLookup.resolve( - context, import_pair, FunctionStep.__name__ + class_ref = imported_name( + FunctionStep.__module__, FunctionStep.__qualname__, context ) - imports.add(import_pair) + imports |= class_ref.imports if not params: - return SourceFragment(f"{class_name}()", frozenset(imports)) + return SourceFragment(f"{class_ref.code}()", frozenset(imports)) inner = f",\n{field_ctx.indent_str}".join(params) - code = f"{class_name}(\n{field_ctx.indent_str}{inner}\n{context.indent_str})" + code = f"{class_ref.code}(\n{field_ctx.indent_str}{inner}\n{context.indent_str})" return SourceFragment(code, frozenset(imports)) diff --git a/openhcs/serialization/source_path_factoring.py b/openhcs/serialization/source_path_factoring.py deleted file mode 100644 index 0c8a5c3128..0000000000 --- a/openhcs/serialization/source_path_factoring.py +++ /dev/null @@ -1,195 +0,0 @@ -"""Document-level factoring for semantically typed Python source paths.""" - -from __future__ import annotations - -from collections import Counter -from collections.abc import Iterable, Mapping -from dataclasses import dataclass, field -from pathlib import Path -from types import MappingProxyType - -from openhcs.core.python_source_literal import PythonSourceLiteral -from pycodify import ( - Assignment, - BlankLine, - CodeBlock, - FormatContext, - generate_python_source, - to_source, -) - - -@dataclass(slots=True) -class SourcePathOccurrenceCollector: - """Ordered render-pass collector for actual or declaration-owned paths.""" - - values: list[Path] = field(default_factory=list) - - def record(self, value: Path) -> None: - self.values.append(value) - - -@dataclass(frozen=True, slots=True) -class FactoredPathSourceLiteral(PythonSourceLiteral): - """One factoring-plan-owned Python path expression.""" - - code: str - imports: frozenset[tuple[str, str]] = frozenset() - - def source_literal(self) -> str: - return self.code - - def source_literal_imports(self) -> frozenset[tuple[str, str]]: - return self.imports - - -@dataclass(frozen=True, slots=True) -class SourcePathFactoringPlan: - """Immutable path bindings and substitutions for one source document.""" - - bindings: tuple[Assignment, ...] - _expressions: Mapping[Path, PythonSourceLiteral] - - @classmethod - def from_occurrences( - cls, - occurrences: Iterable[Path], - ) -> "SourcePathFactoringPlan": - normalized = tuple( - _lexically_normalized(path) for path in occurrences if path.is_absolute() - ) - counts = Counter(normalized) - ordered_values = tuple(dict.fromkeys(normalized)) - partitions: dict[tuple[str, str], list[Path]] = {} - for value in ordered_values: - first_component = value.parts[1] if len(value.parts) > 1 else "" - partitions.setdefault((value.anchor, first_component), []).append(value) - - bindings: list[Assignment] = [] - root_binding_by_value: dict[Path, tuple[str, Path]] = {} - root_index = 0 - for values in partitions.values(): - if len(values) < 2: - continue - root = _common_ancestor(values) - if root is None or root == Path(root.anchor): - continue - root_index += 1 - root_name = "path_root" if root_index == 1 else f"path_root_{root_index}" - bindings.append( - Assignment( - root_name, - FactoredPathSourceLiteral( - f"Path({str(root)!r})", - frozenset((("pathlib", "Path"),)), - ), - ) - ) - for value in values: - root_binding_by_value[value] = (root_name, root) - - expressions: dict[Path, PythonSourceLiteral] = {} - path_index = 0 - for value in ordered_values: - root_binding = root_binding_by_value.get(value) - factored_expression = ( - None - if root_binding is None - else _descendant_expression(root_binding[0], root_binding[1], value) - ) - is_root_binding = root_binding is not None and value == root_binding[1] - if counts[value] > 1 and not is_root_binding: - path_index += 1 - path_name = f"path_{path_index}" - binding_value = factored_expression or FactoredPathSourceLiteral( - f"Path({str(value)!r})", - frozenset((("pathlib", "Path"),)), - ) - bindings.append(Assignment(path_name, binding_value)) - expressions[value] = FactoredPathSourceLiteral(path_name) - continue - if factored_expression is not None: - expressions[value] = factored_expression - - return cls( - bindings=tuple(bindings), - _expressions=MappingProxyType(expressions), - ) - - def expression_for(self, value: Path) -> PythonSourceLiteral | None: - if not value.is_absolute(): - return None - return self._expressions.get(_lexically_normalized(value)) - - -@dataclass(frozen=True, slots=True) -class OpenHCSPythonSourceDocument: - """Complete OpenHCS Python document rendered with one factoring plan.""" - - body: object - header: str = "" - clean_mode: bool = False - - def render(self) -> str: - collector = SourcePathOccurrenceCollector() - discovery_context = FormatContext( - clean_mode=self.clean_mode, - extensions=MappingProxyType({SourcePathOccurrenceCollector: collector}), - ) - to_source(self.body, discovery_context) - plan = SourcePathFactoringPlan.from_occurrences(collector.values) - body_items = ( - self.body.items if isinstance(self.body, CodeBlock) else (self.body,) - ) - items: tuple[object, ...] = body_items - if plan.bindings: - items = (*plan.bindings, BlankLine(), *body_items) - render_context = FormatContext( - clean_mode=self.clean_mode, - extensions=MappingProxyType({SourcePathFactoringPlan: plan}), - ) - return generate_python_source( - CodeBlock.from_items(items), - self.header, - self.clean_mode, - context=render_context, - ) - - -def _lexically_normalized(value: Path) -> Path: - if not value.is_absolute(): - return value - components: list[str] = [] - for component in value.parts[1:]: - if component in ("", "."): - continue - if component == "..": - if components: - components.pop() - continue - components.append(component) - return Path(value.anchor, *components) - - -def _common_ancestor(values: list[Path]) -> Path | None: - first_parts = values[0].parts - common_count = 0 - for components in zip(*(value.parts for value in values), strict=False): - if len(set(components)) != 1: - break - common_count += 1 - if common_count == 0: - return None - return Path(first_parts[0], *first_parts[1:common_count]) - - -def _descendant_expression( - root_name: str, - root: Path, - value: Path, -) -> FactoredPathSourceLiteral: - relative_parts = value.parts[len(root.parts) :] - code = root_name - for part in relative_parts: - code = f"{code} / {part!r}" - return FactoredPathSourceLiteral(code) diff --git a/openhcs/ui/shared/plate_manager_code_document.py b/openhcs/ui/shared/plate_manager_code_document.py index e6f21d893c..50e1d2f683 100644 --- a/openhcs/ui/shared/plate_manager_code_document.py +++ b/openhcs/ui/shared/plate_manager_code_document.py @@ -205,10 +205,7 @@ def render( clean_mode: bool = True, ) -> str: import openhcs.serialization.pycodify_formatters # noqa: F401 - from pycodify import Assignment, BlankLine, CodeBlock - from openhcs.serialization.source_path_factoring import ( - OpenHCSPythonSourceDocument, - ) + from pycodify import Assignment, BlankLine, CodeBlock, generate_python_source normalized = cls.from_values( plate_paths=payload.plate_paths, @@ -252,11 +249,11 @@ def render( ), ) ) - return OpenHCSPythonSourceDocument( + return generate_python_source( body, header=cls.HEADER, clean_mode=clean_mode, - ).render() + ) @staticmethod def _scope_id(value: object, *, field_name: str) -> str: diff --git a/pyproject.toml b/pyproject.toml index 5652ca10e9..8135a42298 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -100,8 +100,8 @@ dependencies = [ # External dependencies (PyPI versions for production builds) "zmqruntime>=0.6.1,<0.7", - "pycodify>=0.1.3,<0.2", - "objectstate>=2.0,<3", + "pycodify>=0.2.0,<0.3", + "objectstate>=2.1,<3", "python-introspect>=0.2.7,<0.3", "metaclass-registry>=0.3.0,<0.4", "arraybridge>=0.4.1,<0.5", diff --git a/tests/unit/agent/test_authoring_roundtrip_boundaries.py b/tests/unit/agent/test_authoring_roundtrip_boundaries.py index 6a548fe6ad..30605b323a 100644 --- a/tests/unit/agent/test_authoring_roundtrip_boundaries.py +++ b/tests/unit/agent/test_authoring_roundtrip_boundaries.py @@ -25,7 +25,6 @@ ) from openhcs.core.pipeline.funcstep_contract_validator import FuncStepContractValidator from openhcs.core.pipeline_document import PipelineDocumentCodec -from openhcs.core.python_source_literal import PythonSourceLiteral from openhcs.core.steps.abstract import AbstractStep from openhcs.mcp.dev_client_core import McpDevToolResult from openhcs.processing.backends.lib_registry.registry_service import RegistryService @@ -227,32 +226,6 @@ def source_expression(self, imported_name): ) -def test_independent_source_capability_composes_through_cooperative_mro(gamma_metadata): - class ExtraImport(PythonSourceLiteral): - def source_literal_imports(self): - return super().source_literal_imports() | frozenset( - {(__name__, "step_receipt")} - ) - - class ComposedReference(RegistryFunctionReference, ExtraImport): - pass - - reference = FunctionReferenceTransportAuthority.function_reference( - gamma_metadata.func - ) - member = ComposedReference( - import_identity=reference.import_identity, composite_key=reference.composite_key - ) - fragment = to_source(member, FormatContext()) - assert fragment.code == "get_function('skimage:exposure.adjust_gamma')" - assert fragment.imports == frozenset( - { - ("openhcs.processing.func_registry", "get_function"), - (__name__, "step_receipt"), - } - ) - - @pytest.mark.parametrize("declaration", (NewJsonOwner, int)) def test_importable_type_formatter_uses_shared_source_literal_owner(declaration): fragment = to_source(declaration, FormatContext()) diff --git a/tests/unit/test_code_mode_path_artifact_architecture.py b/tests/unit/test_code_mode_path_artifact_architecture.py index 2202a89082..504feeee1f 100644 --- a/tests/unit/test_code_mode_path_artifact_architecture.py +++ b/tests/unit/test_code_mode_path_artifact_architecture.py @@ -7,7 +7,6 @@ ROOT = Path(__file__).parents[2] OWNED_BOUNDARIES = ( - ROOT / "openhcs/serialization/source_path_factoring.py", ROOT / "openhcs/core/pipeline/compilation_session.py", ROOT / "openhcs/core/artifact_inspection.py", ROOT / "openhcs/core/progress/runtime_artifacts.py", diff --git a/tests/unit/test_l6_codegen_guards.py b/tests/unit/test_l6_codegen_guards.py new file mode 100644 index 0000000000..064182389c --- /dev/null +++ b/tests/unit/test_l6_codegen_guards.py @@ -0,0 +1,75 @@ +"""L6 guards: generic source generation lives in pycodify and ObjectState.""" + +from __future__ import annotations + +import ast +from pathlib import Path + +ROOT = Path(__file__).parents[2] +OPENHCS = ROOT / "openhcs" + +# Value families pycodify (Enum, Path, type, PythonSourceLiteral) and +# objectstate.codegen (lazy dataclasses) format; OpenHCS adds no formatter for them. +LIBRARY_FORMATTED = { + "Enum", + "Path", + "PythonSourceLiteral", + "LazyDataclass", + "LazyResolutionDataclass", +} + + +def _python_files(): + return sorted(OPENHCS.rglob("*.py")) + + +def test_deleted_codegen_copies_stay_deleted(): + assert not (OPENHCS / "core/python_source_literal.py").exists() + assert not (OPENHCS / "serialization/source_path_factoring.py").exists() + + +def test_no_openhcs_formatter_claims_a_library_value_family(): + violations = [] + for path in _python_files(): + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if not (isinstance(node, ast.FunctionDef) and node.name == "can_format"): + continue + for call in ast.walk(node): + if not isinstance(call, ast.Call): + continue + callee = ast.unparse(call.func) + if callee == "inspect.isclass": + violations.append((path, call.lineno, callee)) + if callee == "isinstance" and len(call.args) == 2: + claimed = { + ast.unparse(name).rpartition(".")[2] + for name in ast.walk(call.args[1]) + if isinstance(name, (ast.Name, ast.Attribute)) + } + if claimed & LIBRARY_FORMATTED: + violations.append((path, call.lineno, sorted(claimed))) + assert violations == [] + + +def test_no_qualified_name_or_lazy_field_restatements(): + forbidden = ("__qualname__.partition", "_explicitly_set_fields") + violations = [ + (path, needle) + for path in _python_files() + for needle in forbidden + if needle in path.read_text(encoding="utf-8") + ] + assert violations == [] + + +def test_objectstate_names_no_application_field(): + source_root = ROOT / "external/ObjectState/src" + violations = [ + path + for path in sorted(source_root.rglob("*")) + if path.is_file() + and path.suffix in {".py", ".rst", ".md"} + and "well_filter" in path.read_text(encoding="utf-8") + ] + assert violations == [] diff --git a/tests/unit/test_pycodify_formatters.py b/tests/unit/test_pycodify_formatters.py index 1d8c7bf527..09538697ee 100644 --- a/tests/unit/test_pycodify_formatters.py +++ b/tests/unit/test_pycodify_formatters.py @@ -4,16 +4,13 @@ import inspect import pytest -from arraybridge.decorators import DtypeConversion from objectstate import semantic_values_equal from objectstate.object_state import ObjectState -from pycodify import Assignment, generate_python_source +from pycodify import Assignment, generate_python_source, to_source import openhcs.serialization.pycodify_formatters # noqa: F401 -from openhcs.constants import InputSource from openhcs.core.callable_contract import CallableContract, CallableImportIdentity from openhcs.core.config import ( - DtypeConfig, LazyDtypeConfig, LazyNapariStreamingConfig, LazyProcessingConfig, @@ -172,13 +169,6 @@ def test_clean_pipeline_config_keeps_explicit_empty_nested_lazy_config(): assert "partition_filter_config=LazyPartitionFilterConfig()" in source -def test_dtype_conversion_serializes_from_owning_enum_module(): - source = _source(DtypeConfig(default_dtype_conversion=DtypeConversion.UINT8)) - - assert "from arraybridge.decorators import DtypeConversion" in source - assert "default_dtype_conversion=DtypeConversion.UINT8" in source - - def test_full_pipeline_config_still_emits_empty_lazy_config_groups(): source = _source(PipelineConfig(), clean_mode=False) @@ -198,7 +188,7 @@ def test_clean_function_step_omits_objectstate_reconstructed_empty_lazy_configs( ) reference = FunctionReferenceTransportAuthority.function_reference(count_cells_simple) - assert f"func={reference.source_literal()}" in source + assert f"func={to_source(reference).code}" in source assert "name='count_cells'" in source assert "dtype_config=" not in source assert "processing_config=" not in source @@ -245,17 +235,6 @@ def test_clean_lazy_dataclass_omits_explicit_none_leaf_fields(): assert "input_source=None" not in source -def test_full_lazy_dataclass_keeps_none_leaf_fields(): - source = _source( - LazyProcessingConfig(input_source=InputSource.PREVIOUS_STEP), - clean_mode=False, - ) - - assert "variable_components=None" in source - assert "group_by=None" in source - assert "input_source=InputSource.PREVIOUS_STEP" in source - - def test_function_pattern_clean_mode_elides_signature_default_kwargs(): source = generate_python_source( Assignment( diff --git a/tests/unit/test_source_path_factoring.py b/tests/unit/test_source_path_factoring.py deleted file mode 100644 index f4d338b4a2..0000000000 --- a/tests/unit/test_source_path_factoring.py +++ /dev/null @@ -1,164 +0,0 @@ -from __future__ import annotations - -from dataclasses import dataclass -from pathlib import Path - -import openhcs.serialization.pycodify_formatters # noqa: F401 -from openhcs.core.vfs_protocol import PlateInputFile -from openhcs.serialization.source_path_factoring import OpenHCSPythonSourceDocument -from pycodify import Assignment, CodeBlock - - -def declared_template_callable( - image: object, - *, - template_path: PlateInputFile, -) -> object: - return image - - -@dataclass(frozen=True, slots=True) -class PathBearingConfig: - actual_path: Path - path_lookalike: str - - -def test_document_factors_deepest_common_root_and_repeated_values() -> None: - plate_a = Path("/media/alice/T7/screen/plate_A") - plate_b = Path("/media/alice/T7/screen/plate_B") - template = Path("/media/alice/T7/screen/templates/fiducial.png") - source = OpenHCSPythonSourceDocument( - CodeBlock.from_items( - ( - Assignment("plate_paths", [plate_a, plate_b]), - Assignment("pipeline_data", {plate_a: template, plate_b: None}), - ) - ) - ).render() - - assert "path_root = Path('/media/alice/T7/screen')" in source - assert "path_1 = path_root / 'plate_A'" in source - assert "path_2 = path_root / 'plate_B'" in source - assert "path_root / 'templates' / 'fiducial.png'" in source - assert source.count("/media/alice/T7/screen") == 1 - - namespace: dict[str, object] = {} - exec(source, namespace) - assert namespace["plate_paths"] == [plate_a, plate_b] - assert namespace["pipeline_data"] == {plate_a: template, plate_b: None} - - -def test_document_does_not_factor_relative_or_string_lookalike_paths() -> None: - source = OpenHCSPythonSourceDocument( - Assignment( - "values", - [ - Path("templates/fiducial.png"), - "/media/alice/T7/screen/plate_A", - "/media/alice/T7/screen/plate_A", - ], - ) - ).render() - - assert "path_root" not in source - assert "path_1" not in source - assert "Path('templates/fiducial.png')" in source - assert source.count("'/media/alice/T7/screen/plate_A'") == 2 - - -def test_document_keeps_unrelated_absolute_trees_separate() -> None: - source = OpenHCSPythonSourceDocument( - Assignment( - "values", - [Path("/data/a/one"), Path("/data/a/two"), Path("/mnt/b/three")], - ) - ).render() - - assert "path_root = Path('/data/a')" in source - assert "path_root_2" not in source - assert "Path('/mnt/b/three')" in source - - -def test_document_extracts_one_repeated_absolute_path_without_root_binding() -> None: - source = OpenHCSPythonSourceDocument( - Assignment( - "values", - [Path("/opt/models/model.bin"), Path("/opt/models/model.bin")], - ) - ).render() - - assert "path_root" not in source - assert "path_1 = Path('/opt/models/model.bin')" in source - assert source.count("/opt/models/model.bin") == 1 - - -def test_document_uses_first_occurrence_order_for_multiple_coherent_roots() -> None: - source = OpenHCSPythonSourceDocument( - Assignment( - "values", - [ - Path("/mnt/zeta/one"), - Path("/mnt/zeta/two"), - Path("/data/alpha/one"), - Path("/data/alpha/two"), - ], - ) - ).render() - - assert source.index("path_root = Path('/mnt/zeta')") < source.index( - "path_root_2 = Path('/data/alpha')" - ) - - -def test_document_collects_dataclass_path_values_but_not_string_lookalikes() -> None: - absolute = Path("/data/screen/config/model.bin") - source = OpenHCSPythonSourceDocument( - Assignment( - "configs", - [ - PathBearingConfig(absolute, str(absolute)), - PathBearingConfig(absolute, str(absolute)), - ], - ) - ).render() - - assert source.count("/data/screen/config/model.bin") == 3 - assert "path_1 = Path('/data/screen/config/model.bin')" in source - - -def test_declared_string_callable_path_is_collected_structurally() -> None: - source = OpenHCSPythonSourceDocument( - Assignment( - "pattern", - ( - declared_template_callable, - {"template_path": "/data/screen/templates/fiducial.png"}, - ), - ) - ).render() - - assert "template_path" in source - assert "Path('/data/screen/templates/fiducial.png')" in source - - -def test_pipeline_declarations_share_their_deepest_common_absolute_root() -> None: - source = OpenHCSPythonSourceDocument( - Assignment( - "pattern", - [ - ( - declared_template_callable, - {"template_path": "/media/alice/T7/screen/a/template.tif"}, - ), - ( - declared_template_callable, - {"template_path": "/media/alice/T7/screen/b/template.tif"}, - ), - ], - ) - ).render() - - assert "path_root = Path('/media/alice/T7/screen')" in source - assert "path_root / 'a' / 'template.tif'" in source - assert "path_root / 'b' / 'template.tif'" in source - assert source.count("/media/alice/T7/screen") == 1 diff --git a/tests/unit/test_source_round_trip.py b/tests/unit/test_source_round_trip.py new file mode 100644 index 0000000000..09417710ec --- /dev/null +++ b/tests/unit/test_source_round_trip.py @@ -0,0 +1,342 @@ +"""Saved Python source loads back as the value that was saved. + +Pipelines and configs persist as Python source generated by pycodify, so every +formatter gap is a data-loss bug. This test generates every registered config +dataclass (lazy and base) with every field set to a non-default value derived +from its annotation, and every function-pattern shape the step editor produces +inside a pipeline document, writes them as source, loads the source back and +compares the raw values. +""" + +from __future__ import annotations + +import enum +import inspect +import itertools +import types +import typing +from collections.abc import Iterator, Mapping +from dataclasses import MISSING, fields, is_dataclass +from pathlib import Path + +import pytest +from objectstate import LazyDataclassFactory +from objectstate.field_access import DataclassFieldAccess +from pycodify import Assignment, generate_python_source + +import openhcs.serialization.pycodify_formatters # noqa: F401 +from openhcs.core.config import PipelineConfig +from openhcs.core.dataset_sources.choice import DatasetSourceChoice +from openhcs.core.pipeline_document import PipelineDocumentCodec +from openhcs.core.steps.abstract import AbstractStep +from openhcs.core.steps.function_step import FunctionStep +from openhcs.core.vfs_protocol import PlateInputFile +from openhcs.processing.backends.analysis.count_cells_simple import count_cells_simple +from openhcs.processing.backends.cellprofiler.crop import CropModule, crop +from openhcs.processing.backends.cellprofiler.shape import measure_object_size_shape +from openhcs.processing.backends.lib_registry.registry_service import RegistryService + +DatasetSourceChoice.choices() # discovers the registered dataset sources + + +ROOT = Path("/data/screen") + + +# --------------------------------------------------------------------------- +# Values derived from annotations +# --------------------------------------------------------------------------- + + +class SampleValues: + """Candidate values of each annotation that differ from a given default. + + The first candidate the owning dataclass accepts is used, so validation + rules (a regex that needs a named group, an axis that cannot vary) pick + the next candidate instead of losing the field. + """ + + def __init__(self) -> None: + self.unsampled: list[str] = [] + + def candidates(self, annotation, default, label: str, depth: int) -> Iterator[object]: + origin = typing.get_origin(annotation) + args = typing.get_args(annotation) + if origin is typing.Annotated: + yield from self.candidates(args[0], default, label, depth) + elif origin in (typing.Union, types.UnionType): + for arg in args: + if arg is not type(None): + yield from self.candidates(arg, default, label, depth) + elif origin is typing.Literal: + yield from (arg for arg in args if arg != default) + elif origin is type: + yield from self.subclasses(args[0], default) + elif origin in (list, tuple): + for item in itertools.islice(self.candidates(args[0], None, label, depth + 1), 4): + yield origin((item,)) + elif origin in (dict, Mapping): + keys = self.candidates(args[0], None, label, depth + 1) + items = self.candidates(args[1], None, label, depth + 1) + for key, item in itertools.islice(zip(keys, items), 4): + yield {key: item} + elif annotation is bool: + yield not default if isinstance(default, bool) else True + elif annotation is int: + yield default + 1 if isinstance(default, int) else 7 + elif annotation is float: + yield default + 0.5 if isinstance(default, float) else 2.5 + elif annotation is str: + yield f"{default}_x" if isinstance(default, str) else "sample" + yield r"(?P\w+)" + elif annotation is Path: + yield ROOT / label.replace(".", "_") + elif isinstance(annotation, type) and issubclass(annotation, enum.Enum): + yield from (member for member in reversed(annotation) if member != default) + elif isinstance(annotation, type) and is_dataclass(annotation) and depth < 3: + instance = self.instance(annotation, label, depth + 1) + if instance is not _NO_SAMPLE: + yield instance + + @staticmethod + def subclasses(base: type, default: object) -> Iterator[type]: + if base is object: + yield int + return + for candidate in reversed([base, *_all_subclasses(base)]): + if candidate is default or "" in candidate.__qualname__: + continue + if inspect.isabstract(candidate) or candidate.__module__.startswith("tests"): + continue + yield candidate + + def accepted(self, owner: type, kwargs: dict, name: str, annotation, default, label, depth): + for sample in itertools.islice(self.candidates(annotation, default, label, depth), 12): + trial = {**kwargs, name: sample} + if _constructs(owner, trial): + return trial + return None + + def instance(self, dataclass_type: type, label: str, depth: int): + kwargs: dict = {} + hints = typing.get_type_hints(dataclass_type) + for declared in fields(dataclass_type): + if not declared.init: + continue + default = _field_default(declared) + trial = self.accepted( + dataclass_type, kwargs, declared.name, hints[declared.name], default, + f"{label}.{declared.name}", depth, + ) + if trial is None and default is _NO_SAMPLE: + trial = self.accepted_required(dataclass_type, kwargs, declared, hints, label, depth) + if trial is None: + if default is _NO_SAMPLE: + return _NO_SAMPLE + continue + kwargs = trial + return dataclass_type(**kwargs) if _constructs(dataclass_type, kwargs) else _NO_SAMPLE + + def accepted_required(self, dataclass_type, kwargs, declared, hints, label, depth): + """A required field whose candidates fail only because later fields are unset.""" + + for sample in itertools.islice( + self.candidates(hints[declared.name], None, f"{label}.{declared.name}", depth), 12 + ): + return {**kwargs, declared.name: sample} + return None + + +_NO_SAMPLE = object() + + +def _all_subclasses(base: type) -> list[type]: + found = [] + for child in type.__subclasses__(base): + found.append(child) + found.extend(_all_subclasses(child)) + return found + + +def _field_default(declared): + if declared.default is not MISSING: + return declared.default + if declared.default_factory is not MISSING: + return declared.default_factory() + return _NO_SAMPLE + + +def _constructs(dataclass_type: type, kwargs: dict) -> bool: + try: + dataclass_type(**kwargs) + except (TypeError, ValueError): + return False + return True + + +def raw(value): + """The value as nested plain data, reading dataclass fields without resolution.""" + + if is_dataclass(value) and not isinstance(value, type): + return ( + type(value), + { + declared.name: raw(DataclassFieldAccess.raw_value(value, declared.name)) + for declared in fields(value) + if declared.init + }, + ) + if isinstance(value, FunctionStep): + return (FunctionStep, {name: raw(vars(value)[name]) for name in STEP_PARAMETERS}) + if isinstance(value, (list, tuple)): + return (type(value), [raw(item) for item in value]) + if isinstance(value, dict): + return (dict, {raw(key): raw(item) for key, item in value.items()}) + return value + + +def load(value, clean_mode: bool): + source = generate_python_source(Assignment("value", value), clean_mode=clean_mode) + namespace: dict[str, object] = {} + exec(compile(source, "", "exec"), namespace) + return source, namespace["value"] + + +# --------------------------------------------------------------------------- +# The config family +# --------------------------------------------------------------------------- + + +CONFIG_TYPES = tuple( + config_type + for pair in LazyDataclassFactory.registered_type_pairs() + for config_type in pair + if not inspect.isabstract(config_type) +) + + +def config_sample(config_type: type, samples: SampleValues): + hints = typing.get_type_hints(config_type) + kwargs: dict = {} + for declared in fields(config_type): + if not declared.init: + continue + trial = samples.accepted( + config_type, kwargs, declared.name, hints[declared.name], + _field_default(declared), f"{config_type.__qualname__}.{declared.name}", 0, + ) + if trial is None: + samples.unsampled.append(f"{config_type.__qualname__}.{declared.name}") + continue + kwargs = trial + return config_type(**kwargs) + + +@pytest.mark.parametrize("clean_mode", (True, False), ids=("clean", "full")) +@pytest.mark.parametrize("config_type", CONFIG_TYPES, ids=lambda t: t.__qualname__) +def test_every_config_loads_back_with_every_field_set(config_type, clean_mode): + samples = SampleValues() + value = config_sample(config_type, samples) + source, loaded = load(value, clean_mode) + assert raw(loaded) == raw(value), source + assert samples.unsampled == [] + + +@pytest.mark.parametrize("clean_mode", (True, False), ids=("clean", "full")) +@pytest.mark.parametrize("config_type", CONFIG_TYPES, ids=lambda t: t.__qualname__) +def test_every_default_config_loads_back(config_type, clean_mode): + value = config_type() + source, loaded = load(value, clean_mode) + assert raw(loaded) == raw(value), source + + +# --------------------------------------------------------------------------- +# Steps and function patterns in a pipeline document +# --------------------------------------------------------------------------- + + +STEP_PARAMETERS = ("func",) + tuple( + name for name in inspect.signature(AbstractStep.__init__).parameters if name != "self" +) + + +def configurable_function(image, threshold: int = 3, enabled: bool = True): + return image + + +def template_function(image, *, template_path: PlateInputFile): + return image + + +PATTERNS = { + "callable": count_cells_simple, + "callable_with_kwargs": (configurable_function, {"threshold": 9, "enabled": False}), + "declared_path_kwarg": ( + template_function, + {"template_path": ROOT / "templates" / "fiducial.png"}, + ), + "nested_enum_kwargs": ( + crop, + {"crop_shape": CropModule.Shape.ELLIPSE, "removal_method": CropModule.RemovalMethod.ALL}, + ), + "hidden_runtime_kwargs": (measure_object_size_shape, {"calculate_zernikes": False}), + "chain": [count_cells_simple, (configurable_function, {"threshold": 4})], + "by_group": { + "1": count_cells_simple, + "2": [(configurable_function, {"threshold": 5}), count_cells_simple], + }, +} + + +@pytest.fixture +def declared_function_catalog(monkeypatch): + """Registry metadata for the pattern functions, without a catalog subprocess.""" + + metadata = ( + RegistryService.declared_metadata_for_callable(func) + for func in (count_cells_simple, crop, measure_object_size_shape) + ) + monkeypatch.setattr(RegistryService, "_metadata_cache", dict(metadata)) + + +def step_sample(func, samples: SampleValues) -> FunctionStep: + hints = typing.get_type_hints(AbstractStep.__init__, globalns=vars(inspect.getmodule(AbstractStep))) + kwargs: dict = {"func": func, "name": "step", "description": "described", "enabled": False} + for name, parameter in inspect.signature(AbstractStep.__init__).parameters.items(): + annotation = hints.get(name) + if name in kwargs or name == "self" or annotation is None: + continue + config_type = typing.get_args(annotation)[0] if typing.get_origin(annotation) else annotation + if isinstance(config_type, type) and is_dataclass(config_type): + kwargs[name] = config_sample(config_type, samples) + else: + kwargs[name] = next(samples.candidates(annotation, parameter.default, name, 0)) + return FunctionStep(**kwargs) + + +@pytest.mark.usefixtures("declared_function_catalog") +@pytest.mark.parametrize("clean_mode", (True, False), ids=("clean", "full")) +def test_pipeline_document_loads_back_every_pattern_shape(clean_mode): + samples = SampleValues() + document = PipelineDocumentCodec.from_values( + pipeline_config=config_sample(PipelineConfig, samples), + pipeline_steps=[step_sample(pattern, samples) for pattern in PATTERNS.values()], + ) + source = PipelineDocumentCodec.render(document, clean_mode=clean_mode) + loaded = PipelineDocumentCodec.from_source(source) + + assert raw(loaded.pipeline_config) == raw(document.pipeline_config), source + assert raw(loaded.pipeline_steps) == raw(document.pipeline_steps), source + assert samples.unsampled == [] + assert "path_root = Path('/data/screen')" in source + assert "CropModule.Shape.ELLIPSE" in source + + +@pytest.mark.usefixtures("declared_function_catalog") +def test_declared_path_kwarg_written_as_string_loads_back_as_path(): + path = ROOT / "templates" / "fiducial.png" + document = PipelineDocumentCodec.from_values( + pipeline_config=PipelineConfig(), + pipeline_steps=[FunctionStep(func=(template_function, {"template_path": str(path)}))], + ) + loaded = PipelineDocumentCodec.from_source(PipelineDocumentCodec.render(document)) + assert loaded.pipeline_steps[0].func[1] == {"template_path": path}