Skip to content

ED/SC64 support #3347

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
UmedMuzl wants to merge 1 commit into
base: v6
Choose a base branch
from
Open

ED/SC64 support #3347

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 112 additions & 35 deletions archipelago/DK64Client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

from archipelago.client.common import DK64MemoryMap, create_task_log_exception, check_version, get_ap_version
from archipelago.client.emu_loader import EmuLoaderClient
from archipelago.client.everdrive_loader import EverdriveClient, SC64Client
from archipelago.client.items import item_ids, item_names_to_id, trap_name_to_index, trap_index_to_name
from archipelago.client.ap_item_packets import APItemPacket, build_packet, ICE_TRAP_TYPES, REQITEM_ICETRAP, REQITEM_GOLDENBANANA, REQITEM_MOVE, CONFIG_APPLY_ICE_TRAP
from archipelago.client.check_flag_locations import location_flag_to_name, location_name_to_flag
Expand All @@ -36,7 +37,8 @@
KONG_COUNT = 5
LEVEL_COUNT = 7
HINT_BITFIELD_SIZE = 5

FLAG_CACHE_SIZE = 0x200
USB_BACKENDS = (EverdriveClient, SC64Client)

class CreateHintsParams:
"""Parameters for creating hints."""
Expand Down Expand Up @@ -153,6 +155,7 @@ class DK64Client:
auth = None
memory_pointer = None
stop_bizhawk_spam = False
use_everdrive = False
seed_started = False
locations_scouted = {}
recvd_checks = {}
Expand All @@ -173,6 +176,7 @@ class DK64Client:
send_mode = 1
current_speed = NORMAL_TEXT_SPEED
current_map = 0
flag_cache = None

# Hint system
last_hint_bitfield = [0] * HINT_BITFIELD_SIZE
Expand All @@ -181,47 +185,103 @@ class DK64Client:

# ==================== CONNECTION METHODS ====================

def _backend_name(self) -> str:
"""Human-readable name of the currently selected backend."""
if isinstance(self.n64_client, USB_BACKENDS):
return self.n64_client.DISPLAY_NAME
return "Emulator"

def _open_emulator(self):
"""Attach to a running emulator that already has a valid DK64 AP ROM."""
client = EmuLoaderClient()
try:
if client.connect() and client.validate_rom():
return client
except Exception:
pass
try:
client.disconnect()
except Exception:
pass
return None

def _open_backend(self):
"""Open a memory-access backend."""
if self.use_everdrive:
usb_classes = list(USB_BACKENDS)
else:
usb_classes = [cls for cls in USB_BACKENDS if cls.cart_present()]

for cls in usb_classes:
client = cls()
try:
if client.connect():
return client
except Exception:
pass
try:
client.disconnect()
except Exception:
pass

if self.use_everdrive:
return None
return self._open_emulator()

async def wait_for_pj64(self):
"""Wait for emulator to connect to the game."""
"""Wait until an emulator or USB flashcart with a valid DK64 ROM is ready."""
clear_waiting_message = True
if not self.stop_bizhawk_spam:
logger.info("Waiting on connection to emulator...")
self.n64_client = EmuLoaderClient()
logger.info("Waiting on a connection to an emulator or EverDrive...")
self.stop_bizhawk_spam = True
while True:
try:
emulator_connected = False
# Open a backend if we don't have one. A USB cart's device is kept
# open across retries — re-opening it interrupts the cart while it
# boots — so only the ROM-readiness check below is retried.
if self.n64_client is None or not self.n64_client.is_connected():
if self.n64_client is not None:
self.n64_client.disconnect()
self.n64_client = None
self.n64_client = self._open_backend()

if self.n64_client is not None and self.n64_client.is_connected():
if self.n64_client.validate_rom():
if isinstance(self.n64_client, USB_BACKENDS):
# Clear any probe backlog so gameplay reads stay synced.
self.n64_client.resync()
self.stop_bizhawk_spam = False
logger.info(f"{self._backend_name()} connected to ROM!")
return

