Skip to content

Core, MultiServer, CommonClient: Store Hint Priority & the found attribute together in an IntFlag for efficiency #4721

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 10 commits into
base: main
Choose a base branch
from
4 changes: 2 additions & 2 deletions Main.py
Original file line number Diff line number Diff line change
Expand Up @@ -290,10 +290,10 @@ def precollect_hint(location: Location, auto_status: HintStatus):
f"{locations_data[location.player][location.address]}")
locations_data[location.player][location.address] = \
location.item.code, location.item.player, location.item.flags
auto_status = HintStatus.HINT_AVOID if location.item.trap else HintStatus.HINT_PRIORITY
auto_status = HintStatus.HINT_PRIORITY_AVOID if location.item.trap else HintStatus.HINT_PRIORITY_PRIORITY
if location.name in multiworld.worlds[location.player].options.start_location_hints:
if not location.item.trap: # Unspecified status for location hints, except traps
auto_status = HintStatus.HINT_UNSPECIFIED
auto_status = HintStatus.HINT_PRIORITY_UNSPECIFIED
precollect_hint(location, auto_status)
elif location.item.name in multiworld.worlds[location.item.player].options.start_hints:
precollect_hint(location, auto_status)
Expand Down
69 changes: 35 additions & 34 deletions MultiServer.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@
import Utils
from Utils import version_tuple, restricted_loads, Version, async_start, get_intended_text
from NetUtils import Endpoint, ClientStatus, NetworkItem, decode, encode, NetworkPlayer, Permission, NetworkSlot, \
SlotType, LocationStore, Hint, HintStatus
SlotType, LocationStore, Hint, HintStatus, status_names_brackets
from BaseClasses import ItemClassification

