import dataclasses import inspect from typing import TYPE_CHECKING, Any, Callable, Optional, Any from enum import IntEnum from operator import eq, ge, le from .Regions import regionMap from .hooks import Rules from .Helpers import clamp, is_item_enabled, is_option_enabled, get_option_value, convert_string_to_type,\ format_to_valid_identifier, format_state_prog_items_key, ProgItemsCat from .Game import game_name from BaseClasses import MultiWorld, CollectionState, Entrance from worlds.AutoWorld import World from worlds.generic.Rules import set_rule, add_rule from Options import Choice, Toggle, Range, NamedRange, NumericOption from Utils import version_tuple import re import math import logging if TYPE_CHECKING: from . import ManualWorld # At some point in the future, we should depreciate the non-RB codepath. But that's not until at least 0.7.X use_rulebuilder = version_tuple >= (0, 6, 7) if TYPE_CHECKING and use_rulebuilder: import rule_builder.rules FUNCTION_REGEX = re.compile(r'\{(\w+)\((.*?)\)\}') ITEM_REGEX = re.compile(r'\|(@?)([^|]+?)(\:[^:|]+)?\|') AND_REGEX = re.compile(r'\s?\bAND\b\s?', re.IGNORECASE) OR_REGEX = re.compile(r'\s?\bOR\b\s?', flags=re.IGNORECASE) class LogicErrorSource(IntEnum): INFIX_TO_POSTFIX = 1 """includes more closing parentheses than opening (but not the opposite)""" EVALUATE_POSTFIX = 2 """ includes missing pipes and missing value on either side of AND/OR""" EVALUATE_STACK_SIZE = 3 """includes missing curly brackets""" def construct_logic_error(location_or_region: dict, source: LogicErrorSource) -> KeyError: object_type = "location/region" object_name = location_or_region.get("name", "Unknown") if location_or_region.get("is_region", False) or "starting" in location_or_region or "connects_to" in location_or_region: object_type = "region" elif "region" in location_or_region or "category" in location_or_region: object_type = "location" if source == LogicErrorSource.INFIX_TO_POSTFIX: source_text = "There may be mismatched parentheses, or other invalid syntax for the requires." elif source == LogicErrorSource.EVALUATE_POSTFIX: source_text = "There may be missing || around item names, or an AND/OR that is missing a value on one side, or other invalid syntax for the requires." elif source == LogicErrorSource.EVALUATE_STACK_SIZE: source_text = "There may be missing {} around requirement functions like YamlEnabled() / YamlDisabled(), or other invalid syntax for the requires." else: source_text = "This requires includes invalid syntax." return KeyError(f"Invalid 'requires' for {object_type} '{object_name}': {source_text} (ERROR {source})") def infix_to_postfix(expr: str, location: dict) -> str: prec: dict[str, int] = {"&": 2, "|": 2, "!": 3} stack: list[str] = [] postfix: str = "" try: for c in expr: if c.isnumeric(): postfix += c elif c in prec: while stack and stack[-1] != "(" and prec[c] <= prec[stack[-1]]: postfix += stack.pop() stack.append(c) elif c == "(": stack.append(c) elif c == ")": while stack and stack[-1] != "(": postfix += stack.pop() stack.pop() else: # added this here since '|Chun-Li| or {YamlCompare(Example_Choice == 1)' is valid because of the 1 at the end raise ValueError(f"Invalid Character '{c}' in expression '{expr}', it should be either a number or a parentheses") while stack: postfix += stack.pop() except ValueError as ex: text = str(ex) if "'{'" in text or ")}" in text: raise construct_logic_error(location, LogicErrorSource.EVALUATE_STACK_SIZE) raise construct_logic_error(location, LogicErrorSource.EVALUATE_POSTFIX) except Exception: raise construct_logic_error(location, LogicErrorSource.INFIX_TO_POSTFIX) return postfix def evaluate_postfix(expr: str, location: dict) -> bool: stack: list[bool] = [] try: for c in expr: if c == "0": stack.append(False) elif c == "1": stack.append(True) elif c == "&": op2 = stack.pop() op1 = stack.pop() stack.append(op1 and op2) elif c == "|": op2 = stack.pop() op1 = stack.pop() stack.append(op1 or op2) elif c == "!": op = stack.pop() stack.append(not op) except Exception: raise construct_logic_error(location, LogicErrorSource.EVALUATE_POSTFIX) if len(stack) != 1: raise construct_logic_error(location, LogicErrorSource.EVALUATE_STACK_SIZE) return stack.pop() def set_rules(world: "ManualWorld", multiworld: MultiWorld, player: int): def evaluate_nonnumeric_count(item_base: str, item_name: str, item_count: str, is_category: bool, area: dict) -> tuple[str, int]: item_count = item_count.strip() if item_count.isnumeric(): return item_name, int(item_count) items_counts = world.get_item_counts(player, only_progression=True) if is_category: total_count = sum([items_counts.get(item, 0) for item in world.item_and_event_name_groups.get(item_name, set())]) else: total_count = items_counts.get(item_name, 0) if item_count.lower() == 'all': count = total_count elif item_count.lower() == 'half': count = int(total_count / 2) elif item_count.endswith('%') and len(item_count) > 1: percent = clamp(float(item_count[:-1]) / 100, 0, 1) count = math.ceil(total_count * percent) # If invalid count assume its actually part of the item name else: item_name = item_base.strip("|") count = 1 return item_name, count def construct_rule_from_string(area: dict) -> "rule_builder.rules.Rule | None": if not use_rulebuilder: return None import rule_builder.rules requires_list = area.get('requires', '') if requires_list == "": return rule_builder.rules.True_() def recursively_tokenize_manual_rule(partial: str) -> "rule_builder.rules.Rule | None": if not partial: return rule_builder.rules.True_() rule: Rule | None = None remaining = '' partial = partial.strip() if match := ITEM_REGEX.match(partial): is_category = bool(match.group(1)) item_name = match.group(2) item_count = (str(match.group(3) or "1")).lstrip(':') if item_count.isnumeric(): count = int(item_count) else: item_name, count = evaluate_nonnumeric_count(match.group(0), item_name, item_count, is_category, area) if is_category: rule = rule_builder.rules.HasFromList(*world.item_and_event_name_groups.get(item_name, set()), count=count) else: rule = rule_builder.rules.Has(item_name, count) remaining = partial[len(match.group(0)):] elif match := FUNCTION_REGEX.match(partial): func_name = match.group(1) func_args = match.group(2).split(",") if func_args == ['']: func_args.pop() rule_class = None search = [ (rule_builder.rules.DEFAULT_RULES, func_name), (globals(), func_name + "Rule"), (globals(), func_name), (Rules, func_name + "Rule"), (Rules, func_name), ] for ns, name in search: if isinstance(ns, dict): func = ns.get(name) else: func = getattr(ns, name, None) if func and inspect.isclass(func) and issubclass(func, rule_builder.rules.Rule): convert_req_function_args(None, func, func_args, area['name'], world) rule_class = func break if func and inspect.signature(func).return_annotation is str: # I'm assuming that functions that return strings don't need states. convert_req_function_args(None, func, func_args, area['name'], world) rule = recursively_tokenize_manual_rule(func(*func_args)) break if rule is None: if not rule_class: logging.warning(f'Warning: Could not find Rule implementation of {func_name}.') # By returning None, we're saying "This entire requires string can't be done with a Rule. Fall back to the pre-rb lambdas" return None rule = rule_class(*func_args) remaining = partial[len(match.group(0)):] elif partial[0] == "(": func_founds: list[str] = [] for match in FUNCTION_REGEX.finditer(partial): if match.group(0) not in partial: # already done all of them continue func_founds.append(match.group(0)) # looks like : {{Function#0}} id = len(func_founds) - 1 partial = partial.replace(match.group(0), f"{{{{Function#{id}}}}}") inner = '' queue = list(partial[1:]) stack = 1 while stack > 0: if not queue: raise construct_logic_error(area, LogicErrorSource.INFIX_TO_POSTFIX) c = queue.pop(0) if c == "(": stack += 1 elif c == ")": stack -= 1 else: inner += c remaining = "".join(queue) for id, func in enumerate(func_founds): remaining = remaining.replace(f"{{{{Function#{id}}}}}", func) inner = inner.replace(f"{{{{Function#{id}}}}}", func) rule = recursively_tokenize_manual_rule(inner) else: logging.warning(f'Warning: Could not convert {partial} into a Rule') return None if rule is None: return None if match := OR_REGEX.match(remaining): remaining = remaining[len(match.group(0)):] right = recursively_tokenize_manual_rule(remaining) if right is None: return None rule = rule | right remaining = '' elif match := AND_REGEX.match(remaining): remaining = remaining[len(match.group(0)):] right = recursively_tokenize_manual_rule(remaining) if right is None: return None rule = rule & right remaining = '' if remaining: raise ValueError(f'Unexpected token `{remaining}` in {area["name"]} rule') return rule return recursively_tokenize_manual_rule(requires_list) # this is only called when the area (think, location or region) has a "requires" field that is a string def checkRequireStringForArea(state: CollectionState, area: dict): requires_list = area["requires"] # Preparing some variables for exception messages area_type = "region" if area.get("is_region",False) else "location" area_name = area.get("name", f"unknown with these parameters: {area}") if requires_list == "": return True def findAndRecursivelyExecuteFunctions(requires_list: str, recursionDepth: int = 0) -> str: found_functions = FUNCTION_REGEX.findall(requires_list) if found_functions: if recursionDepth > world.rules_functions_maximum_recursion: raise RecursionError(f'One or more functions in {area_type} "{area_name}"\'s requires looped too many time (maximum recursion is {world.rules_functions_maximum_recursion}) \ \n As of this Exception the following function(s) are waiting to run: {[f[0] for f in found_functions]} \ \n And the currently processed requires look like this: "{requires_list}"') else: for item in found_functions: func_name = item[0] func_args = item[1].split(",") if func_args == ['']: func_args.pop() func = globals().get(func_name) if func is None: func = getattr(Rules, func_name, None) if not callable(func): raise ValueError(f'Invalid function "{func_name}" in {area_type} "{area_name}".') convert_req_function_args(state, func, func_args, area_name, world) try: result = func(*func_args) except Exception as ex: raise RuntimeError(f'A call to the function "{func_name}" in {area_type} "{area_name}"\'s requires raised an Exception. \ \nUnless it was called by another function, it should look something like "{{{func_name}({item[1]})}}" in {area_type}s.json. \ \nFull error message: \ \n\n{type(ex).__name__}: {ex}') if isinstance(result, bool): requires_list = requires_list.replace("{" + func_name + "(" + item[1] + ")}", "1" if result else "0") else: requires_list = requires_list.replace("{" + func_name + "(" + item[1] + ")}", str(result)) requires_list = findAndRecursivelyExecuteFunctions(requires_list, recursionDepth + 1) return requires_list requires_list = findAndRecursivelyExecuteFunctions(requires_list) # parse user written statement into list of each item for match in ITEM_REGEX.finditer(requires_list): item_base = match.group(0) is_category = bool(match.group(1)) item_name = match.group(2) item_count = match.group(3) if item_base not in requires_list: # previous instance of this item was already processed continue if not item_count: item_count = "1" item_count = item_count.lstrip(':') item_name, numeric_count = evaluate_nonnumeric_count(item_base, item_name, item_count, is_category, area) if is_category: found = state.has_from_list(world.item_and_event_name_groups.get(item_name, set()), player, numeric_count) else: found = state.has(item_name, player, numeric_count) if found: requires_list = requires_list.replace(item_base, "1") else: requires_list = requires_list.replace(item_base, "0") requires_list = AND_REGEX.sub('&', requires_list, count=0) requires_list = OR_REGEX.sub('|', requires_list, count=0) requires_string = infix_to_postfix("".join(requires_list), area) return (evaluate_postfix(requires_string, area)) # this is only called when the area (think, location or region) has a "requires" field that is a dict def checkRequireDictForArea(state: CollectionState, area: dict): canAccess = True for item in area["requires"]: # if the require entry is an object with "or" or a list of items, treat it as a standalone require of its own if (isinstance(item, dict) and "or" in item and isinstance(item["or"], list)) or (isinstance(item, list)): canAccessOr = True or_items = item if isinstance(item, dict): or_items = item["or"] for or_item in or_items: or_item_parts = or_item.split(":") or_item_name = or_item or_item_count = 1 if len(or_item_parts) > 1: or_item_name = or_item_parts[0] or_item_count = int(or_item_parts[1]) if not state.has(or_item_name, player, or_item_count): canAccessOr = False if canAccessOr: canAccess = True break else: item_parts = item.split(":") item_name = item item_count = 1 if len(item_parts) > 1: item_name = item_parts[0] item_count = int(item_parts[1]) if not state.has(item_name, player, item_count): canAccess = False return canAccess # handle any type of checking needed, then ferry the check off to a dedicated method for that check def fullLocationOrRegionCheck(state: CollectionState, area: dict): # if it's not a usable object of some sort, default to true if not area: return True # don't require the "requires" key for locations and regions if they don't need to use it if "requires" not in area.keys(): return True if isinstance(area["requires"], str): return checkRequireStringForArea(state, area) else: # item access is in dict form return checkRequireDictForArea(state, area) used_location_names = [] # Region access rules extra_entrance_rules = {} for region in regionMap.keys(): entrance_rules = regionMap[region].get("entrance_requires", {}) for e in entrance_rules: entrance = world.get_entrance(f'{e}To{region}') area = {"requires": entrance_rules[e]} extra_entrance_rules[entrance.name] = area exit_rules = regionMap[region].get("exit_requires", {}) for e in exit_rules: exit = world.get_entrance(f'{region}To{e}') area = {"requires": exit_rules[e]} extra_entrance_rules[exit.name] = area for region in regionMap.keys(): used_location_names.extend([l.name for l in multiworld.get_region(region, player).locations]) for exitRegion in multiworld.get_region(region, player).entrances: extra = extra_entrance_rules.get(exitRegion.name, {}) area = regionMap[region] area["name"] = exitRegion.name area['is_region'] = True rb_rule = construct_rule_from_string(area) if rb_rule is not None: if extra: rb_extra_rule = construct_rule_from_string(extra) if rb_extra_rule is None: raise ValueError(f'Unable to combine Rule and functions for {exitRegion.name}.') rb_rule = rb_rule & rb_extra_rule world.set_rule(world.get_entrance(exitRegion.name), rb_rule) else: def fullRegionCheck(state: CollectionState, region=area): return fullLocationOrRegionCheck(state, region) add_rule(world.get_entrance(exitRegion.name), fullRegionCheck) if extra: add_rule(world.get_entrance(exitRegion.name), lambda state, extra=extra: fullLocationOrRegionCheck(state, extra)) # Location access rules for location in (world.location_table + world.event_table): if "location_name" in location: name = location["location_name"] else: name = location["name"] if name not in used_location_names: continue locFromWorld = multiworld.get_location(name, player) if "requires" in location: rb_rule = construct_rule_from_string(location) if rb_rule is not None: world.set_rule(locFromWorld, rb_rule) else: def checkLocation(state: CollectionState, location=location): locationCheck = fullLocationOrRegionCheck(state, location) return locationCheck set_rule(locFromWorld, checkLocation) elif use_rulebuilder: import rule_builder.rules world.set_rule(locFromWorld, rule_builder.rules.True_()) else: # No location requires? It's accessible. def allRegionsAccessible(state): return True set_rule(locFromWorld, allRegionsAccessible) # Victory requirement multiworld.completion_condition[player] = lambda state: state.has("__Victory__", player) def convert_req_function_args(state: CollectionState | None, func, args: list[str | Any], areaName: str, world: World) -> None: parameters = inspect.signature(func).parameters knownParameters = [World, 'ManualWorld', MultiWorld, CollectionState] index = -1 for parameter in parameters.values(): target_type = parameter.annotation if str(target_type) == "collections.abc.Iterable[rule_builder.options.OptionFilter]": # parameters >= this one are added by rule builder and do not need to be converted break index += 1 if target_type in knownParameters: if target_type in [World, 'ManualWorld']: args.insert(index, world) elif target_type == MultiWorld: args.insert(index, world.multiworld) elif target_type == CollectionState and state is None: raise ValueError('Function needs a state but none was available') elif target_type == CollectionState: args.insert(index, state) continue if parameter.name.lower() == "player": args.insert(index, world.player) continue if index < len(args) and args[index] != "": value = args[index].strip() else: if parameter.default is not inspect.Parameter.empty: if index < len(args): args[index] = parameter.default else: args.insert(index, parameter.default) continue else: if parameter.annotation is inspect.Parameter.empty: raise Exception(f"A call of the \"{func.__name__}\" function in \"{areaName}\"'s requirement, asks for a value for its argument \"{parameter.name}\" but it's missing.") else: raise Exception(f"A call of the \"{func.__name__}\" function in \"{areaName}\"'s requirement, asks for a value of type {target_type} for its argument \"{parameter.name}\" but it's missing.") if target_type == str or parameter.annotation is inspect.Parameter.empty: #Don't convert since its already a string or if we don't know the type to convert to args[index] = value continue try: value = convert_string_to_type(value, target_type) except Exception as e: raise Exception(f"A call of the \"{func.__name__}\" function in \"{areaName}\"'s requirement, asks for a value of type {target_type}\nfor its argument \"{parameter.name}\" but its value \"{value}\" cannot be converted to {target_type} \nOriginal Error:'{e}'") args[index] = value def ItemValue(state: CollectionState, player: int, valueCount: str): """When passed a string with this format: 'valueName:int', this function will check if the player has collect at least 'int' valueName worth of items\n eg. {ItemValue(Coins:12)} will check if the player has collect at least 12 coins worth of items """ args: list[str] = valueCount.split(":") if not len(args) == 2 or not args[1].isnumeric(): raise Exception(f"ItemValue needs a number after : so it looks something like 'ItemValue({args[0]}:12)'") value_name = format_state_prog_items_key(ProgItemsCat.VALUE, args[0]) requested_count = int(args[1].strip()) return state.has(value_name, player, requested_count) # Two useful functions to make require work if an item is disabled instead of making it inaccessible def OptOne(world: "ManualWorld", item: str) -> str: """Check if the passed item (with or without ||) is enabled, then this returns |item:count| where count is clamped to the maximum number of said item in the itempool.\n Eg. requires: "{OptOne(|DisabledItem|)} and |other items|" become "|DisabledItem:0| and |other items|" if the item is disabled. """ if item == "": return "" #Skip this function if item is left blank items_counts = world.get_item_counts(only_progression=True) require_category = False if '@' in item[:2]: require_category = True item = item.lstrip('|@$').rstrip('|') item_parts = item.split(":") item_name = item item_count = '1' if len(item_parts) > 1: item_name = item_parts[0] item_count = item_parts[1] if require_category: if item_count.isnumeric(): #Only loop if we can use the result to clamp category_items = [item for item in world.item_name_to_item.values() if "category" in item and item_name in item["category"]] category_items_counts = sum([items_counts.get(category_item["name"], 0) for category_item in category_items]) item_count = clamp(int(item_count), 0, category_items_counts) return f"|@{item_name}:{item_count}|" else: if item_count.isnumeric(): item_current_count = items_counts.get(item_name, 0) item_count = clamp(int(item_count), 0, item_current_count) return f"|{item_name}:{item_count}|" # OptAll check the passed require string and loop every item to check if they're enabled, def OptAll(world: "ManualWorld", requires: str) -> bool|str: """Check the passed require string and loop every item to check if they're enabled, then returns the require string with items counts adjusted using OptOne\n eg. requires: "{OptAll(|DisabledItem| and |@CategoryWithModifedCount:10|)} and |other items|" become "|DisabledItem:0| and |@CategoryWithModifedCount:2| and |other items|" """ requires_list = requires if requires_list == "": return True # parse user written statement into list of each item for item in re.findall(r'\|[^|]+\|', requires): itemScanned = OptOne(world, item) requires_list = requires_list.replace(item, itemScanned) return requires_list # going to be deprecated to name consistently to other req functions, in pascal case def canReachLocation(state: CollectionState, player: int, location: str): logging.warning("The 'canReachLocation' requirement function is being renamed to 'CanReachLocation'. Use that instead, as the lowercase version will be deprecated.") return CanReachLocation(state, player, location) # Rule to expose the can_reach_location core function def CanReachLocation(state: CollectionState, player: int, location: str) -> bool: """Can the player reach the given location?""" if state.can_reach_location(location, player): return True return False def OptionCount(world: "ManualWorld", item: str, option_name: str) -> str: """Set the required count of 'item' to be the value set in the player's yaml of the Numerical option 'option_name'.""" return _optionCountLogic(world, item, option_name ) def OptionCountPercent(world: "ManualWorld", item: str, option_name: str) -> str: """Set the required count of 'item' to be a percentage of it total count based on the player's yaml value for Numerical option 'option_name'.""" return _optionCountLogic(world, item, option_name, is_percent=True) def _optionCountLogic(world: "ManualWorld", item: str, option_name: str, is_percent: bool = False) -> str: option_name = option_name.strip() option: NumericOption | None = getattr(world.options, option_name, None) if option is None: raise ValueError(f"Could not find an option named: {option_name}") # Verification that the value is compatible if not isinstance(option.value, int): raise ValueError(f"Cannot use a value that is not a number. Got value of '{option.value}' from option {option_name}") item = item.strip('|').strip() return f"|{item}:{option.value}{'%' if is_percent else ''}|" def YamlEnabled(multiworld: MultiWorld, player: int, param: str) -> bool: """Is a yaml option enabled?""" return is_option_enabled(multiworld, player, param) def YamlDisabled(multiworld: MultiWorld, player: int, param: str) -> bool: """Is a yaml option disabled?""" return not is_option_enabled(multiworld, player, param) def YamlCompare(world: "ManualWorld", args: str, skipCache: bool = False) -> bool: """Is a yaml option's value compared using {comparator} to the requested value \nFormat it like {YamlCompare(OptionName==value)} \nWhere == can be any of the following: ==, !=, >=, <=, <, > \nExample: {YamlCompare(Example_Range > 5)}""" comp_symbols = { #Maybe find a better name for this '==' : eq, '!=' : eq, #complement of == '>=' : ge, '<=' : le, '=': eq, #Alternate to be like yaml_option '<' : ge, #complement of >= '>' : le, #complement of <= } reverse_result = False #Find the comparator symbol to split the string with and for logs if '==' in args: comparator = '==' elif '!=' in args: comparator = '!=' reverse_result = True #complement of == thus reverse by default elif '>=' in args: comparator = '>=' elif '<=' in args: comparator = '<=' elif '=' in args: comparator = '=' elif '<' in args: comparator = '<' reverse_result = True #complement of >= elif '>' in args: comparator = '>' reverse_result = True #complement of <= else: raise ValueError(f"Could not find a valid comparator in given string '{args}', it must be one of {comp_symbols.keys()}") option_name, value = args.split(comparator) initial_option_name = str(option_name).strip() #For exception messages option_name = format_to_valid_identifier(option_name) # Detect !reversing of result like yaml_option if option_name.startswith('!'): reverse_result = not reverse_result option_name = option_name.lstrip('!') initial_option_name = initial_option_name.lstrip('!') value = value.strip() option = getattr(world.options, option_name, None) if option is None: raise ValueError(f"YamlCompare could not find an option called '{initial_option_name}' to compare against, its either missing on misspelt") if not value: #empty string '' raise ValueError(f"Could not find a valid value to compare against in given string '{args}'. \nThere must be a value to compare against after the comparator (in this case '{comparator}').") if not skipCache: #Cache made for optimization purposes cacheindex = option_name + '_' + comp_symbols[comparator].__name__ + '_' + format_to_valid_identifier(value.lower()) if not hasattr(world, 'yaml_compare_rule_cache'): world.yaml_compare_rule_cache = dict[str,bool]() if skipCache or world.yaml_compare_rule_cache.get(cacheindex, None) is None: try: if issubclass(type(option), Choice): value = convert_string_to_type(value, str|int) if isinstance(value, str): value = option.from_text(value).value elif issubclass(type(option), Range): if type(option).__base__ == NamedRange: value = convert_string_to_type(value, str|int) if isinstance(value, str): value = option.from_text(value).value else: value = convert_string_to_type(value, int) elif issubclass(type(option), Toggle): value = int(convert_string_to_type(value, bool)) else: raise ValueError(f"YamlCompare does not currently support Option of type {type(option)} \nAsk about it in #Manual-dev and it might be added.") except KeyError as ex: raise ValueError(f"YamlCompare failed to find the requested value in what the \"{initial_option_name}\" option supports.\ \nRaw error:\ \n\n{type(ex).__name__}:{ex}") except Exception as ex: raise TypeError(f"YamlCompare failed to convert the requested value to what a {type(option).__base__.__name__} option supports.\ \nCaused By:\ \n\n{type(ex).__name__}:{ex}") if isinstance(value, str) and comp_symbols[comparator].__name__ != 'eq': #At this point if its still a string don't try and compare with strings using > < >= <= raise ValueError(f'YamlCompare can only compare strings with one of the following: {[s for s, v in comp_symbols.items() if v.__name__ == "eq"]} and you tried to do: "{option.value} {comparator} {value}"') result = comp_symbols[comparator](option.value, value) if not skipCache: world.yaml_compare_rule_cache[cacheindex] = result else: #if exists and not skipCache result = world.yaml_compare_rule_cache[cacheindex] return not result if reverse_result else result if use_rulebuilder: from rule_builder.rules import Rule, Has, True_, False_ @dataclasses.dataclass() class ItemValueRule(Rule["ManualWorld"], game=game_name): valueCount: str def _instantiate(self, world: "ManualWorld") -> Rule.Resolved: args: list[str] = self.valueCount.split(":") if not len(args) == 2 or not args[1].isnumeric(): raise Exception(f"ItemValue needs a number after : so it looks something like 'ItemValue({args[0]}:12)'") value_name = format_state_prog_items_key(ProgItemsCat.VALUE, args[0]) requested_count = int(args[1].strip()) return Has(value_name, requested_count).resolve(world) @dataclasses.dataclass() class YamlEnabledRule(Rule["ManualWorld"], game=game_name): yaml_option: str def _instantiate(self, world: "ManualWorld") -> Rule.Resolved: if getattr(world.options, self.yaml_option).value: return True_().resolve(world) else: return False_().resolve(world) @dataclasses.dataclass() class YamlDisabledRule(Rule["ManualWorld"], game=game_name): yaml_option: str def _instantiate(self, world: "ManualWorld") -> Rule.Resolved: if getattr(world.options, self.yaml_option).value: return False_().resolve(world) else: return True_().resolve(world) @dataclasses.dataclass() class YamlCompareRule(Rule["ManualWorld"], game=game_name): yaml_comparison: str def _instantiate(self, world: "ManualWorld") -> Rule.Resolved: if YamlCompare(world, self.yaml_comparison): return True_().resolve(world) else: return False_().resolve(world) # # This was an attempt to make a universal lambda caller rule. It's messy, complicated, and probably not worth it. # # As you can see above, you are much better off implementing optimized Rules rather than trying to execute check-time access rules this way. # # This code is left here both as a warning and also a starting point. # # But as things currently stand, you're better off either letting the whole rule fall back to lambdas or writing a proper custom Rule. # # This middleground is not a good idea. # @dataclasses.dataclass() # class FallbackLambdaRule(Rule["ManualWorld"], game=game_name): # access_rule: Callable # args: list[str | Any] # def _instantiate(self, world: "ManualWorld") -> Rule.Resolved: # return self.Resolved(self.access_rule, self.args, world) # class Resolved(Rule.Resolved): # access_rule: Callable # args: list[str | Any] # world: World # @override # def _evaluate(self, state: CollectionState) -> bool: # args = self.args.copy() # convert_req_function_args(state, self.access_rule, args, "", self.world) # value = self.access_rule(*args) # if isinstance(value, bool): # return value # name = self.access_rule.__name__ # raise ValueError(f'Unexpected return value {value} from {name}')