if isinstance(self.n64_client, USB_BACKENDS):
# Cart is connected but the AP ROM hasn't booted yet. Hold
# the USB device open and wait; only drop it if the cart is
# physically unplugged (so we re-detect on reconnect).
if type(self.n64_client).cart_present():
if clear_waiting_message:
logger.info(f"{self._backend_name()} connected over USB; waiting for the ROM to boot...")
clear_waiting_message = False
await asyncio.sleep(1.0)
continue
self.n64_client.disconnect()
self.n64_client = None
else:
# Emulator attached without a valid ROM; drop and re-detect.
self.n64_client.disconnect()
self.n64_client = None

# Try to connect to any available emulator
if not self.n64_client.is_connected():
emulator_connected = self.n64_client.connect()
else:
emulator_connected = True
valid_rom = False
if emulator_connected:
valid_rom = self.n64_client.validate_rom()
logger.info("Emulator connected, validating ROM...")

while not valid_rom:
if not self.n64_client.is_connected():
emulator_connected = self.n64_client.connect()
if clear_waiting_message:
logger.info("Waiting on valid ROM...")
clear_waiting_message = False
await asyncio.sleep(1.0)
if self.n64_client.is_connected():
valid_rom = self.n64_client.validate_rom()

self.stop_bizhawk_spam = False
logger.info("Emulator Connected to ROM!")
return
if clear_waiting_message:
logger.info("Waiting on an emulator or EverDrive with a valid ROM...")
clear_waiting_message = False
await asyncio.sleep(1.0)
except Exception as e:
await asyncio.sleep(1.0)
logger.error(f"Error connecting to emulator, retrying... {str(e)}")
logger.error(f"Error connecting, retrying... {str(e)}")
# Reset connection on error
if self.n64_client:
self.n64_client.disconnect()
pass
self.n64_client = None

# ==================== GAME STATE METHODS ====================

