diff --git a/modules/variant_builder/context.py b/modules/variant_builder/context.py new file mode 100644 index 0000000..3f7c925 --- /dev/null +++ b/modules/variant_builder/context.py @@ -0,0 +1,15 @@ +from dataclasses import dataclass +from typing import Generic + +from modules.utils.types import A, C, Q, V +from modules.variant import QVariant +from modules.variant_builder.task_pool import QTaskPool +from modules.variant_builder.variant_set import QVariantSet + + +@dataclass +class DynamicFilterCtx(Generic[C, V, Q, A]): + task_pool: QTaskPool[C, V, Q, A] + previous_variants: QVariantSet[C, V, Q, A] + current_variant: QVariant[C, V, Q, A] + task_number: int diff --git a/modules/variant_builder/filters/__init__.py b/modules/variant_builder/filters/__init__.py index 6d6744f..ae6d6a5 100644 --- a/modules/variant_builder/filters/__init__.py +++ b/modules/variant_builder/filters/__init__.py @@ -1,121 +1,28 @@ -from collections.abc import Iterable -from dataclasses import dataclass, field -from typing import Callable, Generic, Self, overload, override +from dataclasses import dataclass +from typing import Generic -from modules.tag import Tag, Tags from modules.utils.types import A, C, Q, V -from modules.variant_builder.filters.composite import CompositeFilter from modules.variant_builder.filters.dynamic import FilterDynamic from modules.variant_builder.filters.dynamic.composite import CompositeFilterDynamic from modules.variant_builder.filters.static import FilterStatic from modules.variant_builder.filters.static.composite import CompositeFilterStatic -from modules.variant_builder.filters.static.must_be_one_of import MustBeOneOfFilter -from modules.variant_builder.filters.static.must_include_all_tags import ( - MustIncludeAllTagsFilter, -) -from modules.variant_builder.filters.static.must_include_any_tag import ( - MustIncludeAnyTagFilter, -) -from modules.variant_builder.filters.types import QTaskOrFactory -@dataclass -class FilterBuilder(Generic[C, V, Q, A]): - static_filters: list[FilterStatic[C, V, Q, A]] = field(default_factory=list) - dynamic_filters: list[FilterDynamic[C, V, Q, A]] = field(default_factory=list) +class Filter(Generic[C, V, Q, A]): + static: FilterStatic[C, V, Q, A] + dynamic: FilterDynamic[C, V, Q, A] - def add_static(self, filter: FilterStatic[C, V, Q, A]) -> Self: - self.static_filters.append(filter) - return self - - def add_dynamic(self, filter: FilterDynamic[C, V, Q, A]) -> Self: - self.dynamic_filters.append(filter) - return self - - def add(self, filter: FilterStatic[C, V, Q, A] | FilterDynamic[C, V, Q, A]) -> Self: - if isinstance(filter, FilterStatic): - self.static_filters.append(filter) + def __init__( + self, + static: CompositeFilterStatic[C, V, Q, A] | list[FilterStatic[C, V, Q, A]], + dynamic: CompositeFilterDynamic[C, V, Q, A] | list[FilterDynamic[C, V, Q, A]], + ): + if isinstance(static, list): + self.static = CompositeFilterStatic(static) else: - self.dynamic_filters.append(filter) - return self + self.static = static - def __call__( - self, filter: FilterStatic[C, V, Q, A] | FilterDynamic[C, V, Q, A] - ) -> Self: - return self.add(filter) - - @overload - def be_one_of(self, item: Iterable[QTaskOrFactory[C, V, Q, A]]) -> Self: ... - - @overload - def be_one_of( - self, item: QTaskOrFactory[C, V, Q, A], **kwargs: QTaskOrFactory[C, V, Q, A] - ) -> Self: ... - - def be_one_of( - self, - item: Iterable[QTaskOrFactory[C, V, Q, A]] | QTaskOrFactory[C, V, Q, A], - **kwargs: QTaskOrFactory[C, V, Q, A], - ) -> Self: - return self.add_static(MustBeOneOfFilter(item, **kwargs)) - - @overload - def include_any_tag(self, tag: Tags[C, V]) -> Self: ... - - @overload - def include_any_tag(self, tag: Iterable[Tag[C, V]]) -> Self: ... - - @overload - def include_any_tag(self, tag: Tag[C, V], **kwargs: Tag[C, V]) -> Self: ... - - def include_any_tag( - self, - tag: Tags[C, V] | Tag[C, V] | Iterable[Tag[C, V]], - **kwargs: Tag[C, V], - ) -> Self: - return self.add_static(MustIncludeAnyTagFilter(tag, **kwargs)) - - @overload - def include_all_tags(self, tag: Tags[C, V]) -> Self: ... - - @overload - def include_all_tags(self, tag: Iterable[Tag[C, V]]) -> Self: ... - - @overload - def include_all_tags(self, tag: Tag[C, V], **kwargs: Tag[C, V]) -> Self: ... - - def include_all_tags( - self, - tag: Tags[C, V] | Tag[C, V] | Iterable[Tag[C, V]], - **kwargs: Tag[C, V], - ) -> Self: - return self.add_static(MustIncludeAllTagsFilter(tag, **kwargs)) - - def be_inverse_of( - self, - filter: Callable[["FilterBuilder[C, V, Q, A]"], "FilterBuilder[C, V, Q, A]"], - ) -> "FilterBuilder[C, V, Q, A]": - return filter(FilterBuilder[C, V, Q, A]()).invert() - - def build_static( - self, - ) -> FilterStatic[C, V, Q, A]: - return CompositeFilterStatic(self.static_filters) - - def build_dynamic( - self, - ) -> CompositeFilterDynamic[C, V, Q, A]: - return CompositeFilterDynamic(self.dynamic_filters) - - def build(self) -> CompositeFilter[C, V, Q, A]: - return CompositeFilter(self.build_static(), self.build_dynamic()) - - def invert(self) -> Self: - for i in range(len(self.static_filters)): - self.static_filters[i] = ~self.static_filters[i] - for i in range(len(self.dynamic_filters)): - self.dynamic_filters[i] = ~self.dynamic_filters[i] - return self - - def __invert__(self) -> Self: - return self.invert() + if isinstance(dynamic, list): + self.dynamic = CompositeFilterDynamic(dynamic) + else: + self.dynamic = dynamic diff --git a/modules/variant_builder/filters/builder.py b/modules/variant_builder/filters/builder.py new file mode 100644 index 0000000..afd70d7 --- /dev/null +++ b/modules/variant_builder/filters/builder.py @@ -0,0 +1,112 @@ +from collections.abc import Iterable +from dataclasses import dataclass, field +from typing import Callable, Generic, Self, overload, override + +from modules.tag import Tag, Tags +from modules.utils.types import A, C, Q, V +from modules.variant_builder.filters import Filter +from modules.variant_builder.filters.composite import CompositeFilter +from modules.variant_builder.filters.dynamic import FilterDynamic +from modules.variant_builder.filters.dynamic.composite import CompositeFilterDynamic +from modules.variant_builder.filters.static import FilterStatic +from modules.variant_builder.filters.static.composite import CompositeFilterStatic +from modules.variant_builder.filters.static.must_be_one_of import MustBeOneOfFilter +from modules.variant_builder.filters.static.must_include_all_tags import ( + MustIncludeAllTagsFilter, +) +from modules.variant_builder.filters.static.must_include_any_tag import ( + MustIncludeAnyTagFilter, +) +from modules.variant_builder.filters.types import QTaskOrFactory + + +@dataclass +class FilterBuilder(Generic[C, V, Q, A]): + static_filters: list[FilterStatic[C, V, Q, A]] = field(default_factory=list) + dynamic_filters: list[FilterDynamic[C, V, Q, A]] = field(default_factory=list) + + def add_static(self, filter: FilterStatic[C, V, Q, A]) -> Self: + self.static_filters.append(filter) + return self + + def add_dynamic(self, filter: FilterDynamic[C, V, Q, A]) -> Self: + self.dynamic_filters.append(filter) + return self + + def add(self, filter: FilterStatic[C, V, Q, A] | FilterDynamic[C, V, Q, A]) -> Self: + if isinstance(filter, FilterStatic): + self.static_filters.append(filter) + else: + self.dynamic_filters.append(filter) + return self + + def __call__( + self, filter: FilterStatic[C, V, Q, A] | FilterDynamic[C, V, Q, A] + ) -> Self: + return self.add(filter) + + @overload + def be_one_of(self, item: Iterable[QTaskOrFactory[C, V, Q, A]]) -> Self: ... + + @overload + def be_one_of( + self, item: QTaskOrFactory[C, V, Q, A], **kwargs: QTaskOrFactory[C, V, Q, A] + ) -> Self: ... + + def be_one_of( + self, + item: Iterable[QTaskOrFactory[C, V, Q, A]] | QTaskOrFactory[C, V, Q, A], + **kwargs: QTaskOrFactory[C, V, Q, A], + ) -> Self: + return self.add_static(MustBeOneOfFilter(item, **kwargs)) + + @overload + def include_any_tag(self, tag: Tags[C, V]) -> Self: ... + + @overload + def include_any_tag(self, tag: Iterable[Tag[C, V]]) -> Self: ... + + @overload + def include_any_tag(self, tag: Tag[C, V], **kwargs: Tag[C, V]) -> Self: ... + + def include_any_tag( + self, + tag: Tags[C, V] | Tag[C, V] | Iterable[Tag[C, V]], + **kwargs: Tag[C, V], + ) -> Self: + return self.add_static(MustIncludeAnyTagFilter(tag, **kwargs)) + + @overload + def include_all_tags(self, tag: Tags[C, V]) -> Self: ... + + @overload + def include_all_tags(self, tag: Iterable[Tag[C, V]]) -> Self: ... + + @overload + def include_all_tags(self, tag: Tag[C, V], **kwargs: Tag[C, V]) -> Self: ... + + def include_all_tags( + self, + tag: Tags[C, V] | Tag[C, V] | Iterable[Tag[C, V]], + **kwargs: Tag[C, V], + ) -> Self: + return self.add_static(MustIncludeAllTagsFilter(tag, **kwargs)) + + def be_inverse_of( + self, + filter: Callable[["FilterBuilder[C, V, Q, A]"], "FilterBuilder[C, V, Q, A]"], + ) -> "FilterBuilder[C, V, Q, A]": + return filter(FilterBuilder[C, V, Q, A]()).invert() + + def build(self) -> Filter[C, V, Q, A]: + return Filter(self.static_filters, self.dynamic_filters) + + def invert(self) -> Self: + for i in range(len(self.static_filters)): + self.static_filters[i] = ~self.static_filters[i] + for i in range(len(self.dynamic_filters)): + self.dynamic_filters[i] = ~self.dynamic_filters[i] + return self + + def __invert__(self) -> Self: + return self.invert() diff --git a/modules/variant_builder/filters/composite.py b/modules/variant_builder/filters/composite.py deleted file mode 100644 index 1f02d7e..0000000 --- a/modules/variant_builder/filters/composite.py +++ /dev/null @@ -1,33 +0,0 @@ -from dataclasses import dataclass -from typing import Generic, Self - -from modules.utils.types import A, C, Q, V -from modules.variant_builder.filters.dynamic import FilterDynamic -from modules.variant_builder.filters.dynamic.composite import CompositeFilterDynamic -from modules.variant_builder.filters.static import FilterStatic -from modules.variant_builder.filters.static.composite import CompositeFilterStatic - - -class CompositeFilter(Generic[C, V, Q, A]): - static: CompositeFilterStatic[C, V, Q, A] - dynamic: CompositeFilterDynamic[C, V, Q, A] - - def __init__( - self, - static: CompositeFilterStatic[C, V, Q, A] | list[FilterStatic[C, V, Q, A]], - dynamic: CompositeFilterDynamic[C, V, Q, A] | list[FilterDynamic[C, V, Q, A]], - ): - if isinstance(static, list): - self.static = CompositeFilterStatic(static) - else: - self.static = static - - if isinstance(dynamic, list): - self.dynamic = CompositeFilterDynamic(dynamic) - else: - self.dynamic = dynamic - - def invert(self) -> Self: - self.static = ~self.static - self.dynamic = ~self.dynamic - return self diff --git a/modules/variant_builder/filters/dynamic/__init__.py b/modules/variant_builder/filters/dynamic/__init__.py index bc281c3..dedf8c4 100644 --- a/modules/variant_builder/filters/dynamic/__init__.py +++ b/modules/variant_builder/filters/dynamic/__init__.py @@ -5,6 +5,7 @@ from typing import Generic, Protocol, override, runtime_checkable from modules.task import QTask from modules.utils.types import A, C, Q, V from modules.variant import QVariant +from modules.variant_builder.context import DynamicFilterCtx from modules.variant_builder.task_pool import QTaskPool from modules.variant_builder.variant_set import QVariantSet @@ -13,12 +14,7 @@ from modules.variant_builder.variant_set import QVariantSet class FilterDynamic(ABC, Generic[C, V, Q, A]): @abstractmethod def check_if_satisfied( - self, - task: QTask[C, V, Q, A], - task_pool: QTaskPool[C, V, Q, A], - previous_variants: QVariantSet[C, V, Q, A], - current_variant: QVariant[C, V, Q, A], - task_number: int, + self, task: QTask[C, V, Q, A], ctx: DynamicFilterCtx[C, V, Q, A] ) -> bool: ... def __invert__(self) -> "FilterDynamic[C, V, Q, A]": @@ -37,16 +33,9 @@ class DynamicFilterNegator(FilterDynamic[C, V, Q, A]): @override def check_if_satisfied( - self, - task: QTask[C, V, Q, A], - task_pool: QTaskPool[C, V, Q, A], - previous_variants: QVariantSet[C, V, Q, A], - current_variant: QVariant[C, V, Q, A], - task_number: int, + self, task: QTask[C, V, Q, A], ctx: DynamicFilterCtx[C, V, Q, A] ) -> bool: - return not self.filter.check_if_satisfied( - task, task_pool, previous_variants, current_variant, task_number - ) + return not self.filter.check_if_satisfied(task, ctx) @override def __invert__(self) -> "FilterDynamic[C, V, Q, A]": diff --git a/modules/variant_builder/filters/dynamic/composite.py b/modules/variant_builder/filters/dynamic/composite.py index 9e84833..8c7213c 100644 --- a/modules/variant_builder/filters/dynamic/composite.py +++ b/modules/variant_builder/filters/dynamic/composite.py @@ -1,30 +1,27 @@ -from dataclasses import dataclass +from dataclasses import dataclass, field, replace from typing import override from modules.task import QTask from modules.utils.types import A, C, Q, V -from modules.variant import QVariant +from modules.variant_builder.context import DynamicFilterCtx from modules.variant_builder.filters.dynamic import FilterDynamic -from modules.variant_builder.task_pool import QTaskPool -from modules.variant_builder.variant_set import QVariantSet @dataclass class CompositeFilterDynamic(FilterDynamic[C, V, Q, A]): filters: list[FilterDynamic[C, V, Q, A]] + is_inverted: bool = field(default=False) @override def check_if_satisfied( - self, - task: QTask[C, V, Q, A], - task_pool: QTaskPool[C, V, Q, A], - previous_variants: QVariantSet[C, V, Q, A], - current_variant: QVariant[C, V, Q, A], - task_number: int, + self, task: QTask[C, V, Q, A], ctx: DynamicFilterCtx[C, V, Q, A] ) -> bool: - return all( - filter.check_if_satisfied( - task, task_pool, previous_variants, current_variant, task_number - ) - for filter in self.filters - ) + return all(filter.check_if_satisfied(task, ctx) for filter in self.filters) + + @override + def invert(self) -> "CompositeFilterDynamic[C, V, Q, A]": + return replace(self, is_inverted=not self.is_inverted) + + @override + def __invert__(self) -> "CompositeFilterDynamic[C, V, Q, A]": + return self.invert() diff --git a/modules/variant_builder/filters/static/composite.py b/modules/variant_builder/filters/static/composite.py index 5326b37..8545fca 100644 --- a/modules/variant_builder/filters/static/composite.py +++ b/modules/variant_builder/filters/static/composite.py @@ -22,3 +22,7 @@ class CompositeFilterStatic(FilterStatic[C, V, Q, A]): @override def invert(self) -> "CompositeFilterStatic[C, V, Q, A]": return replace(self, is_inverted=not self.is_inverted) + + @override + def __invert__(self) -> "CompositeFilterStatic[C, V, Q, A]": + return self.invert() diff --git a/modules/variant_builder/variant_task.py b/modules/variant_builder/variant_task.py new file mode 100644 index 0000000..71046f5 --- /dev/null +++ b/modules/variant_builder/variant_task.py @@ -0,0 +1,6 @@ +from typing import Generic + +from modules.utils.types import A, C, Q, V + + +class VariantTask(Generic[C, V, Q, A]):