working on stuff (with great commit messages)

This commit is contained in:
2025-04-18 14:02:48 +03:00
parent 3cc871c3ff
commit 2a45cfe45e
17 changed files with 429 additions and 24 deletions

View File

@ -0,0 +1,106 @@
from dataclasses import dataclass, field
from typing import Generic, Iterable, List, Protocol, Union, cast, overload, override
from modules.constrains.static import VTaskConstraintStatic
from modules.constrains.static.must_be_any import MustBeAnyConstraint
from modules.constrains.static.must_include_all_tags import MustIncludeAllTagsConstraint
from modules.constrains.static.must_include_any_tag import MustIncludeAnyTagConstraint
from modules.constrains.static.must_not import MustNotStatic
from modules.constrains.types import QTaskOrFactory
from modules.tags import Tag, Tags
from utils.types import A, C, Q, V
@dataclass
class VTaskConstraintsAbstract(Generic[C, V, Q, A]):
static_constraints: List[VTaskConstraintStatic[C, V, Q, A]] = field(
default_factory=list
)
dynamic_constraints: None = None
def add_constraint(
self, constraint: VTaskConstraintStatic[C, V, Q, A]
) -> "VTaskConstraints[C, V, Q, A]":
self.static_constraints.append(constraint)
# unsafe, but needed to avoid typing the same thing for two times...
return cast("VTaskConstraints[C, V, Q, A]", self)
def __call__(
self, constraint: VTaskConstraintStatic[C, V, Q, A]
) -> "VTaskConstraints[C, V, Q, A]":
return self.add_constraint(constraint)
@overload
def be_any(
self, item: Iterable[QTaskOrFactory[C, V, Q, A]]
) -> "VTaskConstraints[C, V, Q, A]": ...
@overload
def be_any(
self, item: QTaskOrFactory[C, V, Q, A], **kwargs: QTaskOrFactory[C, V, Q, A]
) -> "VTaskConstraints[C, V, Q, A]": ...
def be_any(
self,
item: Union[Iterable[QTaskOrFactory[C, V, Q, A]], QTaskOrFactory[C, V, Q, A]],
**kwargs: QTaskOrFabric[C, V, Q, A],
) -> "VTaskConstraints[C, V, Q, A]":
return self.add_constraint(MustBeAnyConstraint(item, **kwargs))
@overload
def include_any_tag(self, tag: Tags[C, V]) -> "VTaskConstraints[C, V, Q, A]": ...
@overload
def include_any_tag(
self, tag: Iterable[Tag[C, V]]
) -> "VTaskConstraints[C, V, Q, A]": ...
@overload
def include_any_tag(
self, tag: Tag[C, V], **kwargs: Tag[C, V]
) -> "VTaskConstraints[C, V, Q, A]": ...
def include_any_tag(
self,
tag: Union[Tags[C, V], Tag[C, V], Iterable[Tag[C, V]]],
**kwargs: Tag[C, V],
) -> "VTaskConstraints[C, V, Q, A]":
return self.add_constraint(MustIncludeAnyTagConstraint(tag, **kwargs))
@overload
def include_all_tags(self, tag: Tags[C, V]) -> "VTaskConstraints[C, V, Q, A]": ...
@overload
def include_all_tags(
self, tag: Iterable[Tag[C, V]]
) -> "VTaskConstraints[C, V, Q, A]": ...
@overload
def include_all_tags(
self, tag: Tag[C, V], **kwargs: Tag[C, V]
) -> "VTaskConstraints[C, V, Q, A]": ...
def include_all_tags(
self,
tag: Union[Tags[C, V], Tag[C, V], Iterable[Tag[C, V]]],
**kwargs: Tag[C, V],
):
return self.add_constraint(MustIncludeAllTagsConstraint(tag, **kwargs))
class VTaskConstraintsNegator(VTaskConstraintsAbstract[C, V, Q, A]):
def __init__(self, v_task_constraints: "VTaskConstraints[C, V, Q, A]"):
self.v_task_constraints = v_task_constraints
def add_constraint(
self, constraint: VTaskConstraintStatic[C, V, Q, A]
) -> "VTaskConstraints[C, V, Q, A]":
return self.v_task_constraints.add_constraint(MustNotStatic(constraint))
@dataclass
class VTaskConstraints(VTaskConstraintsAbstract[C, V, Q, A]):
nt: VTaskConstraintsNegator[C, V, Q, A] = field(init=False)
def __post_init__(self):
self.nt = VTaskConstraintsNegator(self)

