reworked structure a lot
This commit is contained in:
15
modules/variant_builder/context.py
Normal file
15
modules/variant_builder/context.py
Normal file
@ -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
|
||||
@ -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
|
||||
|
||||
112
modules/variant_builder/filters/builder.py
Normal file
112
modules/variant_builder/filters/builder.py
Normal file
@ -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()
|
||||
@ -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
|
||||
@ -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]":
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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()
|
||||
|
||||
6
modules/variant_builder/variant_task.py
Normal file
6
modules/variant_builder/variant_task.py
Normal file
@ -0,0 +1,6 @@
|
||||
from typing import Generic
|
||||
|
||||
from modules.utils.types import A, C, Q, V
|
||||
|
||||
|
||||
class VariantTask(Generic[C, V, Q, A]):
|
||||
Reference in New Issue
Block a user