monorail
/
Archipelago
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'main' into multishop

This commit is contained in:
CaitSith2 2021-01-21 22:45:12 -08:00
parent 6f1fc2ad4e 8d63e13ca7
commit d32acd37a8
1 changed files with 31 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
MultiServer.py
View File

@ -113,12 +113,6 @@ class Context(Node):
self.save_dirty = False self.save_dirty = False
self.tags = ['Berserker'] self.tags = ['Berserker']
self.minimum_client_versions: typing.Dict[typing.Tuple[int, int], Utils.Version] = {} self.minimum_client_versions: typing.Dict[typing.Tuple[int, int], Utils.Version] = {}
self.lookup_items_id_to_name = Items.lookup_id_to_name
self.lookup_items_name_to_id = {value: key for key, value in Items.lookup_id_to_name.items()}
self.lookup_region_id_to_name = Regions.lookup_id_to_name
self.lookup_region_name_to_id = Regions.lookup_name_to_id
self.console_names = console_names
self.item_name_groups = Items.item_name_groups
def load(self, multidatapath: str, use_embedded_server_options: bool = False): def load(self, multidatapath: str, use_embedded_server_options: bool = False):
with open(multidatapath, 'rb') as f: with open(multidatapath, 'rb') as f:
@ -156,30 +150,6 @@ class Context(Node):
server_options = jsonobj.get("server_options", {}) server_options = jsonobj.get("server_options", {})
self._set_options(server_options) self._set_options(server_options)
new_console_names = set()
lookups = {"lookup_items_id_to_name": False, "lookup_region_id_to_name": False, "item_name_groups": False}
if "lookup_items_id_to_name" in jsonobj:
lookups["lookup_items_id_to_name"] = True
self.lookup_items_id_to_name = jsonobj["lookup_items_id_to_name"]
self.lookup_items_name_to_id = {value: key for key, value in self.lookup_items_id_to_name.items()}
new_console_names |= set(self.lookup_items_id_to_name.values())
if "lookup_region_id_to_name" in jsonobj:
lookups["lookup_region_id_to_name"] = True
self.lookup_region_id_to_name = jsonobj["lookup_region_id_to_name"]
self.lookup_region_name_to_id = {value: key for key, value in self.lookup_region_id_to_name.items()}
new_console_names |= set(self.lookup_region_id_to_name.values())
if "item_name_groups" in jsonobj:
lookups["item_name_groups"] = True
self.item_name_groups = {key: set(value) for key, value in jsonobj["item_name_groups"]}
new_console_names |= set(self.item_name_groups.keys())
if not all(lookups.values()):
new_console_names |= self.console_names
self.console_names = frozenset(new_console_names)
def _set_options(self, server_options: dict): def _set_options(self, server_options: dict):
for key, value in server_options.items(): for key, value in server_options.items():
data_type = self.simple_options.get(key, None) data_type = self.simple_options.get(key, None)
@ -542,8 +512,8 @@ def register_location_checks(ctx: Context, team: int, slot: int, locations):
if slot != target_player: if slot != target_player:
ctx.broadcast_team(team, [['ItemSent', (slot, location, target_player, target_item)]]) ctx.broadcast_team(team, [['ItemSent', (slot, location, target_player, target_item)]])
logging.info('(Team #%d) %s sent %s to %s (%s)' % ( logging.info('(Team #%d) %s sent %s to %s (%s)' % (
team + 1, ctx.player_names[(team, slot)], ctx.lookup_items_id_to_name(target_item, f"Unknown item (ID: {target_item})"), team + 1, ctx.player_names[(team, slot)], get_item_name_from_id(target_item),
ctx.player_names[(team, target_player)], ctx.lookup_region_id_to_name.get(location, f"Unknown location (ID: {location})"))) ctx.player_names[(team, target_player)], get_location_name_from_address(location)))
found_items = True found_items = True
elif target_player == slot: # local pickup, notify clients of the pickup elif target_player == slot: # local pickup, notify clients of the pickup
if location not in ctx.location_checks[team, slot]: if location not in ctx.location_checks[team, slot]:
@ -583,7 +553,7 @@ def collect_hints(ctx: Context, team: int, slot: int, item: str) -> typing.List[
def collect_hints_location(ctx: Context, team: int, slot: int, location: str) -> typing.List[Utils.Hint]: def collect_hints_location(ctx: Context, team: int, slot: int, location: str) -> typing.List[Utils.Hint]:
hints = [] hints = []
seeked_location = ctx.lookup_region_name_to_id[location] seeked_location = Regions.lookup_name_to_id[location]
for check, result in ctx.locations.items(): for check, result in ctx.locations.items():
location_id, finding_player = check location_id, finding_player = check
if finding_player == slot and location_id == seeked_location: if finding_player == slot and location_id == seeked_location:
@ -597,8 +567,8 @@ def collect_hints_location(ctx: Context, team: int, slot: int, location: str) ->
def format_hint(ctx: Context, team: int, hint: Utils.Hint) -> str: def format_hint(ctx: Context, team: int, hint: Utils.Hint) -> str:
text = f"[Hint]: {ctx.player_names[team, hint.receiving_player]}'s " \ text = f"[Hint]: {ctx.player_names[team, hint.receiving_player]}'s " \
f"{ctx.lookup_items_id_to_name.get(hint.item, f'Unknown item (ID:{hint.item})')} is " \ f"{Items.lookup_id_to_name[hint.item]} is " \
f"at {ctx.lookup_region_id_to_name.get(hint.location, f'Unknown location (ID:{hint.location})')} " \ f"at {get_location_name_from_address(hint.location)} " \
f"in {ctx.player_names[team, hint.finding_player]}'s World" f"in {ctx.player_names[team, hint.finding_player]}'s World"
if hint.entrance: if hint.entrance:
@ -606,7 +576,7 @@ def format_hint(ctx: Context, team: int, hint: Utils.Hint) -> str:
return text + (". (found)" if hint.found else ".") return text + (". (found)" if hint.found else ".")
def get_intended_text(input_text: str, possible_answers: typing.Iterable[str]) -> typing.Tuple[str, bool, str]: def get_intended_text(input_text: str, possible_answers: typing.Iterable[str]= console_names) -> typing.Tuple[str, bool, str]:
picks = fuzzy_process.extract(input_text, possible_answers, limit=2) picks = fuzzy_process.extract(input_text, possible_answers, limit=2)
if len(picks) > 1: if len(picks) > 1:
dif = picks[0][1] - picks[1][1] dif = picks[0][1] - picks[1][1]
@ -831,26 +801,29 @@ class ClientMessageProcessor(CommonCommandProcessor):
"Your client is too old to send game beaten information. Please update, load you savegame and reconnect.") "Your client is too old to send game beaten information. Please update, load you savegame and reconnect.")
return False return False
def remaining_items(self) -> bool:
remaining_item_ids = get_remaining(self.ctx, self.client.team, self.client.slot)
if remaining_item_ids:
self.output("Remaining items: " + ", ".join(self.ctx.lookup_items_id_to_name.get(item_id, "unknown item")
for item_id in remaining_item_ids))
else:
self.output("No remaining items found.")
return True
def _cmd_remaining(self) -> bool: def _cmd_remaining(self) -> bool:
"""List remaining items in your game, but not their location or recipient""" """List remaining items in your game, but not their location or recipient"""
if self.ctx.remaining_mode == "enabled": if self.ctx.remaining_mode == "enabled":
return self.remaining_items() remaining_item_ids = get_remaining(self.ctx, self.client.team, self.client.slot)
if remaining_item_ids:
self.output("Remaining items: " + ", ".join(Items.lookup_id_to_name.get(item_id, "unknown item")
for item_id in remaining_item_ids))
else:
self.output("No remaining items found.")
return True
elif self.ctx.remaining_mode == "disabled": elif self.ctx.remaining_mode == "disabled":
self.output( self.output(
"Sorry, !remaining has been disabled on this server.") "Sorry, !remaining has been disabled on this server.")
return False return False
else: # is goal else: # is goal
if self.ctx.client_game_state[self.client.team, self.client.slot] == CLIENT_GOAL: if self.ctx.client_game_state[self.client.team, self.client.slot] == CLIENT_GOAL:
return self.remaining_items() remaining_item_ids = get_remaining(self.ctx, self.client.team, self.client.slot)
if remaining_item_ids:
self.output("Remaining items: " + ", ".join(Items.lookup_id_to_name.get(item_id, "unknown item")
for item_id in remaining_item_ids))
else:
self.output("No remaining items found.")
return True
else: else:
self.output( self.output(
"Sorry, !remaining requires you to have beaten the game on this server") "Sorry, !remaining requires you to have beaten the game on this server")
@ -900,9 +873,9 @@ class ClientMessageProcessor(CommonCommandProcessor):
def _cmd_getitem(self, item_name: str) -> bool: def _cmd_getitem(self, item_name: str) -> bool:
"""Cheat in an item, if it is enabled on this server""" """Cheat in an item, if it is enabled on this server"""
if self.ctx.item_cheat: if self.ctx.item_cheat:
item_name, usable, response = get_intended_text(item_name, self.ctx.lookup_items_id_to_name.values()) item_name, usable, response = get_intended_text(item_name, Items.item_table.keys())
if usable: if usable:
new_item = ReceivedItem(self.ctx.lookup_items_name_to_id[item_name], -1, self.client.slot) new_item = ReceivedItem(Items.item_table[item_name][3], -1, self.client.slot)
get_received_items(self.ctx, self.client.team, self.client.slot).append(new_item) get_received_items(self.ctx, self.client.team, self.client.slot).append(new_item)
self.ctx.notify_all('Cheat console: sending "' + item_name + '" to ' + self.ctx.get_aliased_name(self.client.team, self.client.slot)) self.ctx.notify_all('Cheat console: sending "' + item_name + '" to ' + self.ctx.get_aliased_name(self.client.team, self.client.slot))
send_new_items(self.ctx) send_new_items(self.ctx)
@ -927,14 +900,14 @@ class ClientMessageProcessor(CommonCommandProcessor):
notify_hints(self.ctx, self.client.team, list(hints)) notify_hints(self.ctx, self.client.team, list(hints))
return True return True
else: else:
item_name, usable, response = get_intended_text(item_or_location, self.ctx.console_names) item_name, usable, response = get_intended_text(item_or_location)
if usable: if usable:
if item_name in Items.hint_blacklist: if item_name in Items.hint_blacklist:
self.output(f"Sorry, \"{item_name}\" is marked as non-hintable.") self.output(f"Sorry, \"{item_name}\" is marked as non-hintable.")
hints = [] hints = []
elif item_name in self.ctx.item_name_groups: elif item_name in Items.item_name_groups:
hints = [] hints = []
for item in self.ctx.item_name_groups[item_name]: for item in Items.item_name_groups[item_name]:
hints.extend(collect_hints(self.ctx, self.client.team, self.client.slot, item)) hints.extend(collect_hints(self.ctx, self.client.team, self.client.slot, item))
elif item_name in Items.item_table: # item name elif item_name in Items.item_table: # item name
hints = collect_hints(self.ctx, self.client.team, self.client.slot, item_name) hints = collect_hints(self.ctx, self.client.team, self.client.slot, item_name)
@ -995,14 +968,14 @@ class ClientMessageProcessor(CommonCommandProcessor):
def get_checked_checks(ctx: Context, client: Client) -> list: def get_checked_checks(ctx: Context, client: Client) -> list:
return [ctx.lookup_region_id_to_name.get(location_id, f'Unknown Location ID: {location_id}') for return [Regions.lookup_id_to_name.get(location_id, f'Unknown Location ID: {location_id}') for
location_id, slot in ctx.locations if location_id, slot in ctx.locations if
slot == client.slot and slot == client.slot and
location_id in ctx.location_checks[client.team, client.slot]] location_id in ctx.location_checks[client.team, client.slot]]
def get_missing_checks(ctx: Context, client: Client) -> list: def get_missing_checks(ctx: Context, client: Client) -> list:
return [ctx.lookup_region_id_to_name.get(location_id, f'Unknown Location ID: {location_id}') for return [Regions.lookup_id_to_name.get(location_id, f'Unknown Location ID: {location_id}') for
location_id, slot in ctx.locations if location_id, slot in ctx.locations if
slot == client.slot and slot == client.slot and
location_id not in ctx.location_checks[client.team, client.slot]] location_id not in ctx.location_checks[client.team, client.slot]]
@ -1251,7 +1224,7 @@ class ServerCommandProcessor(CommonCommandProcessor):
seeked_player, usable, response = get_intended_text(player_name, self.ctx.player_names.values()) seeked_player, usable, response = get_intended_text(player_name, self.ctx.player_names.values())
if usable: if usable:
item = " ".join(item_name) item = " ".join(item_name)
item, usable, response = get_intended_text(item, self.ctx.lookup_items_id_to_name.values()) item, usable, response = get_intended_text(item, Items.item_table.keys())
if usable: if usable:
for client in self.ctx.endpoints: for client in self.ctx.endpoints:
if client.name == seeked_player: if client.name == seeked_player:
@ -1274,11 +1247,11 @@ class ServerCommandProcessor(CommonCommandProcessor):
for (team, slot), name in self.ctx.player_names.items(): for (team, slot), name in self.ctx.player_names.items():
if name == seeked_player: if name == seeked_player:
item = " ".join(item_or_location) item = " ".join(item_or_location)
item, usable, response = get_intended_text(item, self.ctx.console_names) item, usable, response = get_intended_text(item)
if usable: if usable:
if item in self.ctx.item_name_groups: if item in Items.item_name_groups:
hints = [] hints = []
for item in self.ctx.item_name_groups[item]: for item in Items.item_name_groups[item]:
hints.extend(collect_hints(self.ctx, team, slot, item)) hints.extend(collect_hints(self.ctx, team, slot, item))
elif item in Items.item_table: # item name elif item in Items.item_table: # item name
hints = collect_hints(self.ctx, team, slot, item) hints = collect_hints(self.ctx, team, slot, item)