about summary refs log tree commit diff stats
path: root/apworld/__init__.py
blob: 42350bc6d7779b89a4956fb275656ebaf8c3ec2a (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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
"""
Archipelago init file for Lingo 2
"""
from typing import ClassVar

from BaseClasses import ItemClassification, Item, Tutorial
from Options import OptionError
from settings import Group, UserFilePath
from worlds.AutoWorld import WebWorld, World
from .items import Lingo2Item, ANTI_COLLECTABLE_TRAPS
from .options import Lingo2Options
from .player_logic import Lingo2PlayerLogic
from .regions import create_regions, shuffle_entrances, connect_ports_from_ut
from .static_logic import Lingo2StaticLogic
from worlds.LauncherComponents import Component, Type, components, launch as launch_component, icon_paths


class Lingo2WebWorld(WebWorld):
    rich_text_options_doc = True
    theme = "grass"
    tutorials = [Tutorial(
        "Multiworld Setup Guide",
        "A guide to playing Lingo 2 with Archipelago.",
        "English",
        "en_Lingo_2.md",
        "setup/en",
        ["hatkirby"]
    )]


class Lingo2Settings(Group):
    class ExecutableFile(UserFilePath):
        """Path to the Lingo 2 executable"""
        is_exe = True

    exe_file: ExecutableFile = ExecutableFile()
    start_game: bool = True


class Lingo2World(World):
    """
    Lingo 2 is a first person indie puzzle game where you solve word puzzles in a labyrinthe world. Compared to its
    predecessor, Lingo 2 has new mechanics, more areas, and a unique progression system where you have to unlock letters
    before using them in puzzle solutions.
    """
    game = "Lingo 2"
    web = Lingo2WebWorld()

    settings: ClassVar[Lingo2Settings]
    settings_key = "lingo2_options"

    topology_present = True

    options_dataclass = Lingo2Options
    options: Lingo2Options

    static_logic = Lingo2StaticLogic()
    item_name_to_id = static_logic.item_name_to_id
    location_name_to_id = static_logic.location_name_to_id
    item_name_groups = static_logic.item_name_groups
    location_name_groups = static_logic.location_name_groups

    for_tracker: ClassVar[bool] = False

    player_logic: Lingo2PlayerLogic

    port_pairings: dict[int, int]

    def generate_early(self):
        self.player_logic = Lingo2PlayerLogic(self)
        self.port_pairings = {}

    def create_regions(self):
        if hasattr(self.multiworld, "re_gen_passthrough") and "Lingo 2" in self.multiworld.re_gen_passthrough:
            self.player_logic.rte_mapping = [self.world.static_logic.map_id_by_name[map_name]
                                             for map_name in self.multiworld.re_gen_passthrough["Lingo 2"]["rte"]]

        create_regions(self)

    def connect_entrances(self):
        if self.options.shuffle_worldports:
            if hasattr(self.multiworld, "re_gen_passthrough") and "Lingo 2" in self.multiworld.re_gen_passthrough:
                slot_value = self.multiworld.re_gen_passthrough["Lingo 2"]["port_pairings"]
                self.port_pairings = {
                    self.static_logic.port_id_by_ap_id[int(fp)]: self.static_logic.port_id_by_ap_id[int(tp)]
                    for fp, tp in slot_value.items()
                }

                connect_ports_from_ut(self.port_pairings, self)
            else:
                shuffle_entrances(self)

        #from Utils import visualize_regions

        #visualize_regions(self.multiworld.get_region("Menu", self.player), "my_world.puml")

    def create_items(self):
        pool = [self.create_item(name) for name in self.player_logic.real_items]

        total_locations = sum(len(locs) for locs in self.player_logic.locations_by_room.values())

        item_difference = total_locations - len(pool)

        if self.options.trap_percentage > 0:
            num_traps = int(item_difference * self.options.trap_percentage / 100)
            item_difference = item_difference - num_traps

            trap_names = []
            trap_weights = []
            for letter_name, weight in self.static_logic.letter_weights.items():
                trap_names.append(f"Anti {letter_name}")
                trap_weights.append(weight)

            bad_letters = self.random.choices(trap_names, weights=trap_weights, k=num_traps)
            pool += [self.create_item(trap_name) for trap_name in bad_letters]

        for i in range(0, item_difference):
            pool.append(self.create_item(self.get_filler_item_name()))

        if not any(ItemClassification.progression in item.classification for item in pool):
            raise OptionError(f"Lingo 2 player {self.player} has no progression items. Please enable at least one "
                              f"option that would add progression gating to your world, such as Shuffle Doors or "
                              f"Shuffle Letters.")

        self.multiworld.itempool += pool

        for name in self.player_logic.starting_items:
            self.push_precollected(self.create_item(name))

    def create_item(self, name: str) -> Item:
        return Lingo2Item(name, ItemClassification.filler if name == self.get_filler_item_name() else
                                ItemClassification.trap if name in ANTI_COLLECTABLE_TRAPS else
                                ItemClassification.progression,
                          self.item_name_to_id.get(name), self.player)

    def set_rules(self):
        self.multiworld.completion_condition[self.player] = lambda state: state.has("Victory", self.player)

    def fill_slot_data(self):
        slot_options = [
            "cyan_door_behavior",
            "daedalus_only",
            "daedalus_roof_access",
            "enable_gift_maps",
            "enable_icarus",
            "endings_requirement",
            "fast_travel_access",
            "keyholder_sanity",
            "masteries_requirement",
            "shuffle_control_center_colors",
            "shuffle_doors",
            "shuffle_gallery_paintings",
            "shuffle_letters",
            "shuffle_symbols",
            "shuffle_worldports",
            "strict_cyan_ending",
            "strict_purple_ending",
            "victory_condition",
        ]

        slot_data: dict[str, object] = {
            **self.options.as_dict(*slot_options),
            "rte": [self.static_logic.objects.maps[map_id].name for map_id in self.player_logic.rte_mapping],
            "version": self.static_logic.get_data_version(),
        }

        if self.options.shuffle_worldports:
            def get_port_ap_id(port_id):
                return self.static_logic.objects.ports[port_id].ap_id

            slot_data["port_pairings"] = {get_port_ap_id(from_id): get_port_ap_id(to_id)
                                          for from_id, to_id in self.port_pairings.items()}

        return slot_data

    def get_filler_item_name(self) -> str:
        return "A Job Well Done"

    # for the universal tracker, doesn't get called in standard gen
    # docs: https://github.com/FarisTheAncient/Archipelago/blob/tracker/worlds/tracker/docs/re-gen-passthrough.md
    @staticmethod
    def interpret_slot_data(slot_data: dict[str, object]) -> dict[str, object]:
        # returning slot_data so it regens, giving it back in multiworld.re_gen_passthrough
        # we are using re_gen_passthrough over modifying the world here due to complexities with ER
        return slot_data


def launch_client(*args):
    from .context import client_main
    launch_component(client_main, name="Lingo2Client", args=args)


icon_paths["lingo2_ico"] = f"ap:{__name__}/logo.png"
component = Component("Lingo 2 Client", component_type=Type.CLIENT, func=launch_client,
                      description="Open Lingo 2.", supports_uri=True, game_name="Lingo 2", icon="lingo2_ico")
components.append(component)
class="nb">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 # Input should be real IDs, not AP IDs 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") # Input should be real IDs, not AP IDs 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"): port_ids = set(Lingo2World.static_logic.port_id_by_ap_id[ap_id] for ap_id in args["value"]) updates = self.manager.update_worldports(port_ids) 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"): door_ids = set(Lingo2World.static_logic.door_id_by_ap_id[ap_id] for ap_id in args["value"]) updates = self.manager.update_latches(door_ids) 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) # Input should be real IDs, not AP IDs async def update_worldports(self, updates: set[int]): port_ap_ids = [Lingo2World.static_logic.objects.ports[port_id].ap_id for port_id in updates] await self.send_msgs([{ "cmd": "Set", "key": self.get_datastorage_key("worldports"), "want_reply": True, "operations": [{ "operation": "update", "value": port_ap_ids }] }]) 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]): door_ap_ids = [Lingo2World.static_logic.objects.doors[door_id].ap_id for door_id in updates] await self.send_msgs([{ "cmd": "Set", "key": self.get_datastorage_key("latches"), "want_reply": True, "operations": [{ "operation": "update", "value": door_ap_ids }] }]) 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"] port_ap_id = Lingo2World.static_logic.objects.ports[port_id].ap_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_ap_id) in port_pairings and\ port_pairings.get(str(port_pairings[str(port_ap_id)]), None) == port_ap_id: worldports.add(Lingo2World.static_logic.port_id_by_ap_id[port_pairings[str(port_ap_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()