about summary refs log tree commit diff stats
path: root/tools/validator/structs.h
blob: d1d45f27feb47b118581d6f64f751fe08d9ef645 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
#ifndef TOOLS_VALIDATOR_STRUCTS_H_
#define TOOLS_VALIDATOR_STRUCTS_H_

#include <map>
#include <string>
#include <vector>

#include "proto/human.pb.h"
#include "util/identifiers.h"

namespace com::fourisland::lingo2_archipelago {

struct MalformedIdentifiers {
  std::vector<PaintingIdentifier> paintings;
  std::vector<PanelIdentifier> panels;
  std::vector<KeyholderIdentifier> keyholders;

  bool HasAny() const {
    return !paintings.empty() || !panels.empty() || !keyholders.empty();
  }
};

struct GameNodeInfo {
  bool defined = false;
  int uses = 0;
};

struct MapInfo {
  std::map<std::string, GameNodeInfo> game_nodes;
};

struct RoomInfo {
  std::vector<HumanRoom> definitions;

  std::vector<DoorIdentifier> doors_referenced_by;
  std::vector<PanelIdentifier> panels_referenced_by;
  std::vector<HumanConnection> connections_referenced_by;
};

struct DoorInfo {
  std::vector<HumanDoor> definitions;
  bool has_id = false;

  std::vector<HumanConnection> connections_referenced_by;
  std::vector<DoorIdentifier> doors_referenced_by;
  std::vector<PanelIdentifier> panels_referenced_by;
  std::vector<PaintingIdentifier> paintings_referenced_by;
  std::vector<PortIdentifier> ports_referenced_by;
  std::vector<std::string> progressives_referenced_by;
  std::vector<std::string> door_groups_referenced_by;

  MalformedIdentifiers malformed_identifiers;
};

struct PortInfo {
  std::vector<HumanPort> definitions;

  std::vector<HumanConnection> connections_referenced_by;
  std::vector<HumanConnection> target_connections_referenced_by;
};

struct PaintingInfo {
  std::vector<HumanPainting> definitions;

  std::vector<HumanConnection> connections_referenced_by;
  std::vector<HumanConnection> target_connections_referenced_by;
  std::vector<DoorIdentifier> doors_referenced_by;
};

struct ProxyInfo {
  std::vector<Proxy> definitions;

  std::vector<HumanConnection> connections_referenced_by;
  std::vector<DoorIdentifier> doors_referenced_by;
};

struct PanelInfo {
  std::vector<HumanPanel> definitions;
  bool has_id = false;

  std::string map_area_name;

  std::vector<HumanConnection> connections_referenced_by;
  std::vector<HumanConnection> target_connections_referenced_by;
  std::vector<DoorIdentifier> doors_referenced_by;

  std::map<std::string, ProxyInfo> proxies;
};

struct KeyholderInfo {
  std::vector<HumanKeyholder> definitions;
  bool has_id = false;

  std::vector<DoorIdentifier> doors_referenced_by;
};

using LetterIdentifier = std::tuple<char, bool>;

struct LetterInfo {
  std::vector<RoomIdentifier> defined_in;
  bool has_id = false;
};

struct EndingInfo {
  std::vector<RoomIdentifier> defined_in;
  bool has_id = false;

  std::vector<DoorIdentifier> doors_referenced_by;
};

struct PanelNameInfo {
  std::vector<PanelIdentifier> panels_used_by;
};

struct ProgressiveInfo {
  std::vector<HumanProgressive> definitions;
  bool has_id = false;

  std::vector<DoorIdentifier> malformed_doors;
};

struct DoorGroupInfo {
  std::vector<HumanDoorGroup> definitions;
  bool has_id = false;

  std::vector<DoorIdentifier> malformed_doors;
};

struct CollectedInfo {
  std::map<std::string, MapInfo> maps;
  std::map<RoomIdentifier, RoomInfo, RoomIdentifierLess> rooms;
  std::map<DoorIdentifier, DoorInfo, DoorIdentifierLess> doors;
  std::map<PortIdentifier, PortInfo, PortIdentifierLess> ports;
  std::map<PaintingIdentifier, PaintingInfo, PaintingIdentifierLess> paintings;
  std::map<PanelIdentifier, PanelInfo, PanelIdentifierLess> panels;
  std::map<KeyholderIdentifier, KeyholderInfo, KeyholderIdentifierLess>
      keyholders;
  std::map<LetterIdentifier, LetterInfo> letters;
  std::map<std::string, EndingInfo> endings;
  std::map<std::string, PanelNameInfo> panel_names;
  std::map<std::string, ProgressiveInfo> progressives;
  std::map<std::string, DoorGroupInfo> door_groups;
};

}  // namespace com::fourisland::lingo2_archipelago