View File

@ -0,0 +1,21 @@
from typing import Protocol, runtime_checkable
from modules.task import QTask
from modules.task_pool import QTaskPool
from modules.variant import QVariant
from modules.variant_set import QVariantSet
from utils.types import A, C, Q, V
@runtime_checkable
class VTaskConstraintDynamic(Protocol[C, V, Q, A]):
def _dyn(self):
return None
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],
) -> bool: ...

View File

@ -0,0 +1,13 @@
from typing import Protocol, runtime_checkable
from modules.task import QTask
from utils.types import A, C, Q, V
@runtime_checkable
class VTaskConstraintStatic(Protocol[C, V, Q, A]):
# dull func to distinct dynamic and static types
def _sta(self):
return None
def is_satisfied(self, task: QTask[C, V, Q, A]) -> bool: ...

View File

@ -0,0 +1,42 @@
from typing import Iterable, List, Union, overload
from modules.constrains.static import VTaskStaticConstraint
from modules.constrains.types import QTaskOrFabric
from modules.fabric import QTaskFabric
from modules.task import QTask
from utils.types import A, C, Q, V
class MustBeAnyConstraint(VTaskStaticConstraint[C, V, Q, A]):
must_be_generated_by: List[QTaskFabric[C, V, Q, A]] = []
must_be_one_of_tasks: List[QTask[C, V, Q, A]] = []
@overload
def __init__(self, item: Iterable[QTaskOrFabric[C, V, Q, A]]): ...
@overload
def __init__(
self, item: QTaskOrFabric[C, V, Q, A], **kwargs: QTaskOrFabric[C, V, Q, A]
): ...
def __init__(
self,
item: Union[Iterable[QTaskOrFabric[C, V, Q, A]], QTaskOrFabric[C, V, Q, A]],
**kwargs: QTaskOrFabric[C, V, Q, A],
):
all_items = []
if isinstance(item, List):
all_items.extend(item)
else:
all_items.append(item)
all_items.extend(kwargs.values())
self.must_be_generated_by = [v for v in all_items if isinstance(v, QTaskFabric)]
self.must_be_one_of_tasks = [v for v in all_items if isinstance(v, QTask)]
def is_satisfied(self, task: QTask[C, V, Q, A]) -> bool:
return any(
[
task.fabric_metadata.unwrap_or(None) == g.metadata.id
for g in self.must_be_generated_by
]
) or any([task.id == t.id for t in self.must_be_one_of_tasks])

View File

@ -0,0 +1,37 @@
from typing import Iterable, Tuple, Union, overload
from modules.constrains.static import VTaskConstraintStatic
from modules.tags import Tag, Tags
from modules.task import QTask
from utils.types import A, C, Q, V
class MustIncludeAllTagsConstraint(VTaskConstraintStatic[C, V, Q, A]):
tags: Tags[C, V] = Tags()
@overload
def __init__(self, tag: Tags[C, V]): ...
@overload
def __init__(self, tag: Iterable[Tag[C, V]])
@overload
def __init__(self, tag: Tag[C, V], **kwargs: Tag[C, V]): ...
def __init__(
self,
tag: Union[Tags[C, V], Tag[C, V], Iterable[Tag[C, V]]],
**kwargs: Tag[C, V],
):
if isinstance(tag, Tags):
self.tags = tag
elif isinstance(tag, Iterable):
self.tags = Tags[C, V].from_iter(tag)
else:
self.tags = Tags[C, V].from_iter(kwargs.values())
if isinstance(tag, Tag)
self.tags.add_tag(tag)
def is_satisfied(self, task: QTask[C, V, Q, A]) -> bool:
return all(task.tags.has_tag(tag) for tag in self.tags)

View File

@ -0,0 +1,37 @@
from typing import Iterable, List, Tuple, Union, overload
from modules.constrains.static import VTaskStaticConstraint
from modules.tags import Tag, Tags
from modules.task import QTask
from utils.types import A, C, Q, V
class MustIncludeAnyTagConstraint(VTaskStaticConstraint[C, V, Q, A]):
tags: Tags[C, V] = Tags()
@overload
def __init__(self, tag: Tags[C, V]): ...
@overload
def __init__(self, tag: Iterable[Tag[C, V]])
@overload
def __init__(self, tag: Tag[C, V], **kwargs: Tag[C, V]): ...
def __init__(
self,
tag: Union[Tags[C, V], Tag[C, V], Iterable[Tag[C, V]]],
**kwargs: Tag[C, V],
):
if isinstance(tag, Tags):
self.tags = tag
elif isinstance(tag, Iterable):
self.tags = Tags[C, V].from_iter(tag)
else:
self.tags = Tags[C, V].from_iter(kwargs.values())
if isinstance(tag, Tag)
self.tags.add_tag(tag)
def is_satisfied(self, task: QTask[C, V, Q, A]) -> bool:
return any(task.tags.has_tag(tag) for tag in self.tags)