Expand Down Expand Up @@ -262,7 +322,7 @@ async def validate_client_connection(self):
"""Validate the client connection."""
self.memory_pointer = self.n64_client.read_u32(DK64MemoryMap.memory_pointer)
self.n64_client.write_u8(self.memory_pointer + DK64MemoryMap.connection, 0xFF)
if self.n64_client.read_u8(DK64MemoryMap.eeprom_determined) == 1:
if isinstance(self.n64_client, EmuLoaderClient) and self.n64_client.read_u8(DK64MemoryMap.eeprom_determined) == 1:
if self.n64_client.read_u32(DK64MemoryMap.save_type) != 2:
# Map emulator IDs to their setup guides
emulator_setup_guides = {
Expand Down Expand Up @@ -772,6 +832,15 @@ async def readChecks(self, cb):
new_checks = []
checks_to_read = self.remaining_checks.copy()

# Snapshot the whole EEPROM flag block once
self.flag_cache = self.n64_client.read_block(DK64MemoryMap.EEPROM, FLAG_CACHE_SIZE)
try:
return self._scan_checks(checks_to_read, new_checks, cb)
finally:
self.flag_cache = None

def _scan_checks(self, checks_to_read, new_checks, cb):
"""Scan the requested location ids against the cached flag block."""
for id in checks_to_read:
name = check_id_to_name.get(id)
check = location_name_to_flag.get(name)
Expand Down Expand Up @@ -873,8 +942,12 @@ def readFlag(self, index: int) -> int:
"""Read a flag in the game."""
byte_index = index >> 3
shift = index & 7
offset = DK64MemoryMap.EEPROM + byte_index
val = self.n64_client.read_u8(offset)
cache = self.flag_cache
if cache is not None and byte_index < len(cache):
val = cache[byte_index]
else:
offset = DK64MemoryMap.EEPROM + byte_index
val = self.n64_client.read_u8(offset)
return (val >> shift) & 1

def hasKong(self, kong_index: int) -> bool:
Expand Down Expand Up @@ -1802,6 +1875,8 @@ async def disconnect_check():
logger.info("No checks received yet, requesting...")
await self.sync()

tick_delay = 0.066 if isinstance(self.client.n64_client, USB_BACKENDS) else 0.033

await asyncio.sleep(1.0)
while True:
logger.debug("Game loop tick")
Expand All @@ -1812,10 +1887,10 @@ async def disconnect_check():
await victory()
status = self.client.check_safe_gameplay()
if status is False:
await asyncio.sleep(0.033)
await asyncio.sleep(tick_delay)
continue
await self.client.main_tick(on_item_get, deathlink, map_change, ring_link, tag_link, trap_link, hint_accessed)
await asyncio.sleep(0.033)
await asyncio.sleep(tick_delay)
now = time.time()
if self.last_resend + 0.5 < now:
self.last_resend = now
Expand All @@ -1838,11 +1913,13 @@ async def main():
"""Entrypoint of codebase."""
parser = get_base_parser(description="Donkey Kong 64 Client.")
parser.add_argument("--url", help="Archipelago connection url")
parser.add_argument("--everdrive", action="store_true", help="Force USB flashcart mode (EverDrive/SC64) and skip emulator detection. A connected cart is auto-detected without this flag.")

args = parser.parse_args()
check_version()

ctx = DK64Context(args.connect, args.password)
ctx.client.use_everdrive = args.everdrive
ctx.items_handling = 0b001
ctx.server_task = asyncio.create_task(server_loop(ctx), name="server loop")
ctx.la_task = create_task_log_exception(ctx.run_game_loop())
Expand Down
123 changes: 15 additions & 108 deletions archipelago/FillSettings.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -485,25 +485,17 @@ def fillsettings(options: DK64Options, multiworld: MultiWorld, random_obj: Rando
slam_name = options.alter_switch_allocation.value[level_key]
settings_dict[f"prog_slam_level_{i + 1}"] = slam_map.get(slam_name, SlamRequirement.green)


def generate_blocker(option_value: str, blocker_max: int, random: Random):
"""Randomize a B. Locker value, either within a range or up to the maximum."""
upper_bound = blocker_max if option_value == "random" else int(option_value.split("-")[1]) + 1
lower_bound = 0 if option_value == "random" else int(option_value.split("-")[0])
return random.randrange(lower_bound, upper_bound)


def apply_blocker_settings(settings_dict: dict, options, random_obj) -> None:
"""Apply level blocker settings."""
blocker_options = [0, 0, 0, 0, 0, 0, 0, 0]
for blocker, amount in options.level_blockers.value.items():
blocker_number = int(blocker.removeprefix("level_")) - 1
try:
blocker_options[blocker_number] = int(amount)
except (TypeError, ValueError):
blocker_options[blocker_number] = generate_blocker(amount, options.blocker_max.value, random_obj)

# Blocker settings - prioritize chaos blockers, then randomization setting
# Apply blocker settings
blocker_options: list[int] = [
options.level_blockers.value.get("level_1", 0),
options.level_blockers.value.get("level_2", 0),
options.level_blockers.value.get("level_3", 0),
options.level_blockers.value.get("level_4", 0),
options.level_blockers.value.get("level_5", 0),
options.level_blockers.value.get("level_6", 0),
options.level_blockers.value.get("level_7", 0),
options.level_blockers.value.get("level_8", 64),
]
settings_dict["maximize_helm_blocker"] = options.maximize_level8_blocker.value
if options.enable_chaos_blockers.value:
settings_dict["blocker_text"] = options.chaos_ratio.value
Expand Down Expand Up @@ -965,95 +957,10 @@ def apply_blocker_settings(settings_dict: dict, options, random_obj) -> None:
settings_dict["coin_door_item"] = HelmDoorItem(options.coin_door_item.value)
coin_item_key = door_item_to_key.get(settings_dict["coin_door_item"])
settings_dict["coin_door_item_count"] = options.helm_door_item_count.value.get(coin_item_key, 1) if coin_item_key else 1

if hasattr(multiworld, "generation_is_fake"):
if hasattr(multiworld, "re_gen_passthrough"):
if "Donkey Kong 64" in multiworld.re_gen_passthrough:
passthrough = multiworld.re_gen_passthrough["Donkey Kong 64"]
settings_dict["bonus_barrel_auto_complete"] = passthrough["Autocomplete"]
settings_dict["helm_room_bonus_count"] = HelmBonuses(passthrough["HelmBarrelCount"])


def handle_fake_generation_settings(settings: Settings, multiworld) -> None:
"""Handle settings for fake generation (UT mode)."""
if hasattr(multiworld, "generation_is_fake"):
settings.is_ut_generation = True
if hasattr(multiworld, "re_gen_passthrough"):
if "Donkey Kong 64" in multiworld.re_gen_passthrough:
passthrough = multiworld.re_gen_passthrough["Donkey Kong 64"]
settings.level_order = passthrough["LevelOrder"]

# Switch logic lifted out of level shuffle due to static levels for UT
if settings.alter_switch_allocation:
for x in range(8):
settings.switch_allocation[x] = passthrough["SlamLevels"][x]

settings.starting_kong_list = passthrough["StartingKongs"]
settings.starting_kong = settings.starting_kong_list[0] # fake a starting kong so that we don't force a different kong
settings.medal_requirement = passthrough["JetpacReq"]
settings.rareware_gb_fairies = passthrough["FairyRequirement"]
settings.BLockerEntryItems = passthrough["BLockerEntryItems"]
settings.BLockerEntryCount = passthrough["BLockerEntryCount"]
settings.medal_cb_req = passthrough["MedalCBRequirement"]
settings.medal_cb_req_level = [settings.medal_cb_req] * 8

for level, value in enumerate(passthrough["MedalCBRequirementLevel"]):
settings.medal_cb_req_level[Levels(level)] = int(value)

settings.mermaid_gb_pearls = passthrough["MermaidPearls"]
settings.BossBananas = passthrough["BossBananas"]
settings.boss_maps = passthrough["BossMaps"]
settings.boss_kongs = passthrough["BossKongs"]
settings.lanky_freeing_kong = passthrough["LankyFreeingKong"]
settings.helm_order = passthrough["HelmOrder"]
settings.logic_type = LogicType[passthrough["LogicType"]]
settings.tricks_selected = passthrough["TricksSelected"]
settings.glitches_selected = passthrough["GlitchesSelected"]
settings.open_lobbies = passthrough["OpenLobbies"]
settings.starting_key_list = passthrough["StartingKeyList"]
settings.galleon_water = GalleonWaterSetting[passthrough["GalleonWater"]]
settings.galleon_water_internal = GalleonWaterSetting[passthrough["GalleonWater"]]

# There's multiple sources of truth for helm order.
settings.helm_donkey = 0 in settings.helm_order
settings.helm_diddy = 4 in settings.helm_order
settings.helm_lanky = 3 in settings.helm_order
settings.helm_tiny = 2 in settings.helm_order
settings.helm_chunky = 1 in settings.helm_order

# Switchsanity
for switch, data in passthrough["SwitchSanity"].items():
needed_kong = Kongs[data["kong"]]
switch_type = SwitchType[data["type"]]
settings.switchsanity_data[Switches[switch]] = SwitchInfo(switch, needed_kong, switch_type, 0, 0, [])

if passthrough["Shopkeepers"]:
settings.shuffled_location_types.append(Types.Cranky)
settings.shuffled_location_types.append(Types.Funky)
settings.shuffled_location_types.append(Types.Candy)
settings.shuffled_location_types.append(Types.Snide)


def fillsettings(options, multiworld, random_obj):
"""Fill and configure all DK64 settings."""
# Start with default settings
settings_dict = get_default_settings()

# Apply all setting categories
apply_archipelago_settings(settings_dict, options, multiworld)
apply_blocker_settings(settings_dict, options, random_obj)
apply_item_randomization_settings(settings_dict, options)
apply_hard_mode_settings(settings_dict, options)
apply_kong_settings(settings_dict, options)
apply_switchsanity_settings(settings_dict, options)
apply_logic_and_barriers_settings(settings_dict, options)
apply_glitches_and_tricks_settings(settings_dict, options)
apply_boss_and_key_settings(settings_dict, options)
apply_goal_settings(settings_dict, options, random_obj)
apply_starting_moves_settings(settings_dict, options)
apply_hint_settings(settings_dict, options)
apply_minigame_settings(settings_dict, options, multiworld)
apply_enemies(settings_dict, options)
if hasattr(multiworld, "generation_is_fake") and hasattr(multiworld, "re_gen_passthrough") and "Donkey Kong 64" in multiworld.re_gen_passthrough:
passthrough = multiworld.re_gen_passthrough["Donkey Kong 64"]
settings_dict["bonus_barrel_auto_complete"] = passthrough["Autocomplete"]
settings_dict["helm_room_bonus_count"] = HelmBonuses(passthrough["HelmBarrelCount"])

# Handle fake generation keys if needed
if hasattr(multiworld, "generation_is_fake"):
Expand Down
Loading