min_client_version = Version(0, 1, 6)
Expand Down Expand Up @@ -1124,7 +1124,7 @@ def register_location_checks(ctx: Context, team: int, slot: int, locations: typi
ctx.save()


def collect_hints(ctx: Context, team: int, slot: int, item: typing.Union[int, str], auto_status: HintStatus) \
def collect_hints(ctx: Context, team: int, slot: int, item: typing.Union[int, str], auto_priority: HintStatus) \
-> typing.List[Hint]:
hints = []
slots: typing.Set[int] = {slot}
Expand All @@ -1141,13 +1141,17 @@ def collect_hints(ctx: Context, team: int, slot: int, item: typing.Union[int, st
else:
found = location_id in ctx.location_checks[team, finding_player]
entrance = ctx.er_hint_data.get(finding_player, {}).get(location_id, "")
new_status = auto_status
if found:
new_status = HintStatus.HINT_FOUND
elif item_flags & ItemClassification.trap:
new_status = HintStatus.HINT_AVOID
hints.append(Hint(receiving_player, finding_player, location_id, item_id, found, entrance,
item_flags, new_status))

is_pure_trap = (
ItemClassification.trap in item_flags
and not ItemClassification.useful in item_flags
Copy link
Contributor

@EmilyV99 EmilyV99 Mar 10, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The fact that useful trap is in many cases used to mark a trap as "never exclude", would still make these wanted to be avoided. Then again, if someone uses the flag to mean "actually useful", then this would not be desired. Neither way here is correct, but, that's entirely the fault of the never_exclude/useful bullshit, so, 🤷‍♀️ I guess this is fine

Also if we go with the change of "default to unspecified, add commands for other priorities", then this is obsolete anyway- which I think is a change strongly worth making.

and not ItemClassification.progression in item_flags
)
if is_pure_trap:
auto_priority = HintStatus.HINT_PRIORITY_AVOID
new_status = HintStatus.from_found_and_priority(found, auto_priority)

hints.append(Hint(receiving_player, finding_player, location_id, item_id, entrance, item_flags, new_status))

return hints

Expand All @@ -1158,34 +1162,31 @@ def collect_hint_location_name(ctx: Context, team: int, slot: int, location: str
return collect_hint_location_id(ctx, team, slot, seeked_location, auto_status)


def collect_hint_location_id(ctx: Context, team: int, slot: int, seeked_location: int, auto_status: HintStatus) \
def collect_hint_location_id(ctx: Context, team: int, slot: int, seeked_location: int, auto_priority: HintStatus) \
-> typing.List[Hint]:
prev_hint = ctx.get_hint(team, slot, seeked_location)
if prev_hint:
return [prev_hint]
result = ctx.locations[slot].get(seeked_location, (None, None, None))
if any(result):
item_id, receiving_player, item_flags = result

found = seeked_location in ctx.location_checks[team, slot]
entrance = ctx.er_hint_data.get(slot, {}).get(seeked_location, "")
new_status = auto_status
if found:
new_status = HintStatus.HINT_FOUND
elif item_flags & ItemClassification.trap:
new_status = HintStatus.HINT_AVOID
return [Hint(receiving_player, slot, seeked_location, item_id, found, entrance, item_flags,
new_status)]

is_pure_trap = (
ItemClassification.trap in item_flags
and not ItemClassification.useful in item_flags
and not ItemClassification.progression in item_flags
)
Comment on lines +1176 to +1180
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe this could be added to ItemClassification as a property? The same 5 lines are duplicated at two places in this file.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I like that idea

if is_pure_trap:
auto_priority = HintStatus.HINT_PRIORITY_AVOID

new_status = HintStatus.from_found_and_priority(found, auto_priority)

return [Hint(receiving_player, slot, seeked_location, item_id, entrance, item_flags, new_status)]
return []


status_names: typing.Dict[HintStatus, str] = {
HintStatus.HINT_FOUND: "(found)",
HintStatus.HINT_UNSPECIFIED: "(unspecified)",
HintStatus.HINT_NO_PRIORITY: "(no priority)",
HintStatus.HINT_AVOID: "(avoid)",
HintStatus.HINT_PRIORITY: "(priority)",
}
def format_hint(ctx: Context, team: int, hint: Hint) -> str:
text = f"[Hint]: {ctx.player_names[team, hint.receiving_player]}'s " \
f"{ctx.item_names[ctx.slot_info[hint.receiving_player].game][hint.item]} is " \
Expand All @@ -1195,7 +1196,7 @@ def format_hint(ctx: Context, team: int, hint: Hint) -> str:
if hint.entrance:
text += f" at {hint.entrance}"

return text + ". " + status_names.get(hint.status, "(unknown)")
return text + ". " + status_names_brackets.get(hint.status.as_display_status(), "(unknown)")


def json_format_send_event(net_item: NetworkItem, receiving_player: int):
Expand Down Expand Up @@ -1599,7 +1600,7 @@ def _cmd_getitem(self, item_name: str) -> bool:
def get_hints(self, input_text: str, for_location: bool = False) -> bool:
points_available = get_client_points(self.ctx, self.client)
cost = self.ctx.get_hint_cost(self.client.slot)
auto_status = HintStatus.HINT_UNSPECIFIED if for_location else HintStatus.HINT_PRIORITY
auto_status = HintStatus.HINT_PRIORITY_UNSPECIFIED if for_location else HintStatus.HINT_PRIORITY_PRIORITY
if not input_text:
hints = {hint.re_check(self.ctx, self.client.team) for hint in
self.ctx.hints[self.client.team, self.client.slot]}
Expand Down Expand Up @@ -1935,7 +1936,7 @@ async def process_client_cmd(ctx: Context, client: Client, args: dict):
target_item, target_player, flags = ctx.locations[client.slot][location]
if create_as_hint:
hints.extend(collect_hint_location_id(ctx, client.team, client.slot, location,
HintStatus.HINT_UNSPECIFIED))
HintStatus.HINT_PRIORITY_UNSPECIFIED))
locs.append(NetworkItem(target_item, location, target_player, flags))
ctx.notify_hints(client.team, hints, only_new=create_as_hint == 2)
if locs and create_as_hint:
Expand Down Expand Up @@ -1975,7 +1976,7 @@ async def process_client_cmd(ctx: Context, client: Client, args: dict):
[{'cmd': 'InvalidPacket', "type": "arguments",
"text": 'UpdateHint: Cannot manually update status to "HINT_FOUND"', "original_cmd": cmd}])
return
new_hint = new_hint.re_prioritize(ctx, status)
new_hint = new_hint.re_prioritize(status)
if hint == new_hint:
return
ctx.replace_hint(client.team, hint.finding_player, hint, new_hint)
Expand Down Expand Up @@ -2289,9 +2290,9 @@ def _cmd_hint(self, player_name: str, *item_name: str) -> bool:
hints = []
for item_name_from_group in self.ctx.item_name_groups[game][item]:
if item_name_from_group in self.ctx.item_names_for_game(game): # ensure item has an ID
hints.extend(collect_hints(self.ctx, team, slot, item_name_from_group, HintStatus.HINT_PRIORITY))
hints.extend(collect_hints(self.ctx, team, slot, item_name_from_group, HintStatus.HINT_PRIORITY_PRIORITY))
else: # item name or id
hints = collect_hints(self.ctx, team, slot, item, HintStatus.HINT_PRIORITY)
hints = collect_hints(self.ctx, team, slot, item, HintStatus.HINT_PRIORITY_PRIORITY)

if hints:
self.ctx.notify_hints(team, hints)
Expand Down Expand Up @@ -2326,16 +2327,16 @@ def _cmd_hint_location(self, player_name: str, *location_name: str) -> bool:
if usable:
if isinstance(location, int):
hints = collect_hint_location_id(self.ctx, team, slot, location,
HintStatus.HINT_UNSPECIFIED)
HintStatus.HINT_PRIORITY_UNSPECIFIED)
elif game in self.ctx.location_name_groups and location in self.ctx.location_name_groups[game]:
hints = []
for loc_name_from_group in self.ctx.location_name_groups[game][location]:
if loc_name_from_group in self.ctx.location_names_for_game(game):
hints.extend(collect_hint_location_name(self.ctx, team, slot, loc_name_from_group,
HintStatus.HINT_UNSPECIFIED))
HintStatus.HINT_PRIORITY_UNSPECIFIED))
else:
hints = collect_hint_location_name(self.ctx, team, slot, location,
HintStatus.HINT_UNSPECIFIED)
HintStatus.HINT_PRIORITY_UNSPECIFIED)
if hints:
self.ctx.notify_hints(team, hints)
else:
Expand Down
147 changes: 112 additions & 35 deletions NetUtils.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,87 @@
from Utils import ByValue, Version


