Core: Add rule builder (#5048)

* initial commit of rules engine

* implement most of the stuff

* add docs and fill out rest of the functionality

* add in explain functions

* dedupe items and add more docs

* pr feedback and optimization updates

* Self is not in typing on 3.10

* fix test

* Update docs/rule builder.md

Co-authored-by: BadMagic100 <dempsey.sean@outlook.com>

* pr feedback

* love it when CI gives me different results than local

* add composition with bitwise and and or

* strongly typed option filtering

* skip resolving location parent region

* update docs

* update typing and add decorator

* add string explains

* move simplify code to world

* add wrapper rule

* I may need to abandon the generic typing

* missing space for faris

* fix hashing for resolved rules

* thank u typing extensions ilu

* remove bad cacheable check

* add decorator to assign hash and rule name

* more type crimes...

* region access rules are now cached

* break compatibility so new features work

* update docs

* replace decorators with __init_subclass__

* ok now the frozen dataclass is automatic

* one more type fix for the road

* small fixes and caching tests

* play nicer with tests

* ok actually fix the tests

* add item_mapping for faris

* add more state helpers as rules

* fix has from list rules

* fix can reach location caching and add set completion condition

* fix can reach entrance caching

* implement HasGroup and HasGroupUnique

* add more tests and fix some bugs

* Add name arg to create_entrance

Co-authored-by: roseasromeo <11944660+roseasromeo@users.noreply.github.com>

* fix json dumping option filters

* restructure and test serialization

* add prop to disable caching

* switch to __call__ and revert access_rule changes

* update docs and make edge cases match

* ruff has lured me into a false sense of security

* also unused

* fix disabling caching

* move filter function to filter class

* add more docs

* tests for explain functions

* Update docs/rule builder.md

Co-authored-by: roseasromeo <11944660+roseasromeo@users.noreply.github.com>

* chore: Strip out uses of TYPE_CHECKING as much as possible

* chore: add empty webworld for test

* chore: optimize rule evaluations

* remove getattr from hot code paths

* testing new cache flags

* only clear cache for rules cached as false in collect

* update test for new behaviour

* do not have rules inherit from each other

* update docs on caching

* fix name of attribute

* make explain messages more colorful

* fix issue with combining rules with different options

* add convenience functions for filtering

* use an operator with higher precedence

* name conflicts less with optionfilter

* move simplify and instance caching code

* update docs

* kill resolve_rule

* kill true_rule and false_rule

* move helpers to base classes

* update docs

* I really should finish all of my

* fix test

* rename mixin

* fix typos

* refactor rule builder into folder for better imports

* update docs

* do not dupe collectionrule

* docs review feedback

* missed a file

* remove rule_caching_enabled from base World

* update docs on caching

* shuffle around some docs

* use option instead of option.value

* add in operator and more testing

* rm World = object

* test fixes

* move cache to logic mixin

* keep test rule builder world out of global registry

* todone

* call register_dependencies automatically

* move register deps call to call_single

* add filtered_resolution

* allow bool opts on filters

* fix serialization tests

* allow reverse operations

---------

Co-authored-by: BadMagic100 <dempsey.sean@outlook.com>
Co-authored-by: roseasromeo <11944660+roseasromeo@users.noreply.github.com>

rule_builder/cached_world.py

@@ -0,0 +1,146 @@
+from collections import defaultdict
+from typing import ClassVar, cast
+
+from typing_extensions import override
+
+from BaseClasses import CollectionState, Item, MultiWorld, Region
+from worlds.AutoWorld import LogicMixin, World
+
+from .rules import Rule
+
+
+class CachedRuleBuilderWorld(World):
+    """A World subclass that provides helpers for interacting with the rule builder"""
+
+    rule_item_dependencies: dict[str, set[int]]
+    """A mapping of item name to set of rule ids"""
+
+    rule_region_dependencies: dict[str, set[int]]
+    """A mapping of region name to set of rule ids"""
+
+    rule_location_dependencies: dict[str, set[int]]
+    """A mapping of location name to set of rule ids"""
+
+    rule_entrance_dependencies: dict[str, set[int]]
+    """A mapping of entrance name to set of rule ids"""
+
+    item_mapping: ClassVar[dict[str, str]] = {}
+    """A mapping of actual item name to logical item name.
+    Useful when there are multiple versions of a collected item but the logic only uses one. For example:
+    item = Item("Currency x500"), rule = Has("Currency", count=1000), item_mapping = {"Currency x500": "Currency"}"""
+
+    rule_caching_enabled: ClassVar[bool] = True
+    """Flag to inform rules that the caching system for this world is enabled. It should not be overridden."""
+
+    def __init__(self, multiworld: MultiWorld, player: int) -> None:
+        super().__init__(multiworld, player)
+        self.rule_item_dependencies = defaultdict(set)
+        self.rule_region_dependencies = defaultdict(set)
+        self.rule_location_dependencies = defaultdict(set)
+        self.rule_entrance_dependencies = defaultdict(set)
+
+    @override
+    def register_rule_dependencies(self, resolved_rule: Rule.Resolved) -> None:
+        for item_name, rule_ids in resolved_rule.item_dependencies().items():
+            self.rule_item_dependencies[item_name] |= rule_ids
+        for region_name, rule_ids in resolved_rule.region_dependencies().items():
+            self.rule_region_dependencies[region_name] |= rule_ids
+        for location_name, rule_ids in resolved_rule.location_dependencies().items():
+            self.rule_location_dependencies[location_name] |= rule_ids
+        for entrance_name, rule_ids in resolved_rule.entrance_dependencies().items():
+            self.rule_entrance_dependencies[entrance_name] |= rule_ids
+
+    def register_rule_builder_dependencies(self) -> None:
+        """Register all rules that depend on locations or entrances with their dependencies"""
+        for location_name, rule_ids in self.rule_location_dependencies.items():
+            try:
+                location = self.get_location(location_name)
+            except KeyError:
+                continue
+            if not isinstance(location.access_rule, Rule.Resolved):
+                continue
+            for item_name in location.access_rule.item_dependencies():
+                self.rule_item_dependencies[item_name] |= rule_ids
+            for region_name in location.access_rule.region_dependencies():
+                self.rule_region_dependencies[region_name] |= rule_ids
+
+        for entrance_name, rule_ids in self.rule_entrance_dependencies.items():
+            try:
+                entrance = self.get_entrance(entrance_name)
+            except KeyError:
+                continue
+            if not isinstance(entrance.access_rule, Rule.Resolved):
+                continue
+            for item_name in entrance.access_rule.item_dependencies():
+                self.rule_item_dependencies[item_name] |= rule_ids
+            for region_name in entrance.access_rule.region_dependencies():
+                self.rule_region_dependencies[region_name] |= rule_ids
+
+    @override
+    def collect(self, state: CollectionState, item: Item) -> bool:
+        changed = super().collect(state, item)
+        if changed and self.rule_item_dependencies:
+            player_results = cast(dict[int, bool], state.rule_builder_cache[self.player])  # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType]
+            mapped_name = self.item_mapping.get(item.name, "")
+            rule_ids = self.rule_item_dependencies[item.name] | self.rule_item_dependencies[mapped_name]
+            for rule_id in rule_ids:
+                if player_results.get(rule_id, None) is False:
+                    del player_results[rule_id]
+
+        return changed
+
+    @override
+    def remove(self, state: CollectionState, item: Item) -> bool:
+        changed = super().remove(state, item)
+        if not changed:
+            return changed
+
+        player_results = cast(dict[int, bool], state.rule_builder_cache[self.player])  # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType]
+        if self.rule_item_dependencies:
+            mapped_name = self.item_mapping.get(item.name, "")
+            rule_ids = self.rule_item_dependencies[item.name] | self.rule_item_dependencies[mapped_name]
+            for rule_id in rule_ids:
+                player_results.pop(rule_id, None)
+
+        # clear all region dependent caches as none can be trusted
+        if self.rule_region_dependencies:
+            for rule_ids in self.rule_region_dependencies.values():
+                for rule_id in rule_ids:
+                    player_results.pop(rule_id, None)
+
+        # clear all location dependent caches as they may have lost region access
+        if self.rule_location_dependencies:
+            for rule_ids in self.rule_location_dependencies.values():
+                for rule_id in rule_ids:
+                    player_results.pop(rule_id, None)
+
+        # clear all entrance dependent caches as they may have lost region access
+        if self.rule_entrance_dependencies:
+            for rule_ids in self.rule_entrance_dependencies.values():
+                for rule_id in rule_ids:
+                    player_results.pop(rule_id, None)
+
+        return changed
+
+    @override
+    def reached_region(self, state: CollectionState, region: Region) -> None:
+        super().reached_region(state, region)
+        if self.rule_region_dependencies:
+            player_results = cast(dict[int, bool], state.rule_builder_cache[self.player])  # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType]
+            for rule_id in self.rule_region_dependencies[region.name]:
+                player_results.pop(rule_id, None)
+
+
+class CachedRuleBuilderLogicMixin(LogicMixin):
+    multiworld: MultiWorld  # pyright: ignore[reportUninitializedInstanceVariable]
+    rule_builder_cache: dict[int, dict[int, bool]]  # pyright: ignore[reportUninitializedInstanceVariable]
+
+    def init_mixin(self, multiworld: "MultiWorld") -> None:
+        players = multiworld.get_all_ids()
+        self.rule_builder_cache = {player: {} for player in players}
+
+    def copy_mixin(self, new_state: "CachedRuleBuilderLogicMixin") -> "CachedRuleBuilderLogicMixin":
+        new_state.rule_builder_cache = {
+            player: player_results.copy() for player, player_results in self.rule_builder_cache.items()
+        }
+        return new_state

rule_builder/options.py

@@ -0,0 +1,91 @@
+import dataclasses
+import importlib
+import operator
+from collections.abc import Callable, Iterable
+from typing import Any, Final, Literal, Self, cast
+
+from typing_extensions import override
+
+from Options import CommonOptions, Option
+
+Operator = Literal["eq", "ne", "gt", "lt", "ge", "le", "contains", "in"]
+
+OPERATORS: Final[dict[Operator, Callable[..., bool]]] = {
+    "eq": operator.eq,
+    "ne": operator.ne,
+    "gt": operator.gt,
+    "lt": operator.lt,
+    "ge": operator.ge,
+    "le": operator.le,
+    "contains": operator.contains,
+    "in": operator.contains,
+}
+OPERATOR_STRINGS: Final[dict[Operator, str]] = {
+    "eq": "==",
+    "ne": "!=",
+    "gt": ">",
+    "lt": "<",
+    "ge": ">=",
+    "le": "<=",
+}
+REVERSE_OPERATORS: Final[tuple[Operator, ...]] = ("in",)
+
+
+@dataclasses.dataclass(frozen=True)
+class OptionFilter:
+    option: type[Option[Any]]
+    value: Any
+    operator: Operator = "eq"
+
+    def to_dict(self) -> dict[str, Any]:
+        """Returns a JSON compatible dict representation of this option filter"""
+        return {
+            "option": f"{self.option.__module__}.{self.option.__name__}",
+            "value": self.value,
+            "operator": self.operator,
+        }
+
+    def check(self, options: CommonOptions) -> bool:
+        """Tests the given options dataclass to see if it passes this option filter"""
+        option_name = next(
+            (name for name, cls in options.__class__.type_hints.items() if cls is self.option),
+            None,
+        )
+        if option_name is None:
+            raise ValueError(f"Cannot find option {self.option.__name__} in options class {options.__class__.__name__}")
+        opt = cast(Option[Any] | None, getattr(options, option_name, None))
+        if opt is None:
+            raise ValueError(f"Invalid option: {option_name}")
+
+        fn = OPERATORS[self.operator]
+        return fn(self.value, opt) if self.operator in REVERSE_OPERATORS else fn(opt, self.value)
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> Self:
+        """Returns a new OptionFilter instance from a dict representation"""
+        if "option" not in data or "value" not in data:
+            raise ValueError("Missing required value and/or option")
+
+        option_path = data["option"]
+        try:
+            option_mod_name, option_cls_name = option_path.rsplit(".", 1)
+            option_module = importlib.import_module(option_mod_name)
+            option = getattr(option_module, option_cls_name, None)
+        except (ValueError, ImportError) as e:
+            raise ValueError(f"Cannot parse option '{option_path}'") from e
+        if option is None or not issubclass(option, Option):
+            raise ValueError(f"Invalid option '{option_path}' returns type '{option}' instead of Option subclass")
+
+        value = data["value"]
+        operator = data.get("operator", "eq")
+        return cls(option=cast(type[Option[Any]], option), value=value, operator=operator)
+
+    @classmethod
+    def multiple_from_dict(cls, data: Iterable[dict[str, Any]]) -> tuple[Self, ...]:
+        """Returns a tuple of OptionFilters instances from an iterable of dict representations"""
+        return tuple(cls.from_dict(o) for o in data)
+
+    @override
+    def __str__(self) -> str:
+        op = OPERATOR_STRINGS.get(self.operator, self.operator)
+        return f"{self.option.__name__} {op} {self.value}"