View File

@ -0,0 +1,13 @@
from dataclasses import dataclass
from modules.constrains.static import VTaskStaticConstraint
from modules.task import QTask
from utils.types import A, C, Q, V
@dataclass
class MustNotStatic(VTaskStaticConstraint[C, V, Q, A]):
constraint: VTaskStaticConstraint[C, V, Q, A]
def is_satisfied(self, task: QTask[C, V, Q, A]) -> bool:
return not self.constraint.is_satisfied(task)

View File

@ -0,0 +1,6 @@
from typing import Union
from modules.fabric import QTaskFabric
from modules.task import QTask
type QTaskOrFabric[C, V, Q, A] = Union[QTask[C, V, Q, A], QTaskFabric[C, V, Q, A]]

15
modules/fabric.py Normal file
View File

@ -0,0 +1,15 @@
from typing import Generic, Protocol, runtime_checkable
from option import Option
from modules.fabric_metadata import QTaskFactoryMetadata
from modules.task import QTask
from utils.types import A, C, Q, V
@runtime_checkable
class QTaskFactory(Protocol, Generic[C, V, Q, A]):
metadata: QTaskFactoryMetadata[C, V]
default_tasks_to_generate: Option[int] = Option.maybe(None)
def generate(self) -> QTask[C, V, Q, A]: ...

View File

@ -0,0 +1,24 @@
import uuid
from dataclasses import dataclass, field
from typing import Generic, Optional
from option import Option
from utils.utils import C, V
@dataclass(frozen=True)
class QTaskFactoryMetadata(Generic[C, V]):
name: Option[str] = Option.maybe(None)
description: Option[str] = Option.maybe(None)
id: uuid.UUID = uuid.uuid4()
@staticmethod
def from_values(
name: Optional[str] = None,
description: Optional[str] = None,
) -> "QTaskFactoryMetadata[C, V]":
return QTaskFactoryMetadata(
name=Option.maybe(name),
description=Option.maybe(description),
)

View File