class HintStatus(ByValue, enum.IntEnum):
HINT_UNSPECIFIED = 0
HINT_NO_PRIORITY = 10
HINT_AVOID = 20
HINT_PRIORITY = 30
HINT_FOUND = 40
class HintStatus(ByValue, enum.IntFlag):
# Lower 5 bits: Priorities as integers
HINT_PRIORITY_UNSPECIFIED = 0 # For readable code
HINT_PRIORITY_NO_PRIORITY = 1
HINT_PRIORITY_AVOID = 2
HINT_PRIORITY_PRIORITY = 3
PRIORITY_MASK = 0b11111

# Bits 6+: Technical status
HINT_FOUND = 0b10000000
Comment on lines +15 to +23
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So if I understand correctly, bits 0-4 are used for priority (encoded as integer, not as bit flag), bits 5-6 are unused and bit 7 is for found status encoded as a bit flag?

I get that we want to be efficient with the multiserver memory, but that looks like a very complex encoding to save 1 boolean per hint... This has probably been discussed already, I missed the discord conversations lately. Feel free to dismiss.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have plans for bits 5 and 6 fwiw, but yes, it's all about memory.


# For "compatibility", only used in CommonClient, thus should not affect memory for MultiServer
HINT_UNFOUND_LEGACY = OLD_HINT_FORMAT = 0b1000000000000000
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What would you think of using bit shifts here? I think it is easier to understand that the bit 15 is set with something like

Suggested change
HINT_UNFOUND_LEGACY = OLD_HINT_FORMAT = 0b1000000000000000
HINT_UNFOUND_LEGACY = OLD_HINT_FORMAT = 1 << 15

(same comment for HINT_FOUND)


@property
def priority(self):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You might want to add the return types here, same for the other methods.

Suggested change
def priority(self):
def priority(self) -> HintStatus:

Copy link
Member Author

@NewSoupVi NewSoupVi Mar 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you do that? I thought this causes a problem and you need typing.Self for it or sth

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I guess you're right. Looking at IntFlag, the implementation of & does return Self, we should do the same here.

