Added the_door - lingo2-archipelago - Randomizer for LINGO 2 using Archipelago Multiworld

diff options

author	Star Rauchenberger <fefferburbia@gmail.com>	2025-08-11 10:09:44 -0400
committer	Star Rauchenberger <fefferburbia@gmail.com>	2025-08-11 10:09:44 -0400
commit	2e76227a915cd6012d0b363918a5eb3c124b40d8 (patch)
tree	aa219adc7eea4626cbb804c1a5c140fdb1cefc7b /data/maps/the_door/rooms
parent	664fd37cea4185d53561c98880be7ccd15b18223 (diff)
download	lingo2-archipelago-2e76227a915cd6012d0b363918a5eb3c124b40d8.tar.gz lingo2-archipelago-2e76227a915cd6012d0b363918a5eb3c124b40d8.tar.bz2 lingo2-archipelago-2e76227a915cd6012d0b363918a5eb3c124b40d8.zip

Added the_door

Diffstat (limited to 'data/maps/the_door/rooms')

-rw-r--r--

data/maps/the_door/rooms/Ending.txtpb

-rw-r--r--

data/maps/the_door/rooms/Main Area.txtpb

2 files changed, 19 insertions, 0 deletions


diff --git a/data/maps/the_door/rooms/Ending.txtpb b/data/maps/the_door/rooms/Ending.txtpb new file mode 100644 index 0000000..67937c5 --- /dev/null +++ b/data/maps/the_door/rooms/Ending.txtpb
@@ -0,0 +1,3 @@
	1	name: "Ending"
	2	display_name: "Main Area"
	3	# TODO: Plum Ending


diff --git a/data/maps/the_door/rooms/Main Area.txtpb b/data/maps/the_door/rooms/Main Area.txtpb new file mode 100644 index 0000000..47d2224 --- /dev/null +++ b/data/maps/the_door/rooms/Main Area.txtpb
@@ -0,0 +1,16 @@
	1	name: "Main Area"
	2	display_name: "Main Area"
	3	panels {
	4	name: "HAIR"
	5	path: "Panels/Room_1/panel"
	6	clue: "hair"
	7	answer: "rabbit"
	8	symbols: "question"
	9	}
	10	panels {
	11	name: "PART"
	12	path: "Panels/Room_1/panel2"
	13	clue: "part"
	14	answer: "hole"
	15	symbols: "question"
	16	}

import asyncio import os import pkgutil import subprocess import sys from typing import Any import websockets import Utils import settings from BaseClasses import ItemClassification from CommonClient import CommonContext, server_loop, gui_enabled, logger, get_base_parser, handle_url_arg from NetUtils import Endpoint, decode, encode, ClientStatus from Utils import async_start from . import Lingo2World from .tracker import Tracker ALL_LETTERS = "abcdefghijklmnopqrstuvwxyz" MESSAGE_MAX_SIZE = 16*1024*1024 PORT = 43182 KEY_STORAGE_MAPPING = { "a": (1, 0), "b": (1, 1), "c": (1, 2), "d": (1, 3), "e": (1, 4), "f": (1, 5), "g": (1, 6), "h": (1, 7), "i": (1, 8), "j": (1, 9), "k": (1, 10), "l": (1, 11), "m": (1, 12), "n": (2, 0), "o": (2, 1), "p": (2, 2), "q": (2, 3), "r": (2, 4), "s": (2, 5), "t": (2, 6), "u": (2, 7), "v": (2, 8), "w": (2, 9), "x": (2, 10), "y": (2, 11), "z": (2, 12), } REVERSE_KEY_STORAGE_MAPPING = {t: k for k, t in KEY_STORAGE_MAPPING.items()} class Lingo2Manager: game_ctx: "Lingo2GameContext" client_ctx: "Lingo2ClientContext" tracker: Tracker keyboard: dict[str, int] worldports: set[int] goaled: bool latches: set[int] 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.latches = set() self.reset() def reset(self): for k in ALL_LETTERS: self.keyboard[k] = 0 self.worldports = set() self.goaled = False self.latches = set() 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 def update_latches(self, new_latches: set[int]) -> set[int]: ret = new_latches.difference(self.latches) self.latches.update(new_latches) 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") def send_update_latches(self, latches): if self.server is None: return msg = { "cmd": "UpdateLatches", "latches": latches, } async_start(self.send_msgs([msg]), name="update latches") 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, self.get_datastorage_key("latches")) 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}] }, { "cmd": "Set", "key": self.get_datastorage_key("latches"), "default": [], "want_reply": True, "operations": [{"operation": "default", "value": []}] }] 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"]) elif args["key"] == self.get_datastorage_key("latches"): updates = self.manager.update_latches(set(args["value"])) if len(updates) > 0: self.manager.game_ctx.send_update_latches(updates) 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 update_latches(self, updates: set[int]): await self.send_msgs([{ "cmd": "Set", "key": self.get_datastorage_key("latches"), "want_reply": True, "operations": [{ "operation": "update", "value": updates }] }]) 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} # Also check the reverse port if it's a two-way connection. port_pairings = manager.client_ctx.slot_data["port_pairings"] if str(port_id) in port_pairings and port_pairings.get(str(port_pairings[str(port_id)]), None) == port_id: worldports.add(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 == "LatchDoor": updates = manager.update_latches({args["door"]}) if len(updates) > 0: async_start(manager.client_ctx.update_latches(updates), name="client update latches") 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): if settings.get_settings().lingo2_options.start_game: async_start(run_game()) client_ctx = Lingo2ClientContext(args.connect, args.password) client_ctx.auth = args.name 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()


context:
space:
mode: