diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml index 5477bb2..c73b346 100644 --- a/.github/workflows/benchmarks.yml +++ b/.github/workflows/benchmarks.yml @@ -84,7 +84,7 @@ jobs: output-file-path: cosy-luigi-nested/benchmarks.json github-token: ${{ steps.cls-python-workflow-token.outputs.token }} auto-push: ${{ env.PUSH_TO_BENCHMARKS }} - alert-threshold: '300%' + alert-threshold: '50%' comment-on-alert: true fail-on-alert: true comment-always: ${{ env.COMMENT_ALWAYS }} diff --git a/benchmarks/test_benchmark_chain_100.py b/benchmarks/test_benchmark_chain_100.py index a7ce05c..013666c 100644 --- a/benchmarks/test_benchmark_chain_100.py +++ b/benchmarks/test_benchmark_chain_100.py @@ -4,7 +4,7 @@ from cosy.maestro import Maestro from luigi.mock import MockTarget -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter class ChainLink(CoSyLuigiTask, ABC): @@ -50,3 +50,7 @@ def create_infinite_chain(repo): def test_benchmark_chain_creation(repo, benchmark): benchmark(create_infinite_chain, repo) + + +if __name__ == "__main__": + create_infinite_chain(repo) diff --git a/examples/getting_started/basic_example.py b/examples/getting_started/basic_example.py index 59e9e28..a079bc6 100644 --- a/examples/getting_started/basic_example.py +++ b/examples/getting_started/basic_example.py @@ -3,7 +3,7 @@ import luigi from cosy.maestro import Maestro -from src.cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter class TaskA(CoSyLuigiTask): diff --git a/examples/getting_started/variation_example.py b/examples/getting_started/variation_example.py index 5b5c3e1..49ab853 100644 --- a/examples/getting_started/variation_example.py +++ b/examples/getting_started/variation_example.py @@ -5,7 +5,7 @@ import luigi from cosy.maestro import Maestro -from src.cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter class WriteTemplateTask(CoSyLuigiTask): diff --git a/examples/lot_sizing/lot_sizing_pipeline.py b/examples/lot_sizing/lot_sizing_pipeline.py index da65380..a775326 100644 --- a/examples/lot_sizing/lot_sizing_pipeline.py +++ b/examples/lot_sizing/lot_sizing_pipeline.py @@ -13,7 +13,7 @@ from lot_optimizers.silver_meal_heuristic import SilverMeal from lot_optimizers.wagner_whitin import WagnerWhitin -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter class GetCosts(CoSyLuigiTask): diff --git a/examples/ml_blood_sugar_level/ml_blood_sugar_level.py b/examples/ml_blood_sugar_level/ml_blood_sugar_level.py index 0042995..2e5365d 100644 --- a/examples/ml_blood_sugar_level/ml_blood_sugar_level.py +++ b/examples/ml_blood_sugar_level/ml_blood_sugar_level.py @@ -1,7 +1,6 @@ import json import textwrap from abc import ABC -from collections.abc import Callable, Iterable, Mapping, Sequence from pathlib import Path import luigi @@ -15,7 +14,7 @@ from sklearn.model_selection import train_test_split from sklearn.preprocessing import MinMaxScaler, RobustScaler -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter ninetydegaisle = True @@ -146,7 +145,7 @@ def run(self): class EvaluateRegressionModel(CoSyLuigiTask): regressor = CoSyLuigiTaskParameter(TrainRegressionModel) - scaled_feats = CoSyLuigiTaskParameter(FitTransformScaler) + scaled_feats = CoSyLuigiTaskParameter(FitTransformScaler, unique_across_prior_tasks=True) splitted_data = CoSyLuigiTaskParameter(TrainTestSplit) def _get_variant_label(self): @@ -170,31 +169,6 @@ def run(self): y_pred.to_json(self.output().path) - @classmethod - def constraints(cls) -> Sequence[Callable[..., bool]]: - return [lambda vs: check_unique(vs, [FitTransformScaler]), lambda _: ninetydegaisle] - - -def traverse_pipeline(vs: Iterable[CoSyLuigiTask]) -> Iterable[CoSyLuigiTask]: - result = [*vs] - for v in result: - result.extend(traverse_pipeline(v.requires().values())) - return result - - -def check_unique(vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: Iterable[type[CoSyLuigiTask]]) -> bool: - classes = [pc.__class__ for pc in traverse_pipeline(vs.values())] - seen_subclasses = {} - for c in classes: - for unique in required_to_be_unique: - if issubclass(c, unique): - if unique in seen_subclasses: - if seen_subclasses[unique] != c: - return False - else: - seen_subclasses[unique] = c - return True - def main(): repo = CoSyLuigiRepo( diff --git a/pyproject.toml b/pyproject.toml index a18c127..90ff5a9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -76,9 +76,12 @@ exclude_lines = [ "no cov", "if __name__ == .__main__.:", "if TYPE_CHECKING:", - "@abstractmethod", - "@abc.abstractmethod" ] +exclude_also = [ + "raise AssertionError", + "raise NotImplementedError", + "@(abc\\.)?abstractmethod", + ] [tool.ruff.lint.extend-per-file-ignores] diff --git a/src/cosy_luigi/__init__.py b/src/cosy_luigi/__init__.py index e69de29..ac789a5 100644 --- a/src/cosy_luigi/__init__.py +++ b/src/cosy_luigi/__init__.py @@ -0,0 +1,3 @@ +from cosy_luigi.core import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter + +__all__ = ["CoSyLuigiRepo", "CoSyLuigiTask", "CoSyLuigiTaskParameter"] diff --git a/src/cosy_luigi/constraints/__init__.py b/src/cosy_luigi/constraints/__init__.py new file mode 100644 index 0000000..8c8ee8b --- /dev/null +++ b/src/cosy_luigi/constraints/__init__.py @@ -0,0 +1,3 @@ +from cosy_luigi.constraints.unique import is_unique_in_prior_tasks + +__all__ = ["is_unique_in_prior_tasks"] diff --git a/src/cosy_luigi/constraints/unique.py b/src/cosy_luigi/constraints/unique.py new file mode 100644 index 0000000..8f0e9df --- /dev/null +++ b/src/cosy_luigi/constraints/unique.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from collections.abc import Sequence +from typing import TYPE_CHECKING + +from cosy_luigi.utils import traverse_pipeline + +if TYPE_CHECKING: + from collections.abc import Mapping + + from cosy_luigi import CoSyLuigiTask + + +def _is_unique_in_prior_tasks( + vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: Sequence[type[CoSyLuigiTask]] +) -> bool: + classes = [pc.__class__ for pc in traverse_pipeline(vs.values())] + seen_subclasses: dict[type[CoSyLuigiTask], type[CoSyLuigiTask]] = {} + for c in classes: + for unique in required_to_be_unique: + if issubclass(c, unique): + if unique in seen_subclasses: + if seen_subclasses[unique] != c: + return False + else: + seen_subclasses[unique] = c + return True + + +def is_unique_in_prior_tasks( + vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: type[CoSyLuigiTask] | Sequence[type[CoSyLuigiTask]] +) -> bool: + return _is_unique_in_prior_tasks( + vs, + required_to_be_unique if isinstance(required_to_be_unique, Sequence) else [required_to_be_unique], + ) diff --git a/src/cosy_luigi/core/__init__.py b/src/cosy_luigi/core/__init__.py new file mode 100644 index 0000000..160decc --- /dev/null +++ b/src/cosy_luigi/core/__init__.py @@ -0,0 +1,3 @@ +from cosy_luigi.core.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter + +__all__ = ["CoSyLuigiRepo", "CoSyLuigiTask", "CoSyLuigiTaskParameter"] diff --git a/src/cosy_luigi/combinatorics.py b/src/cosy_luigi/core/combinatorics.py similarity index 51% rename from src/cosy_luigi/combinatorics.py rename to src/cosy_luigi/core/combinatorics.py index 56899b1..fb09bd7 100644 --- a/src/cosy_luigi/combinatorics.py +++ b/src/cosy_luigi/core/combinatorics.py @@ -1,25 +1,28 @@ from __future__ import annotations -import inspect -from abc import ABC +import logging +import textwrap from collections import defaultdict -from functools import cache -from typing import TYPE_CHECKING, cast +from functools import cache, partial +from typing import TYPE_CHECKING import luigi from cosy.core import Constructor, SpecificationBuilder from luigi.task_register import Register if TYPE_CHECKING: - from collections.abc import Callable, Iterable, Mapping, Sequence + from collections.abc import Callable, Mapping, Sequence from cosy.core.synthesizer import Specification +logger = logging.getLogger(__name__) + class CoSyLuigiTaskParameter(luigi.TaskParameter): - def __init__(self, required_task: type[CoSyLuigiTask]): + def __init__(self, required_task: type[CoSyLuigiTask], *, unique_across_prior_tasks: bool = False): super().__init__() self.required_task = required_task + self.unique_across_prior_tasks = unique_across_prior_tasks class CoSyLuigiTask(luigi.Task): @@ -60,13 +63,32 @@ def requires(self): @classmethod @cache - def _requirements(cls): + def _requirements(cls) -> Mapping[str, CoSyLuigiTaskParameter]: return { k: v for k, v in cls.get_all_class_attributes().items() if not k.startswith("__") and not callable(v) and issubclass(v.__class__, CoSyLuigiTaskParameter) } + @classmethod + @cache + def get_params(cls): + return list(cls._requirements().items()) + + @classmethod + @cache + def requirements_unique_in_prior_tasks(cls) -> Mapping[str, CoSyLuigiTaskParameter]: + return { + k: task_parameter + for k, task_parameter in cls._requirements().items() + if task_parameter.unique_across_prior_tasks + } + + @classmethod + @cache + def unique_required_tasks_in_prior(cls) -> Sequence[type[CoSyLuigiTask]]: + return [task_parameter.required_task for task_parameter in cls.requirements_unique_in_prior_tasks().values()] + @classmethod @cache def target(cls): @@ -76,11 +98,21 @@ def target(cls): def constraints(cls) -> Sequence[Callable[..., bool]]: return [] + @classmethod + def __constraints(cls) -> Sequence[Callable[..., bool]]: + from cosy_luigi.constraints.unique import _is_unique_in_prior_tasks # noqa: PLC0415 + + if cls.requirements_unique_in_prior_tasks(): + return [partial(_is_unique_in_prior_tasks, required_to_be_unique=cls.unique_required_tasks_in_prior())] + return [] + @classmethod def combinator_type(cls): sp = SpecificationBuilder() for name in [v.required_task.__name__ for v in cls._requirements().values()]: sp = sp.argument(name, Constructor(name)) + for constraint in cls.__constraints(): + sp = sp.constraint(constraint) for constraint in cls.constraints(): sp = sp.constraint(constraint) return sp.suffix(cls.target()) @@ -93,32 +125,53 @@ def combinator(cls): class CoSyLuigiRepo: - def __init__(self, *tasks: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]): + def __init__(self, *tasks: type[CoSyLuigiTask] | Sequence[type[CoSyLuigiTask]]): Register.disable_instance_cache() # Accepts completely heterogeneous nested collections - def flatten(*heterogeneous_task_collection: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]): - return ( - task - for task_or_task_collection in heterogeneous_task_collection - for task in ( - flatten(*cast("Iterable[type[CoSyLuigiTask]]", task_or_task_collection)) - if isinstance(task_or_task_collection, (tuple, list)) - else cast("type[CoSyLuigiTask]", task_or_task_collection).get_all_variants() - if inspect.isabstract(task_or_task_collection) - or ABC in cast("type[CoSyLuigiTask]", task_or_task_collection).__bases__ - else (task_or_task_collection,) - ) - ) # This doesn't technically need to unpack as flatten could be typed to accept packed tuples # But performance is equivalent/faster because the first layer doesn't need to be checked this way + from cosy_luigi.utils import flatten # noqa: PLC0415 + self.luigi_repo: set[type[CoSyLuigiTask]] = set(flatten(*tasks)) + self.check_unique_in_prior_tasks_sanity() self.taxonomy: Mapping[str, set[str]] = defaultdict(set) - self.cls_repo: set[tuple[str, Callable, Specification]] = set() + self.cls_repo: list[tuple[str, Callable, Specification]] = [] for task in self.luigi_repo: - self.cls_repo.add(task.combinator()) + self.cls_repo.append(task.combinator()) for tpe in task.mro()[1:]: if issubclass(tpe, CoSyLuigiTask): # Is a subclass of CosyLuigiTask, but a superclass of task self.taxonomy[task.__name__].add(tpe.__name__) + + def check_unique_in_prior_tasks_sanity(self): + for source_task, param_name, required_type in [ + (task, k, required_unique_task.required_task) + for task in self.luigi_repo + for k, required_unique_task in task.requirements_unique_in_prior_tasks().items() + if not any( + issubclass(task, required_unique_task.required_task) and task is not required_unique_task.required_task + for task in self.luigi_repo + ) + ]: + logger.warning( + textwrap.dedent( + f""" + ================================================================= + WARNING ABOUT POTENTIALLY INCORRECT MODEL + + Class: {source_task.__name__} + Parameter: {param_name} + Type: {required_type.__name__} + + is required to be unique, but there are no sub- + classes of it present in the repository. Either + you forgot adding sub-classes of it to the rep- + ository or this will not behave as expected. + + Please head over to the documentation: + ================================================================= + """ + ) + ) diff --git a/src/cosy_luigi/utils/__init__.py b/src/cosy_luigi/utils/__init__.py new file mode 100644 index 0000000..ebd4586 --- /dev/null +++ b/src/cosy_luigi/utils/__init__.py @@ -0,0 +1,3 @@ +from cosy_luigi.utils.traversals import flatten, traverse_pipeline + +__all__ = ["flatten", "traverse_pipeline"] diff --git a/src/cosy_luigi/utils/traversals.py b/src/cosy_luigi/utils/traversals.py new file mode 100644 index 0000000..f90bdb7 --- /dev/null +++ b/src/cosy_luigi/utils/traversals.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import inspect +from abc import ABC +from typing import TYPE_CHECKING, cast + +from cosy_luigi import CoSyLuigiTask + +if TYPE_CHECKING: + from collections.abc import Iterable, Sequence + + +def flatten(*heterogeneous_task_collection: type[CoSyLuigiTask] | Sequence[type[CoSyLuigiTask]]): + return ( + task + for task_or_task_collection in heterogeneous_task_collection + for task in ( + flatten(*cast("Sequence[type[CoSyLuigiTask]]", task_or_task_collection)) + if isinstance(task_or_task_collection, (tuple, list)) + else cast("type[CoSyLuigiTask]", task_or_task_collection).get_all_variants() + if inspect.isabstract(task_or_task_collection) + or ABC in cast("type[CoSyLuigiTask]", task_or_task_collection).__bases__ + else (task_or_task_collection,) + ) + ) + + +def _traverse_pipeline(vs: Sequence[CoSyLuigiTask] | Iterable[CoSyLuigiTask]) -> Sequence[CoSyLuigiTask]: + result = [*vs] + for v in vs: + result.extend(traverse_pipeline(v.requires().values())) + return result + + +def traverse_pipeline( + to_traverse: CoSyLuigiTask | Sequence[CoSyLuigiTask] | Iterable[CoSyLuigiTask], +) -> Sequence[CoSyLuigiTask]: + return ( + _traverse_pipeline([to_traverse]) if isinstance(to_traverse, CoSyLuigiTask) else _traverse_pipeline(to_traverse) + ) diff --git a/tests/test_abstract_variant_expansion.py b/tests/test_abstract_variant_expansion.py index 6bbcfd1..6978448 100644 --- a/tests/test_abstract_variant_expansion.py +++ b/tests/test_abstract_variant_expansion.py @@ -1,6 +1,10 @@ from abc import ABC, abstractmethod -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask + + +class ABCInheritedTaskWithNoInheritors(CoSyLuigiTask, ABC): + pass class ABCInheritedTask(CoSyLuigiTask, ABC): @@ -19,7 +23,14 @@ class DeeperConcreteTaskFromABCInherited(ConcreteTaskFromABCInherited): class AbstractTask(CoSyLuigiTask): @abstractmethod def get_class_name(self): - pass + raise NotImplementedError + + +# noinspection PyAbstractClass +class AbstractTaskWithNoInheritors(CoSyLuigiTask): + @abstractmethod + def get_class_name(self): + raise NotImplementedError class ConcreteTaskFromAbstract(AbstractTask): @@ -32,13 +43,19 @@ def get_class_name(self): return "DeeperConcreteTaskFromAbstract" -def test_abstract_variant_expansion(): +def test_expansion_from_abc(): repo = CoSyLuigiRepo(ABCInheritedTask) assert repo.luigi_repo == {ConcreteTaskFromABCInherited, DeeperConcreteTaskFromABCInherited} + + +def test_expansion_from_abstract(): repo = CoSyLuigiRepo(AbstractTask) assert ConcreteTaskFromAbstract().get_class_name() == "ConcreteTaskFromAbstract" assert DeeperConcreteTaskFromAbstract().get_class_name() == "DeeperConcreteTaskFromAbstract" assert repo.luigi_repo == {ConcreteTaskFromAbstract, DeeperConcreteTaskFromAbstract} + + +def test_expansion_from_abc_and_abstract(): repo = CoSyLuigiRepo(ABCInheritedTask, AbstractTask) assert repo.luigi_repo == { ConcreteTaskFromABCInherited, @@ -46,5 +63,23 @@ def test_abstract_variant_expansion(): ConcreteTaskFromAbstract, DeeperConcreteTaskFromAbstract, } + + +def test_implementation_of_abstract_does_not_expand(): repo = CoSyLuigiRepo(ConcreteTaskFromAbstract) assert repo.luigi_repo == {ConcreteTaskFromAbstract} + + +def test_implementation_of_abc_does_not_expand(): + repo = CoSyLuigiRepo(ConcreteTaskFromABCInherited) + assert repo.luigi_repo == {ConcreteTaskFromABCInherited} + + +def test_expansion_to_nothing_from_abc_with_no_inheritors(): + repo = CoSyLuigiRepo(ABCInheritedTaskWithNoInheritors) + assert repo.luigi_repo == set() + + +def test_expansion_to_nothing_from_abstract_with_no_inheritors(): + repo = CoSyLuigiRepo(AbstractTaskWithNoInheritors) + assert repo.luigi_repo == set() diff --git a/tests/test_heterogeneous_repo_input.py b/tests/test_heterogeneous_repo_input.py index 204b292..6ffa6dd 100644 --- a/tests/test_heterogeneous_repo_input.py +++ b/tests/test_heterogeneous_repo_input.py @@ -1,4 +1,5 @@ -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask +from cosy_luigi.utils import flatten class TaskA(CoSyLuigiTask): @@ -48,3 +49,24 @@ class TaskK(CoSyLuigiTask): def test_heterogeneous_repo_input(): repo = CoSyLuigiRepo(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])) assert repo.luigi_repo == {TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK} + + +def test_heterogeneous_args_input(): + flattened_collection = set( + flatten(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])) + ) + assert flattened_collection == {TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK} + + +def test_heterogeneous_list_input(): + flattened_collection = list( + flatten([TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])]) + ) + assert flattened_collection == [TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK] + + +def test_heterogeneous_tuple_input(): + flattened_collection = tuple( + flatten((TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK]))) + ) + assert flattened_collection == (TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK) diff --git a/tests/test_infinite_chain.py b/tests/test_infinite_chain.py index 1d85513..7e3bc88 100644 --- a/tests/test_infinite_chain.py +++ b/tests/test_infinite_chain.py @@ -2,20 +2,13 @@ import pytest from cosy.maestro import Maestro -from luigi.mock import MockTarget -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter class ChainLink(CoSyLuigiTask, ABC): chain_link: CoSyLuigiTaskParameter | None - def output(self): - return MockTarget("ChainLink") - - def run(self): - self.output().open("w").write("Ok.") - class StartingLink(ChainLink): chain_link = None @@ -28,12 +21,6 @@ class RepeatingLink(ChainLink): class FinalLink(CoSyLuigiTask): chain_link = CoSyLuigiTaskParameter(ChainLink) - def output(self): - return MockTarget("FinalLink") - - def run(self): - self.output().open("w").write("Ok.") - @pytest.fixture def repo(): diff --git a/tests/test_unique_task_parameter.py b/tests/test_unique_task_parameter.py new file mode 100644 index 0000000..8cc556f --- /dev/null +++ b/tests/test_unique_task_parameter.py @@ -0,0 +1,146 @@ +import logging +from abc import ABC +from collections.abc import Callable, Sequence + +import pytest +from cosy.maestro import Maestro + +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi.constraints import is_unique_in_prior_tasks + + +class ScaleDataABC(CoSyLuigiTask, ABC): + pass + + +class ScaleData(ScaleDataABC): + pass + + +class ScaleDataVariantA(ScaleData): + pass + + +class ScaleDataVariantB(ScaleData): + pass + + +class TrainModel(CoSyLuigiTask, ABC): + scaled_data = CoSyLuigiTaskParameter(ScaleDataABC) + + +class TrainModelVariantA(TrainModel): + pass + + +class TrainModelVariantB(TrainModel): + pass + + +class EvaluatePipelineWithUniqueScaler(CoSyLuigiTask): + train_model = CoSyLuigiTaskParameter(TrainModel) + scaled_data = CoSyLuigiTaskParameter(ScaleDataABC, unique_across_prior_tasks=True) + + +class EvaluatePipelineWithConstraintUniqueScaler(CoSyLuigiTask): + train_model = CoSyLuigiTaskParameter(TrainModel) + scaled_data = CoSyLuigiTaskParameter(ScaleDataABC) + + @classmethod + def constraints(cls) -> Sequence[Callable[..., bool]]: + return [lambda vs: is_unique_in_prior_tasks(vs, ScaleDataABC)] + + +class EvaluatePipelineWithUniqueScalerAndNonAbstractSuper(CoSyLuigiTask): + train_model = CoSyLuigiTaskParameter(TrainModel) + scaled_data = CoSyLuigiTaskParameter(ScaleData, unique_across_prior_tasks=True) + + +class EvaluatePipeline(CoSyLuigiTask): + train_model = CoSyLuigiTaskParameter(TrainModel) + scaled_data = CoSyLuigiTaskParameter(ScaleDataABC) + + +@pytest.fixture +def repo_without_constraints(): + return CoSyLuigiRepo(TrainModel, ScaleDataABC, EvaluatePipeline) + + +@pytest.fixture +def repo_with_constraints(): + return CoSyLuigiRepo(TrainModel, ScaleDataABC, EvaluatePipelineWithUniqueScaler) + + +@pytest.fixture +def repo_with_manual_constraints(): + return CoSyLuigiRepo(TrainModel, ScaleDataABC, EvaluatePipelineWithConstraintUniqueScaler) + + +@pytest.fixture +def repo_with_non_abstract_super(): + return CoSyLuigiRepo( + TrainModel, ScaleData, ScaleDataVariantA, ScaleDataVariantB, EvaluatePipelineWithUniqueScalerAndNonAbstractSuper + ) + + +def test_implementation_is_not_unique_across_prior_tasks(repo_without_constraints): + maestro = Maestro( + repo_without_constraints.cls_repo, + repo_without_constraints.taxonomy, + ) + results = list(maestro.query(EvaluatePipeline.target())) + assert len(results) == 18 + + +def test_implementation_is_unique_across_prior_tasks(repo_with_constraints): + maestro = Maestro( + repo_with_constraints.cls_repo, + repo_with_constraints.taxonomy, + ) + results: list[EvaluatePipelineWithUniqueScaler] = list(maestro.query(EvaluatePipelineWithUniqueScaler.target())) + assert len(results) == 6 + for result in results: + assert result.scaled_data == result.train_model.scaled_data + + +def test_implementation_is_unique_across_prior_tasks_with_manual_constraint(repo_with_manual_constraints): + maestro = Maestro( + repo_with_manual_constraints.cls_repo, + repo_with_manual_constraints.taxonomy, + ) + results: list[EvaluatePipelineWithConstraintUniqueScaler] = list( + maestro.query(EvaluatePipelineWithConstraintUniqueScaler.target()) + ) + assert len(results) == 6 + for result in results: + assert result.scaled_data == result.train_model.scaled_data + + +def test_implementation_is_unique_across_prior_tasks_with_non_abstract_super(repo_with_non_abstract_super): + maestro = Maestro( + repo_with_non_abstract_super.cls_repo, + repo_with_non_abstract_super.taxonomy, + ) + results: list[EvaluatePipelineWithUniqueScalerAndNonAbstractSuper] = list( + maestro.query(EvaluatePipelineWithUniqueScalerAndNonAbstractSuper.target()) + ) + assert len(results) == 6 + for result in results: + assert result.scaled_data == result.train_model.scaled_data + + +def test_warning_if_unique_across_prior_tasks_but_no_variance(caplog): + caplog.set_level(logging.WARNING) + repo_with_constraints_and_no_variance = CoSyLuigiRepo( + TrainModel, ScaleData, EvaluatePipelineWithUniqueScalerAndNonAbstractSuper + ) + assert len(caplog.records) == 1 + assert caplog.records[0].levelname == "WARNING" + assert "ScaleData" in caplog.records[0].message + + maestro = Maestro( + repo_with_constraints_and_no_variance.cls_repo, + repo_with_constraints_and_no_variance.taxonomy, + ) + results = list(maestro.query(EvaluatePipelineWithUniqueScalerAndNonAbstractSuper.target())) + assert len(results) == 2