See https://docs.python.org/3/library/typing.html#typing.Self

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well, that feature was introduced in Python 3.11, and we still support 3.10 :D

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We have typing_extensions tho, so you can use typing_extensions.Self

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's true. I wonder if there are other instances of this in this file, if there are I feel like that should maybe be its own PR, if not then I can add it for sure

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Typing in this file seems not exactly consistent. Some methods are typed correctly, some are missing return types, some are missing args... I would not hold on adding typing to the methods of Hint/HintStatus because of that since you already changed those lines, but that's your call.

return self & HintStatus.PRIORITY_MASK

def with_priority(self, priority: HintStatus):
return (self & ~HintStatus.PRIORITY_MASK) | (priority & HintStatus.PRIORITY_MASK)

@property
def found(self):
return HintStatus.HINT_FOUND in self

def with_found(self, found: bool):
return (self & ~HintStatus.HINT_FOUND) | (found * HintStatus.HINT_FOUND)

@classmethod
def from_found_and_priority(cls, found: bool, priority: HintStatus):
return cls((HintStatus.HINT_FOUND * found) | priority)

def as_display_status(self):
if self.found:
return HintStatus.HINT_FOUND
if HintStatus.OLD_HINT_FORMAT in self:
return HintStatus.HINT_UNFOUND_LEGACY

# Make sure we're only returning statuses that are actually displayable
return next(
(priority for priority in status_names if priority == self.priority),
self.HINT_PRIORITY_UNSPECIFIED
)

@property
def changeable(self):
return not self.found and not HintStatus.OLD_HINT_FORMAT in self


status_names_brackets: typing.Dict[HintStatus, str] = {
HintStatus.HINT_FOUND: "(found)",
HintStatus.HINT_PRIORITY_PRIORITY: "(priority)",
HintStatus.HINT_PRIORITY_AVOID: "(avoid)",
HintStatus.HINT_PRIORITY_NO_PRIORITY: "(no priority)",
HintStatus.HINT_PRIORITY_UNSPECIFIED: "(unspecified)",
HintStatus.HINT_UNFOUND_LEGACY: "(not found)"
}
status_names: typing.Dict[HintStatus, str] = {
HintStatus.HINT_FOUND: "Found",
HintStatus.HINT_PRIORITY_UNSPECIFIED: "Unspecified",
HintStatus.HINT_PRIORITY_NO_PRIORITY: "No Priority",
HintStatus.HINT_PRIORITY_AVOID: "Avoid",
HintStatus.HINT_PRIORITY_PRIORITY: "Priority",
HintStatus.HINT_UNFOUND_LEGACY: "Not Found"
}
status_colors: typing.Dict[HintStatus, str] = {
HintStatus.HINT_FOUND: "green",
HintStatus.HINT_PRIORITY_PRIORITY: "plum",
HintStatus.HINT_PRIORITY_AVOID: "salmon",
HintStatus.HINT_PRIORITY_NO_PRIORITY: "slateblue",
HintStatus.HINT_PRIORITY_UNSPECIFIED: "white",
HintStatus.HINT_UNFOUND_LEGACY: "red",
}
status_sort_weights: dict[HintStatus, int] = {
HintStatus.HINT_FOUND: 0,
HintStatus.HINT_PRIORITY_UNSPECIFIED: 1,
HintStatus.HINT_PRIORITY_NO_PRIORITY: 2,
HintStatus.HINT_PRIORITY_AVOID: 3,
HintStatus.HINT_PRIORITY_PRIORITY: 4,
HintStatus.HINT_UNFOUND_LEGACY: 5,
}


class JSONMessagePart(typing.TypedDict, total=False):
Expand Down Expand Up @@ -278,7 +353,7 @@ def _handle_entrance_name(self, node: JSONMessagePart):
return self._handle_color(node)

def _handle_hint_status(self, node: JSONMessagePart):
node["color"] = status_colors.get(node["hint_status"], "red")
node["color"] = status_colors[HintStatus(node["hint_status"]).as_display_status()]
return self._handle_color(node)


Expand Down Expand Up @@ -313,51 +388,37 @@ def add_json_location(parts: list, location_id: int, player: int = 0, **kwargs)
parts.append({"text": str(location_id), "player": player, "type": JSONTypes.location_id, **kwargs})