#endif /* TOOLS_VALIDATOR_STRUCTS_H_ */
Context" client_ctx: "Lingo2ClientContext" tracker: Tracker keyboard: dict[str, int] worldports: set[int] goaled: bool def __init__(self, game_ctx: "Lingo2GameContext", client_ctx: "Lingo2ClientContext"): self.game_ctx = game_ctx self.game_ctx.manager = self self.client_ctx = client_ctx self.client_ctx.manager = self self.tracker = Tracker(self) self.keyboard = {} self.worldports = set() self.reset() def reset(self): for k in ALL_LETTERS: self.keyboard[k] = 0 self.worldports = set() self.goaled = False def update_keyboard(self, new_keyboard: dict[str, int]) -> dict[str, int]: ret: dict[str, int] = {} for k, v in new_keyboard.items(): if v > self.keyboard.get(k, 0): self.keyboard[k] = v ret[k] = v if len(ret) > 0: self.tracker.refresh_state() self.game_ctx.send_accessible_locations() return ret def update_worldports(self, new_worldports: set[int]) -> set[int]: ret = new_worldports.difference(self.worldports) self.worldports.update(new_worldports) if len(ret) > 0: self.tracker.refresh_state() self.game_ctx.send_accessible_locations() return ret class Lingo2GameContext: server: Endpoint | None manager: Lingo2Manager def __init__(self): self.server = None def send_connected(self): if self.server is None: return msg = { "cmd": "Connected", "user": self.manager.client_ctx.username, "seed_name": self.manager.client_ctx.seed_name, "version": self.manager.client_ctx.server_version, "generator_version": self.manager.client_ctx.generator_version, "team": self.manager.client_ctx.team, "slot": self.manager.client_ctx.slot, "checked_locations": self.manager.client_ctx.checked_locations, "slot_data": self.manager.client_ctx.slot_data, } async_start(self.send_msgs([msg]), name="game Connected") def send_connection_refused(self, text): if self.server is None: return msg = { "cmd": "ConnectionRefused", "text": text, } async_start(self.send_msgs([msg]), name="game ConnectionRefused") def send_item_sent_notification(self, item_name, receiver_name, item_flags): if self.server is None: return msg = { "cmd": "ItemSentNotif", "item_name": item_name, "receiver_name": receiver_name, "item_flags": item_flags, } async_start(self.send_msgs([msg]), name="item sent notif") def send_hint_received(self, item_name, location_name, receiver_name, item_flags, for_self): if self.server is None: return msg = { "cmd": "HintReceived", "item_name": item_name, "location_name": location_name, "receiver_name": receiver_name, "item_flags": item_flags, "self": int(for_self), } async_start(self.send_msgs([msg]), name="hint received notif") def send_item_received(self, items): if self.server is None: return msg = { "cmd": "ItemReceived", "items": items, } async_start(self.send_msgs([msg]), name="item received") def send_location_info(self, locations): if self.server is None: return msg = { "cmd": "LocationInfo", "locations": locations, } async_start(self.send_msgs([msg]), name="location info") def send_text_message(self, parts): if self.server is None: return msg = { "cmd": "TextMessage", "data": parts, } async_start(self.send_msgs([msg]), name="notif") def send_accessible_locations(self): if self.server is None: return msg = { "cmd": "AccessibleLocations", "locations": list(self.manager.tracker.accessible_locations), } if len(self.manager.tracker.accessible_worldports) > 0: msg["worldports"] = list(self.manager.tracker.accessible_worldports) if self.manager.tracker.goal_accessible and not self.manager.goaled: msg["goal"] = True async_start(self.send_msgs([msg]), name="accessible locations") def send_update_locations(self, locations): if self.server is None: return msg = { "cmd": "UpdateLocations", "locations": locations, } async_start(self.send_msgs([msg]), name="update locations") def send_update_keyboard(self, updates): if self.server is None: return msg = { "cmd": "UpdateKeyboard", "updates": updates, } async_start(self.send_msgs([msg]), name="update keyboard") def send_update_worldports(self, worldports): if self.server is None: return msg = { "cmd": "UpdateWorldports", "worldports": worldports, } async_start(self.send_msgs([msg]), name="update worldports") def send_path_reply(self, object_type: str, object_id: int | None, path: list[str]): if self.server is None: return msg = { "cmd": "PathReply", "type": object_type, "path": path, } if object_id is not None: msg["id"] = object_id async_start(self.send_msgs([msg]), name="path reply") async def send_msgs(self, msgs: list[Any]) -> None: """ `msgs` JSON serializable """ if not self.server or not self.server.socket.open or self.server.socket.closed: return await self.server.socket.send(encode(msgs)) class Lingo2ClientContext(CommonContext): manager: Lingo2Manager game = "Lingo 2" items_handling = 0b111 slot_data: dict[str, Any] | None victory_data_storage_key: str def __init__(self, server_address: str | None = None, password: str | None = None): super().__init__(server_address, password) def make_gui(self): ui = super().make_gui() ui.base_title = "Archipelago Lingo 2 Client" return ui async def server_auth(self, password_requested: bool = False): if password_requested and not self.password: self.manager.game_ctx.send_connection_refused("Invalid password.") else: self.auth = self.username await self.send_connect() def handle_connection_loss(self, msg: str): super().handle_connection_loss(msg) exc_info = sys.exc_info() self.manager.game_ctx.send_connection_refused(str(exc_info[1])) def on_package(self, cmd: str, args: dict): if cmd == "RoomInfo": self.seed_name = args.get("seed_name", None) elif cmd == "Connected": self.slot_data = args.get("slot_data", None) self.manager.reset() self.manager.game_ctx.send_connected() self.manager.tracker.setup_slot(self.slot_data) self.manager.tracker.set_checked_locations(self.checked_locations) self.manager.game_ctx.send_accessible_locations() self.victory_data_storage_key = f"_read_client_status_{self.team}_{self.slot}" self.set_notify(self.get_datastorage_key("keyboard1"), self.get_datastorage_key("keyboard2"), self.victory_data_storage_key) msg_batch = [{ "cmd": "Set", "key": self.get_datastorage_key("keyboard1"), "default": 0, "want_reply": True, "operations": [{"operation": "default", "value": 0}] }, { "cmd": "Set", "key": self.get_datastorage_key("keyboard2"), "default": 0, "want_reply": True, "operations": [{"operation": "default", "value": 0}] }] if self.slot_data.get("shuffle_worldports", False): self.set_notify(self.get_datastorage_key("worldports")) msg_batch.append({ "cmd": "Set", "key": self.get_datastorage_key("worldports"), "default": [], "want_reply": True, "operations": [{"operation": "default", "value": []}] }) async_start(self.send_msgs(msg_batch), name="default keys") elif cmd == "RoomUpdate": if "checked_locations" in args: self.manager.tracker.set_checked_locations(self.checked_locations) self.manager.game_ctx.send_update_locations(args["checked_locations"]) elif cmd == "ReceivedItems": self.manager.tracker.set_collected_items(self.items_received) cur_index = 0 items = [] for item in args["items"]: index = cur_index + args["index"] cur_index += 1 item_msg = { "id": item.item, "index": index, "flags": item.flags, "text": self.item_names.lookup_in_slot(item.item, self.slot), } if item.player != self.slot: item_msg["sender"] = self.player_names.get(item.player) items.append(item_msg) self.manager.game_ctx.send_item_received(items) if any(ItemClassification.progression in ItemClassification(item.flags) for item in args["items"]): self.manager.game_ctx.send_accessible_locations() elif cmd == "PrintJSON": if "receiving" in args and "item" in args and args["item"].player == self.slot: item_name = self.item_names.lookup_in_slot(args["item"].item, args["receiving"]) location_name = self.location_names.lookup_in_slot(args["item"].location, args["item"].player) receiver_name = self.player_names.get(args["receiving"]) if args["type"] == "Hint" and not args.get("found", False): self.manager.game_ctx.send_hint_received(item_name, location_name, receiver_name, args["item"].flags, int(args["receiving"]) == self.slot) elif args["receiving"] != self.slot: self.manager.game_ctx.send_item_sent_notification(item_name, receiver_name, args["item"].flags) parts = [] for message_part in args["data"]: if "type" not in message_part and "text" in message_part: parts.append({"type": "text", "text": message_part["text"]}) elif message_part["type"] == "player_id": parts.append({ "type": "player", "text": self.player_names.get(int(message_part["text"])), "self": int(int(message_part["text"]) == self.slot), }) elif message_part["type"] == "item_id": parts.append({ "type": "item", "text": self.item_names.lookup_in_slot(int(message_part["text"]), message_part["player"]), "flags": message_part["flags"], }) elif message_part["type"] == "location_id": parts.append({ "type": "location", "text": self.location_names.lookup_in_slot(int(message_part["text"]), message_part["player"]) }) elif "text" in message_part: parts.append({"type": "text", "text": message_part["text"]}) self.manager.game_ctx.send_text_message(parts) elif cmd == "LocationInfo": locations = [] for location in args["locations"]: locations.append({ "id": location.location, "item": self.item_names.lookup_in_slot(location.item, location.player), "player": self.player_names.get(location.player), "flags": location.flags, "self": int(location.player) == self.slot, }) self.manager.game_ctx.send_location_info(locations) elif cmd == "Retrieved": for k, v in args["keys"].items(): if k == self.victory_data_storage_key: self.handle_status_update(v) elif cmd == "SetReply": if args["key"] == self.get_datastorage_key("keyboard1"): self.handle_keyboard_update(1, args) elif args["key"] == self.get_datastorage_key("keyboard2"): self.handle_keyboard_update(2, args) elif args["key"] == self.get_datastorage_key("worldports"): updates = self.manager.update_worldports(set(args["value"])) if len(updates) > 0: self.manager.game_ctx.send_update_worldports(updates) elif args["key"] == self.victory_data_storage_key: self.handle_status_update(args["value"]) def get_datastorage_key(self, name: str): return f"Lingo2_{self.slot}_{name}" async def update_keyboard(self, updates: dict[str, int]): kb1 = 0 kb2 = 0 for k, v in updates.items(): if v == 0: continue effect = 0 if v >= 1: effect |= 1 if v == 2: effect |= 2 pos = KEY_STORAGE_MAPPING[k] if pos[0] == 1: kb1 |= (effect << pos[1] * 2) else: kb2 |= (effect << pos[1] * 2) msgs = [] if kb1 != 0: msgs.append({ "cmd": "Set", "key": self.get_datastorage_key("keyboard1"), "want_reply": True, "operations": [{ "operation": "or", "value": kb1 }] }) if kb2 != 0: msgs.append({ "cmd": "Set", "key": self.get_datastorage_key("keyboard2"), "want_reply": True, "operations": [{ "operation": "or", "value": kb2 }] }) if len(msgs) > 0: await self.send_msgs(msgs) def handle_keyboard_update(self, field: int, args: dict[str, Any]): keys = {} value = args["value"] for i in range(0, 13): if (value & (1 << (i * 2))) != 0: keys[REVERSE_KEY_STORAGE_MAPPING[(field, i)]] = 1 if (value & (1 << (i * 2 + 1))) != 0: keys[REVERSE_KEY_STORAGE_MAPPING[(field, i)]] = 2 updates = self.manager.update_keyboard(keys) if len(updates) > 0: self.manager.game_ctx.send_update_keyboard(updates) async def update_worldports(self, updates: set[int]): await self.send_msgs([{ "cmd": "Set", "key": self.get_datastorage_key("worldports"), "want_reply": True, "operations": [{ "operation": "update", "value": updates }] }]) def handle_status_update(self, value: int): self.manager.goaled = (value == ClientStatus.CLIENT_GOAL) self.manager.tracker.refresh_state() self.manager.game_ctx.send_accessible_locations() async def pipe_loop(manager: Lingo2Manager): while not manager.client_ctx.exit_event.is_set(): try: socket = await websockets.connect("ws://localhost", port=PORT, ping_timeout=None, ping_interval=None, max_size=MESSAGE_MAX_SIZE) manager.game_ctx.server = Endpoint(socket) logger.info("Connected to Lingo 2!") if manager.client_ctx.auth is not None: manager.game_ctx.send_connected() manager.game_ctx.send_accessible_locations() async for data in manager.game_ctx.server.socket: for msg in decode(data): await process_game_cmd(manager, msg) except ConnectionRefusedError: logger.info("Could not connect to Lingo 2.") finally: manager.game_ctx.server = None async def process_game_cmd(manager: Lingo2Manager, args: dict): cmd = args["cmd"] if cmd == "Connect": manager.client_ctx.seed_name = None server = args.get("server") player = args.get("player") password = args.get("password") if password != "": server_address = f"{player}:{password}@{server}" else: server_address = f"{player}:None@{server}" async_start(manager.client_ctx.connect(server_address), name="client connect") elif cmd == "Disconnect": manager.client_ctx.seed_name = None async_start(manager.client_ctx.disconnect(), name="client disconnect") elif cmd in ["Sync", "LocationChecks", "Say", "StatusUpdate", "LocationScouts"]: async_start(manager.client_ctx.send_msgs([args]), name="client forward") elif cmd == "UpdateKeyboard": updates = manager.update_keyboard(args["keyboard"]) if len(updates) > 0: async_start(manager.client_ctx.update_keyboard(updates), name="client update keyboard") elif cmd == "CheckWorldport": port_id = args["port_id"] worldports = {port_id} if str(port_id) in manager.client_ctx.slot_data["port_pairings"]: worldports.add(manager.client_ctx.slot_data["port_pairings"][str(port_id)]) updates = manager.update_worldports(worldports) if len(updates) > 0: async_start(manager.client_ctx.update_worldports(updates), name="client update worldports") manager.game_ctx.send_update_worldports(updates) elif cmd == "GetPath": path = None if args["type"] == "location": path = manager.tracker.get_path_to_location(args["id"]) elif args["type"] == "worldport": path = manager.tracker.get_path_to_port(args["id"]) elif args["type"] == "goal": path = manager.tracker.get_path_to_goal() manager.game_ctx.send_path_reply(args["type"], args.get("id", None), path) elif cmd == "Quit": manager.client_ctx.exit_event.set() async def run_game(): exe_file = settings.get_settings().lingo2_options.exe_file # This ensures we can use Steam features without having to open the game # through steam. steam_appid_path = os.path.join(os.path.dirname(exe_file), "steam_appid.txt") with open(steam_appid_path, "w") as said_handle: said_handle.write("2523310") if Lingo2World.zip_path is not None: # This is a packaged apworld. init_scene = pkgutil.get_data(__name__, "client/run_from_apworld.tscn") init_path = Utils.local_path("data", "lingo2_init.tscn") with open(init_path, "wb") as file_handle: file_handle.write(init_scene) subprocess.Popen( [ exe_file, "--scene", init_path, "--", str(Lingo2World.zip_path.absolute()), ], cwd=os.path.dirname(exe_file), ) else: # The world is unzipped and being run in source. subprocess.Popen( [ exe_file, "--scene", Utils.local_path("worlds", "lingo2", "client", "run_from_source.tscn"), "--", Utils.local_path("worlds", "lingo2", "client"), ], cwd=os.path.dirname(exe_file), ) def client_main(*launch_args: str) -> None: async def main(args): async_start(run_game()) client_ctx = Lingo2ClientContext(args.connect, args.password) game_ctx = Lingo2GameContext() manager = Lingo2Manager(game_ctx, client_ctx) client_ctx.server_task = asyncio.create_task(server_loop(client_ctx), name="ServerLoop") if gui_enabled: client_ctx.run_gui() client_ctx.run_cli() pipe_task = asyncio.create_task(pipe_loop(manager), name="GameWatcher") try: await pipe_task except Exception as e: logger.exception(e) await client_ctx.exit_event.wait() client_ctx.ui.stop() await client_ctx.shutdown() Utils.init_logging("Lingo2Client", exception_logger="Client") import colorama parser = get_base_parser(description="Lingo 2 Archipelago Client") parser.add_argument('--name', default=None, help="Slot Name to connect as.") parser.add_argument("url", nargs="?", help="Archipelago connection url") args = parser.parse_args(launch_args) args = handle_url_arg(args, parser=parser) colorama.just_fix_windows_console() asyncio.run(main(args)) colorama.deinit()