dockipelago/worlds/cv_dos/Client.py

from typing import TYPE_CHECKING

from NetUtils import ClientStatus
from .in_game_data import location_ram_table, global_soul_table, world_version, button_item_table
from .static_location_data import location_ids
import worlds._bizhawk as bizhawk
from worlds._bizhawk.client import BizHawkClient
import time
import struct
import asyncio

if TYPE_CHECKING:
    from worlds._bizhawk.context import BizHawkClientContext


class DoSClient(BizHawkClient):
    game = "Castlevania: Dawn of Sorrow"
    system = ("NDS")
    patch_suffix = ".apcvdos"
    most_recent_connect: str = ""
    client_version: str = world_version
    has_received_death: bool = False
    state_is_dying: int = 0
    has_reset_from_death: bool = True

    def __init__(self) -> None:
        super().__init__()

    async def validate_rom(self, ctx: "BizHawkClientContext") -> bool:

        try:
            # Check ROM name/patch version
            validation_data = await bizhawk.read(ctx.bizhawk_ctx, [(0x0, 18, "ROM"), (0x02F6DD7C, 16, "ROM"), (0x02F6DD8D, 1, "ROM")])
            base_rom_name = validation_data[0].decode("ascii") # AP ROM name

            if not base_rom_name.startswith("CASTLEVANIA1ACVEA4"):
                print("A!!!!")
                print(base_rom_name)
                return False

            # This is a DoS ROM
            patch_version = validation_data[1].rstrip(b"\x69")  # APworld version
            patch_version = patch_version.decode("ascii")

            if patch_version != self.client_version:
                if patch_version != self.most_recent_connect:
                    # We only want to display this error once
                    ctx.gui_error("Bad Version", f"Installed Dawn of Sorrow APworld version {self.client_version} does not match patch version {patch_version}")
                    self.most_recent_connect = patch_version
                return False
            
        except UnicodeDecodeError:
            return False
        except bizhawk.RequestFailedError:
            return False  # Should verify on the next pass

        death_link_flag = int.from_bytes(validation_data[2])
        if death_link_flag:
            await ctx.update_death_link(True)

        ctx.game = self.game
        ctx.items_handling = 0b101
        return True

    async def set_auth(self, ctx: "BizHawkClientContext") -> None:

        slot_name_bytes = await bizhawk.read(
            ctx.bizhawk_ctx, [(0x2F6DD50, 0x14, "ROM")])

        slot_name_bytes = slot_name_bytes[0].rstrip(b'\xFF')
        ctx.auth = slot_name_bytes.decode("ascii")

    def on_package(self, ctx: "BizHawkClientContext", cmd: str, args: dict) -> None:
        if cmd != "Bounced":
            return
        if "tags" not in args:
            return
        if "DeathLink" in args["tags"]:
            self.has_received_death = True

    async def game_watcher(self, ctx: "BizHawkClientContext") -> None:
        from CommonClient import logger
        from .in_game_data import location_ram_table

        if ctx.server_version.build > 0:
            ctx.connected = True
        else:
            ctx.connected = False
            ctx.refresh_connect = True

        if ctx.slot_data is not None:
            ctx.data_present = True
        else:
            ctx.data_present = False

        if ctx.server is None or ctx.server.socket.closed or ctx.slot_data is None:
            return


        read_state = await bizhawk.read(ctx.bizhawk_ctx, [(0x0F7190, 0x10, "Main RAM"), # Check table
                                                          (0x0F7257, 0x01, "Main RAM"), # Game Mode
                                                          (0x11504C, 0x01, "Main RAM"), # Current Map
                                                          (0x0F703C, 0x04, "Main RAM"), # Gameplay timer. Will be 0 if not in game
                                                          (0x308930, 0x20, "Main RAM"), # AP data
                                                          (0x0F6DFC, 0x01, "Main RAM")]) # Game state, we only care about the Dead flag


        location_flag_table = bytearray(read_state[0])
        game_mode = int.from_bytes(read_state[1], "little")
        cur_map = int.from_bytes(read_state[2], "little")
        game_timer = int.from_bytes(read_state[3], "little")
        ap_data = bytearray(read_state[4])
        death_state = int.from_bytes(read_state[5])

        soul_flag_table = list(ap_data[:0x10])
        button_items = ap_data[0x13]
        current_received_item = ap_data[0x10]
        total_items_received = int.from_bytes(ap_data[0x1E:0x20], "little")
        if "DeathLink" in ctx.tags:
            await self.handle_deathlink(death_state, ctx)

        new_checks = []

        if game_mode == 1:  # Ignore AP handling if the game is in Julius mode
            return

        if not game_timer: # The in-game itmer is only 0 when not in-game
            return

        for location_name in location_ids:
            loc_id = location_ids[location_name]
            if loc_id not in ctx.locations_checked:
                if location_name in global_soul_table:
                    index = global_soul_table.index(location_name)
                    bit = 1 << (index % 8)
                    offset = int(index / 8)
                    location = soul_flag_table[offset]
                elif location_name in button_item_table:
                    bit = 1 << button_item_table.index(location_name)
                    location = button_items
                else:
                    pointer = location_ram_table[location_name][0]
                    bit = location_ram_table[location_name][1]
                    location = location_flag_table[pointer]

                if location & bit:
                    new_checks.append(loc_id)
                
        for new_check_id in new_checks:
            ctx.locations_checked.add(new_check_id)
            location = ctx.location_names.lookup_in_slot(new_check_id)
            await ctx.send_msgs([{"cmd": "LocationChecks", "locations": [new_check_id]}])

        if total_items_received < len(ctx.items_received) and current_received_item == 0:
            item = ctx.items_received[total_items_received]
            total_items_received += 1
            item_data = struct.pack(">H", item.item)
            await bizhawk.write(ctx.bizhawk_ctx, [(0x308940, item_data, "Main RAM")])
            await bizhawk.write(ctx.bizhawk_ctx, [(0x30894E, bytes([total_items_received]), "Main RAM")])

        if not ctx.finished_game and cur_map == 0x0D:  # Map 0x0D is used for the Epilogue. If we're here, trigger goal
            await ctx.send_msgs([{
                "cmd": "StatusUpdate",
                "status": ClientStatus.CLIENT_GOAL
            }])

    async def handle_deathlink(self, current_death_state, ctx):
        if current_death_state & 0x40:  # If the player is currently dead
            if self.has_received_death:  # This is the death that we just got from the server
                self.has_received_death = False
                self.has_reset_from_death = False
            else:  # Received death is false, meaning the player actually died here
                if self.has_reset_from_death: # We only want this to run once per death
                    await ctx.send_death(f"{ctx.player_names[ctx.slot]} died!")
                    self.has_reset_from_death = False
        else:
            if self.has_received_death:
                # Kill the player
                await bizhawk.write(ctx.bizhawk_ctx, [(0x308AAC, int.to_bytes(0x01), "Main RAM")])
            else:
                # This should be normal gameplay after relaoding
                self.has_reset_from_death = True