status_names: typing.Dict[HintStatus, str] = {
HintStatus.HINT_FOUND: "(found)",
HintStatus.HINT_UNSPECIFIED: "(unspecified)",
HintStatus.HINT_NO_PRIORITY: "(no priority)",
HintStatus.HINT_AVOID: "(avoid)",
HintStatus.HINT_PRIORITY: "(priority)",
}
status_colors: typing.Dict[HintStatus, str] = {
HintStatus.HINT_FOUND: "green",
HintStatus.HINT_UNSPECIFIED: "white",
HintStatus.HINT_NO_PRIORITY: "slateblue",
HintStatus.HINT_AVOID: "salmon",
HintStatus.HINT_PRIORITY: "plum",
}


def add_json_hint_status(parts: list, hint_status: HintStatus, text: typing.Optional[str] = None, **kwargs):
parts.append({"text": text if text != None else status_names.get(hint_status, "(unknown)"),
"hint_status": hint_status, "type": JSONTypes.hint_status, **kwargs})
if text is None:
text = status_names_brackets[hint_status.as_display_status()]
parts.append({"text": text, "hint_status": hint_status, "type": JSONTypes.hint_status, **kwargs})


class Hint(typing.NamedTuple):
receiving_player: int
finding_player: int
location: int
item: int
found: bool
entrance: str = ""
item_flags: int = 0
status: HintStatus = HintStatus.HINT_UNSPECIFIED
status: HintStatus = HintStatus.from_found_and_priority(False, HintStatus.HINT_PRIORITY_UNSPECIFIED)

@property
def found(self):
return self.status.found

def re_check(self, ctx, team) -> Hint:
if self.found and self.status == HintStatus.HINT_FOUND:
if self.found:
return self
found = self.location in ctx.location_checks[team, self.finding_player]
if found:
return self._replace(found=found, status=HintStatus.HINT_FOUND)
return self._replace(status=self.status.with_found(True))
return self

def re_prioritize(self, ctx, status: HintStatus) -> Hint:
if self.found and status != HintStatus.HINT_FOUND:
status = HintStatus.HINT_FOUND
if status != self.status:
return self._replace(status=status)
return self
def re_prioritize(self, status: HintStatus) -> Hint:
if status.priority == self.status.priority:
return self
return self._replace(status=status.with_priority(status.priority))

def __hash__(self):
return hash((self.receiving_player, self.finding_player, self.location, self.item, self.entrance))
Expand Down Expand Up @@ -390,6 +451,22 @@ def local(self):
return self.receiving_player == self.finding_player


# From what I can tell, all the methods to make unpickling backwards compatible don't work for NamedTuples.
# This is because you can't override __getattributes__ or __new__ on a NamedTuple.
# The only approach I found was to reroute the Unpickler to a dummy class with a custom __new__ in its find_class.
class CompatibleHint:
def __new__(cls, *args, **kwargs):
if type(args[4]) == bool:
# old hint format
legacy_found = args[4]

# No compatibility for old priority system, just revert to unspecified
new_hint_status = HintStatus.from_found_and_priority(legacy_found, HintStatus.HINT_PRIORITY_UNSPECIFIED)

args = (*args[:4], *args[5:-1], new_hint_status)
return Hint.__new__(Hint, *args, **kwargs)


class _LocationStore(dict, typing.MutableMapping[int, typing.Dict[int, typing.Tuple[int, int, int]]]):
def __init__(self, values: typing.MutableMapping[int, typing.Dict[int, typing.Tuple[int, int, int]]]):
super().__init__(values)
Expand Down
2 changes: 2 additions & 0 deletions Utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -430,6 +430,8 @@ def find_class(self, module: str, name: str) -> type:
# used by MultiServer -> savegame/multidata
if module == "NetUtils" and name in {"NetworkItem", "ClientStatus", "Hint",
"SlotType", "NetworkSlot", "HintStatus"}:
if name == "Hint":
name = "CompatibleHint"
return getattr(self.net_utils_module, name)
# Options and Plando are unpickled by WebHost -> Generate
if module == "worlds.generic" and name == "PlandoItem":
Expand Down
Loading
Loading