Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 14 additions & 12 deletions benchmarks/test_benchmark_chain_100.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,46 +13,48 @@


class ChainLink(CoSyLuigiTask, ABC):
"""_summary_."""
"""An abstract class representing a chain link in an infinite chain."""

chain_link: CoSyLuigiTaskParameter | None


class StartingLink(ChainLink):
"""_summary_."""
"""A class that terminates the chain by needing no further chain links."""

chain_link = None


class RepeatingLink(ChainLink):
"""_summary_."""
"""A class that recurses the chain by requiring a further chain link."""

chain_link = CoSyLuigiTaskParameter(ChainLink)

def output(self):
"""_summary_.
"""Assign each chain link a unique identifier. This is required because CoSy-Luigi considers tasks with
identical names, identical requirements, and identical outputs to be Singletons. This differs from Luigi,
which considers tasks with identical names and identical requirements Singletons.

Returns:
_type_: _description_
Mapping[str, MockTarget]_: The named unique target for each chain link.
"""
return {"counter": MockTarget(str(next(counter)))}


@pytest.fixture
def repo():
"""_summary_.
"""Creates a CoSyLuigiRepo that contains the StartingLink and the RepeatingLink.

Returns:
_type_: _description_
CoSyLuigiRepo: The created CoSyLuigiRepo.
"""
return CoSyLuigiRepo(ChainLink)


def create_infinite_chain(repo):
"""_summary_.
"""Synthesizes all pipelines up to those that carry out the same step 100 times.

Args:
repo (_type_): _description_
repo (CoSyLuigiRepo): The repository to use for synthesis.
"""
maestro = Maestro(
repo.cls_repo,
Expand All @@ -62,11 +64,11 @@ def create_infinite_chain(repo):


def test_benchmark_chain_creation(repo, benchmark):
"""_summary_.
"""Benchmarks how long synthesizing and enumerating the pipelines takes.

Args:
repo (_type_): _description_
benchmark (_type_): _description_
repo (CoSyLuigiRepo): The repository to use for synthesis.
benchmark (BenchmarkFixture): The benchmark fixture.
"""
benchmark(create_infinite_chain, repo)

Expand Down
2 changes: 1 addition & 1 deletion src/cosy_luigi/constraints/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""_summary_."""
"""This module contains common constraints that may be needed when modeling pipelines."""

from cosy_luigi.constraints.unique import is_unique_in_prior_tasks

Expand Down
26 changes: 17 additions & 9 deletions src/cosy_luigi/constraints/unique.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""_summary_."""
"""Contains the function for the constraint that makes a required task be unique throughout all prior tasks in a
pipeline."""

from __future__ import annotations

Expand All @@ -16,14 +17,20 @@
def _is_unique_in_prior_tasks(
vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: Sequence[type[CoSyLuigiTask]]
) -> bool:
"""_summary_.
"""Examines the output of traverse_pipeline against a Sequence of CoSyLuigiTask's types that are intended to be
unique. Whenever a task that is a subclass or the required to be unique class itself is encountered in the
pipeline, remember the encountered task. If any further task that is a subclass that is not identical to the
previously encountered task is encountered, return False, else return True.

Encountering two different subclasses within the same pipeline for given required to be unique tasks means that
it is not unique.

Args:
vs (Mapping[str, CoSyLuigiTask]): _description_
required_to_be_unique (Sequence[type[CoSyLuigiTask]]): _description_
vs (Mapping[str, CoSyLuigiTask]): The variables passed to the function by CoSy during synthesis. Populated by the partial pipelines beginning at current tasks required tasks.
required_to_be_unique (Sequence[type[CoSyLuigiTask]]): The CoSyLuigiTasks' types that are intended to be unique.

Returns:
bool: _description_
bool: True if all types contained in required_to_be_unique are unique, False otherwise.
"""
classes = [pc.__class__ for pc in traverse_pipeline(vs.values())]
seen_subclasses: dict[type[CoSyLuigiTask], type[CoSyLuigiTask]] = {}
Expand All @@ -41,14 +48,15 @@ def _is_unique_in_prior_tasks(
def is_unique_in_prior_tasks(
vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: type[CoSyLuigiTask] | Sequence[type[CoSyLuigiTask]]
) -> bool:
"""_summary_.
"""Wrapper around _is_unique_in_prior_tasks that allows passing either a single type of a CoSyLuigiTask or a
Sequence of CoSyLuigiTasks' types.

Args:
vs (Mapping[str, CoSyLuigiTask]): _description_
required_to_be_unique (type[CoSyLuigiTask] | Sequence[type[CoSyLuigiTask]]): _description_
vs (Mapping[str, CoSyLuigiTask]): The variables passed to the function by CoSy during synthesis. Populated by the partial pipelines beginning at current tasks required tasks.
required_to_be_unique (type[CoSyLuigiTask] | Sequence[type[CoSyLuigiTask]]): The CoSyLuigiTask's type or types that are intended to be unique.

Returns:
bool: _description_
bool: True if all types contained in required_to_be_unique are unique, False otherwise.
"""
return _is_unique_in_prior_tasks(
vs,
Expand Down
Loading
Loading