@ -1,24 +1,38 @@
from dataclasses import dataclass, field from dataclasses import dataclass, field
from enum import Enum from enum import Enum
from typing import Dict, Generic, List, Optional, Set, Tuple, TypeVar, Union, overload from typing import (
Dict,
Generic,
Iterable,
List,
Optional,
Set,
Tuple,
TypeVar,
Union,
overload,
override,
)
from utils.utils import C, V from utils.utils import C, V
type Tag[C, V] = Tuple[C, V]
@dataclass(frozen=True)
class Tag(Generic[C, V]):
cat: C
val: V
@dataclass @dataclass
class Tags(Dict[C, Set[V]], Generic[C, V]): class Tags(Generic[C, V]):
@staticmethod _dict: Dict[C, Set[V]] = {}
def from_list(tags_list: List[Tag[C, V]]) -> "Tags[C, V]":
tags: Tags[C, V] = Tags()
for cat, val in tags_list:
tags.setdefault(cat, set()).add(val)
return tags
def has_tag_tuple(self, tag: Tag[C, V]) -> bool: @staticmethod
cat, val = tag def from_iter(iter: Iterable[Tag[C, V]]) -> "Tags[C, V]":
return val in self.get(cat, set()) tags: Tags[C, V] = Tags()
for tag in iter:
tags._dict.setdefault(tag.cat, set()).add(tag.val)
return tags
@overload @overload
def has_tag(self, category: C, value: V) -> bool: ... def has_tag(self, category: C, value: V) -> bool: ...
@ -27,17 +41,14 @@ class Tags(Dict[C, Set[V]], Generic[C, V]):
def has_tag(self, category: Tag[C, V]) -> bool: ... def has_tag(self, category: Tag[C, V]) -> bool: ...
def has_tag(self, category: Union[C, Tag[C, V]], value: Optional[V] = None) -> bool: def has_tag(self, category: Union[C, Tag[C, V]], value: Optional[V] = None) -> bool:
if isinstance(category, Tuple): if isinstance(category, Tag):
return self.has_tag_tuple(category) tag = category
return tag.val in self._dict.get(tag.cat, set())
else: else:
assert ( assert (
value is not None value is not None
), "Value must be provided if category is not a tuple" ), "Value must be provided if category is not a tuple"
return value in self.get(category, set()) return value in self._dict.get(category, set())
def add_tag_tuple(self, tag: Tag[C, V]) -> None:
cat, val = tag
self.get(cat, set()).add(val)
@overload @overload
def add_tag(self, category: C, value: V) -> None: ... def add_tag(self, category: C, value: V) -> None: ...
@ -46,23 +57,29 @@ class Tags(Dict[C, Set[V]], Generic[C, V]):
def add_tag(self, category: Tag[C, V]) -> None: ... def add_tag(self, category: Tag[C, V]) -> None: ...
def add_tag(self, category: Union[C, Tag[C, V]], value: Optional[V] = None) -> None: def add_tag(self, category: Union[C, Tag[C, V]], value: Optional[V] = None) -> None:
if isinstance(category, Tuple): if isinstance(category, Tag):
self.add_tag_tuple(category) tag = category
self._dict.get(tag.cat, set()).add(tag.val)
else: else:
assert ( assert (
value is not None value is not None
), "Value must be provided if category is not a tuple" ), "Value must be provided if category is not a tuple"
self.get(category, set()).add(value) self._dict.get(category, set()).add(value)
def __str__(self) -> str: def __str__(self) -> str:
if len(self) == 0: if len(self._dict) == 0:
return "No tags" return "No tags"
lines = [] lines = []
for category, values in self.items(): for category, values in self._dict.items():
cat_str = category.value if isinstance(category, Enum) else str(category) cat_str = category.value if isinstance(category, Enum) else str(category)
val_strs = sorted( val_strs = sorted(
[v.value if isinstance(v, Enum) else str(v) for v in values] [v.value if isinstance(v, Enum) else str(v) for v in values]
) )
lines.append(f"{cat_str}: {', '.join(val_strs)}") lines.append(f"{cat_str}: {', '.join(val_strs)}")
return "\n".join(lines) return "\n".join(lines)
def __iter__(self):
for category, values in self._dict.items():
for value in values:
yield Tag(category, value)

35
modules/task.py Normal file
View File

@ -0,0 +1,35 @@
import uuid
from dataclasses import dataclass
from typing import Generic, List, Optional, Union
from option import NONE, Option
from tags import Tag, Tags
from modules.fabric_metadata import QTaskFactoryMetadata
from utils.types import A, C, Q, V
from utils.utils import indent
@dataclass
class QTask(Generic[C, V, Q, A]):
question: Q
answer: A
tags: Tags[C, V] = Tags()
fabric_metadata: Option[QTaskFactoryMetadata[C, V]] = Option.maybe(NONE)
id: uuid.UUID = uuid.uuid4()
def __init__(
self,
question: Q,
answer: A,
tags: Optional[Union[Tags[C, V], List[Tag[C, V]]]] = None,
):
self.question = question
self.answer = answer
if isinstance(tags, List):
self.tags = Tags[C, V].from_list(tags)
elif isinstance(tags, Tags):
self.tags = tags
def __str__(self):
return f"Question:\n{indent(str(self.question))}\nAnswer:\n{indent(str(self.answer))}\nTags:\n{indent(str(self.tags))}"

9
modules/task_pool.py Normal file
View File

@ -0,0 +1,9 @@
from typing import Generic, List
from task import QTask
from utils.types import A, C, Q, V
class QTaskPool(Generic[C, V, Q, A]):
pool: List[QTask[C, V, Q, A]]

12
modules/variant.py Normal file
View File

@ -0,0 +1,12 @@
from typing import Generic, List
from modules.task import QTask
from utils.types import A, C, Q, V
class QVariant(Generic[C, V, Q, A]):
tasks: List[QTask[C, V, Q, A]]
def __iter__(self):
for task in self.tasks:
yield task

12
modules/variant_set.py Normal file
View File

@ -0,0 +1,12 @@
from typing import Generic, List
from modules.variant import QVariant
from utils.types import A, C, Q, V
class QVariantSet(Generic[C, V, Q, A]):
variants: List[QVariant[C, V, Q, A]]
def __iter__(self):
for variant in self.variants:
yield variant

View File

6
utils/types.py Normal file
View File

@ -0,0 +1,6 @@
from typing import TypeVar
C = TypeVar("C", default=str)
V = TypeVar("V", default=str)
Q = TypeVar("Q", default=str)
A = TypeVar("A", default=Q)