Initial FF1R implementation (#123)

FF1R

FF1Client.py

@@ -0,0 +1,255 @@
+import asyncio
+import json
+import time
+from asyncio import StreamReader, StreamWriter
+from typing import List
+
+
+import Utils
+from CommonClient import CommonContext, server_loop, gui_enabled, console_loop, ClientCommandProcessor, logger, \
+    get_base_parser
+
+SYSTEM_MESSAGE_ID = 0
+
+CONNECTION_TIMING_OUT_STATUS = "Connection timing out. Please restart your emulator then restart ff1_connector.lua"
+CONNECTION_REFUSED_STATUS = "Connection Refused. Please start your emulator make sure ff1_connector.lua is running"
+CONNECTION_RESET_STATUS = "Connection was reset. Please restart your emulator then restart ff1_connector.lua"
+CONNECTION_TENTATIVE_STATUS = "Initial Connection Made"
+CONNECTION_CONNECTED_STATUS = "Connected"
+CONNECTION_INITIAL_STATUS = "Connection has not been initiated"
+
+
+class FF1CommandProcessor(ClientCommandProcessor):
+    def __init__(self, ctx: CommonContext):
+        super().__init__(ctx)
+
+    def _cmd_nes(self):
+        """Check NES Connection State"""
+        if isinstance(self.ctx, FF1Context):
+            logger.info(f"NES Status: {self.ctx.nes_status}")
+
+
+class FF1Context(CommonContext):
+    def __init__(self, server_address, password):
+        super().__init__(server_address, password)
+        self.nes_streams: (StreamReader, StreamWriter) = None
+        self.nes_sync_task = None
+        self.messages = {}
+        self.locations_array = None
+        self.nes_status = CONNECTION_INITIAL_STATUS
+        self.game = 'Final Fantasy'
+        self.awaiting_rom = False
+
+    command_processor = FF1CommandProcessor
+
+    async def server_auth(self, password_requested: bool = False):
+        if password_requested and not self.password:
+            await super(FF1Context, self).server_auth(password_requested)
+        if not self.auth:
+            self.awaiting_rom = True
+            logger.info('Awaiting connection to NES to get Player information')
+            return
+
+        await self.send_connect()
+
+    def _set_message(self, msg: str, msg_id: int):
+        self.messages[(time.time(), msg_id)] = msg
+
+    def on_package(self, cmd: str, args: dict):
+        if cmd == 'Connected':
+            self.game = self.games.get(self.slot, None)
+            asyncio.create_task(parse_locations(self.locations_array, self, True))
+        elif cmd == 'Print':
+            msg = args['text']
+            if ': !' not in msg:
+                self._set_message(msg, SYSTEM_MESSAGE_ID)
+        elif cmd == "ReceivedItems":
+            msg = f"Received {', '.join([self.item_name_getter(item.item) for item in args['items']])}"
+            self._set_message(msg, SYSTEM_MESSAGE_ID)
+        elif cmd == 'PrintJSON':
+            print_type = args['type']
+            item = args['item']
+            receiving_player_id = args['receiving']
+            receiving_player_name = self.player_names[receiving_player_id]
+            sending_player_id = item.player
+            sending_player_name = self.player_names[item.player]
+            if print_type == 'Hint':
+                msg = f"Hint: Your {self.item_name_getter(item.item)} is at" \
+                      f" {self.player_names[item.player]}'s {self.location_name_getter(item.location)}"
+                self._set_message(msg, item.item)
+            elif print_type == 'ItemSend' and receiving_player_id != self.slot:
+                if sending_player_id == self.slot:
+                    if receiving_player_id == self.slot:
+                        msg = f"You found your own {self.item_name_getter(item.item)}"
+                    else:
+                        msg = f"You sent {self.item_name_getter(item.item)} to {receiving_player_name}"
+                else:
+                    if receiving_player_id == sending_player_id:
+                        msg = f"{sending_player_name} found their {self.item_name_getter(item.item)}"
+                    else:
+                        msg = f"{sending_player_name} sent {self.item_name_getter(item.item)} to " \
+                              f"{receiving_player_name}"
+                self._set_message(msg, item.item)
+
+
+def get_payload(ctx: FF1Context):
+    current_time = time.time()
+    return json.dumps(
+        {
+            "items": [item.item for item in ctx.items_received],
+            "messages": {f'{key[0]}:{key[1]}': value for key, value in ctx.messages.items()
+                         if key[0] > current_time - 10}
+        }
+    )
+
+
+async def parse_locations(locations_array: List[int], ctx: FF1Context, force: bool):
+    if locations_array == ctx.locations_array and not force:
+        return
+    else:
+        # print("New values")
+        ctx.locations_array = locations_array
+        locations_checked = []
+        if len(locations_array) > 0xFE and locations_array[0xFE] & 0x02 != 0 and not ctx.finished_game:
+            await ctx.send_msgs([
+                {"cmd": "StatusUpdate",
+                 "status": 30}
+            ])
+            ctx.finished_game = True
+        for location in ctx.missing_locations:
+            # index will be - 0x100 or 0x200
+            index = location
+            if location < 0x200:
+                # Location is a chest
+                index -= 0x100
+                flag = 0x04
+            else:
+                # Location is an NPC
+                index -= 0x200
+                flag = 0x02
+
+            # print(f"Location: {ctx.location_name_getter(location)}")
+            # print(f"Index: {str(hex(index))}")
+            # print(f"value: {locations_array[index] & flag != 0}")
+            if locations_array[index] & flag != 0:
+                locations_checked.append(location)
+        if locations_checked:
+            # print([ctx.location_name_getter(location) for location in locations_checked])
+            await ctx.send_msgs([
+                {"cmd": "LocationChecks",
+                 "locations": locations_checked}
+            ])
+
+
+async def nes_sync_task(ctx: FF1Context):
+    logger.info("Starting nes connector. Use /nes for status information")
+    while not ctx.exit_event.is_set():
+        error_status = None
+        if ctx.nes_streams:
+            (reader, writer) = ctx.nes_streams
+            msg = get_payload(ctx).encode()
+            writer.write(msg)
+            writer.write(b'\n')
+            try:
+                await asyncio.wait_for(writer.drain(), timeout=1.5)
+                try:
+                    # Data will return a dict with up to two fields:
+                    # 1. A keepalive response of the Players Name (always)
+                    # 2. An array representing the memory values of the locations area (if in game)
+                    data = await asyncio.wait_for(reader.readline(), timeout=5)
+                    data_decoded = json.loads(data.decode())
+                    # print(data_decoded)
+                    if ctx.game is not None and 'locations' in data_decoded:
+                        # Not just a keep alive ping, parse
+                        asyncio.create_task(parse_locations(data_decoded['locations'], ctx, False))
+                    if not ctx.auth:
+                        ctx.auth = ''.join([chr(i) for i in data_decoded['playerName'] if i != 0])
+                        if ctx.awaiting_rom:
+                            await ctx.server_auth(False)
+                except asyncio.TimeoutError:
+                    logger.debug("Read Timed Out, Reconnecting")
+                    error_status = CONNECTION_TIMING_OUT_STATUS
+                    writer.close()
+                    ctx.nes_streams = None
+                except ConnectionResetError as e:
+                    logger.debug("Read failed due to Connection Lost, Reconnecting")
+                    error_status = CONNECTION_RESET_STATUS
+                    writer.close()
+                    ctx.nes_streams = None
+            except TimeoutError:
+                logger.debug("Connection Timed Out, Reconnecting")
+                error_status = CONNECTION_TIMING_OUT_STATUS
+                writer.close()
+                ctx.nes_streams = None
+            except ConnectionResetError:
+                logger.debug("Connection Lost, Reconnecting")
+                error_status = CONNECTION_RESET_STATUS
+                writer.close()
+                ctx.nes_streams = None
+            if ctx.nes_status == CONNECTION_TENTATIVE_STATUS:
+                if not error_status:
+                    logger.info("Successfully Connected to NES")
+                    ctx.nes_status = CONNECTION_CONNECTED_STATUS
+                else:
+                    ctx.nes_status = f"Was tentatively connected but error occured: {error_status}"
+            elif error_status:
+                ctx.nes_status = error_status
+                logger.info("Lost connection to nes and attempting to reconnect. Use /nes for status updates")
+        else:
+            try:
+                logger.debug("Attempting to connect to NES")
+                ctx.nes_streams = await asyncio.wait_for(asyncio.open_connection("localhost", 52980), timeout=10)
+                ctx.nes_status = CONNECTION_TENTATIVE_STATUS
+            except TimeoutError:
+                logger.debug("Connection Timed Out, Trying Again")
+                ctx.nes_status = CONNECTION_TIMING_OUT_STATUS
+                continue
+            except ConnectionRefusedError:
+                logger.debug("Connection Refused, Trying Again")
+                ctx.nes_status = CONNECTION_REFUSED_STATUS
+                continue
+
+
+if __name__ == '__main__':
+    # Text Mode to use !hint and such with games that have no text entry
+    Utils.init_logging("FF1Client")
+
+    async def main(args):
+        ctx = FF1Context(args.connect, args.password)
+        ctx.server_task = asyncio.create_task(server_loop(ctx), name="ServerLoop")
+        if gui_enabled:
+            input_task = None
+            from kvui import TextManager
+            ctx.ui = TextManager(ctx)
+            ui_task = asyncio.create_task(ctx.ui.async_run(), name="UI")
+        else:
+            input_task = asyncio.create_task(console_loop(ctx), name="Input")
+            ui_task = None
+
+        ctx.nes_sync_task = asyncio.create_task(nes_sync_task(ctx), name="NES Sync")
+
+        await ctx.exit_event.wait()
+        ctx.server_address = None
+
+        await ctx.shutdown()
+
+        if ctx.nes_sync_task:
+            await ctx.nes_sync_task
+
+        if ui_task:
+            await ui_task
+
+        if input_task:
+            input_task.cancel()
+
+
+    import colorama
+
+    parser = get_base_parser()
+    args, rest = parser.parse_known_args()
+    colorama.init()
+
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(main(args))
+    loop.close()
+    colorama.deinit()