forked from mirror/Archipelago
Jonathan Tinney
7971961166
Some checks failed
Analyze modified files / flake8 (push) Failing after 2m28s
Build / build-win (push) Has been cancelled
Build / build-ubuntu2204 (push) Has been cancelled
ctest / Test C++ ubuntu-latest (push) Has been cancelled
ctest / Test C++ windows-latest (push) Has been cancelled
Analyze modified files / mypy (push) Has been cancelled
Build and Publish Docker Images / Push Docker image to Docker Hub (push) Successful in 5m4s
Native Code Static Analysis / scan-build (push) Failing after 5m2s
type check / pyright (push) Successful in 1m7s
unittests / Test Python 3.11.2 ubuntu-latest (push) Failing after 16m23s
unittests / Test Python 3.12 ubuntu-latest (push) Failing after 28m19s
unittests / Test Python 3.13 ubuntu-latest (push) Failing after 14m49s
unittests / Test hosting with 3.13 on ubuntu-latest (push) Successful in 5m0s
unittests / Test Python 3.13 macos-latest (push) Has been cancelled
unittests / Test Python 3.11 windows-latest (push) Has been cancelled
unittests / Test Python 3.13 windows-latest (push) Has been cancelled
1015 lines
47 KiB
Python
1015 lines
47 KiB
Python
import os
|
|
import time
|
|
from typing import Dict, List, Tuple
|
|
|
|
import ModuleUpdate
|
|
from Utils import async_start
|
|
|
|
import asyncio
|
|
from pymem import pymem
|
|
|
|
from NetUtils import ClientStatus, NetworkItem
|
|
from CommonClient import gui_enabled, logger, get_base_parser, CommonContext, server_loop, ClientCommandProcessor, handle_url_arg
|
|
|
|
from .Items import item_data_table, inv_item_table
|
|
from .Locations import location_data_table, FF12OpenWorldLocationData
|
|
|
|
tracker_loaded = False
|
|
try:
|
|
from worlds.tracker.TrackerClient import TrackerGameContext, TrackerCommandProcessor
|
|
CommonContext = TrackerGameContext
|
|
ClientCommandProcessor = TrackerCommandProcessor
|
|
tracker_loaded = True
|
|
except ModuleNotFoundError:
|
|
pass
|
|
|
|
ModuleUpdate.update()
|
|
|
|
sort_start_addresses = [
|
|
0x204FD4C, # Items
|
|
0x204FDCC, # Weapons
|
|
0x204FF5C, # Armor
|
|
0x2050074, # Accessories
|
|
0x20500D4, # Ammo
|
|
0x2050364, # Technicks
|
|
0x2050394, # Magicks
|
|
0x2050436, # Key Items
|
|
0x2050836, # Loot
|
|
]
|
|
|
|
sort_count_addresses = [
|
|
0x2050C38, # Items
|
|
0x2050C3C, # Weapons
|
|
0x2050C40, # Armor
|
|
0x2050C44, # Accessories
|
|
0x2050C48, # Ammo
|
|
0x2050C58, # Technicks
|
|
0x2050C5C, # Magicks
|
|
0x2050C60, # Key Items
|
|
0x2050C64, # Loot
|
|
]
|
|
|
|
tracker_event_offsets = [ # list of save byte offsets that poptracker wants to know about
|
|
0x0408, # Ktjn location
|
|
0x040E, # Viera Rendezvous sidequest progress
|
|
0x040F, # Viera Rendezvous sidequest flags
|
|
0x0416, # Grimy Fragment sidequest progress
|
|
0x068B, # Desert patient sidequest progress
|
|
0x0919, # Mosphoran Highwaste flags
|
|
0x0999, 0x099A ,0x099B ,0x099C ,0x099D ,0x099E, # Giza trees
|
|
0x09F3, # Draklor bulkhead colour
|
|
0x1064 + 11, # Dreadnaught Leviathan entry
|
|
0x1064 + 53, # Earth Tyrant sidequest progress
|
|
0x1064 + 57, # Medallion sidequest progress
|
|
0x1064 + 58, # Medallion of Bravery
|
|
0x1064 + 59, # Medallion of Love
|
|
]
|
|
tracker_event_offsets.extend(range(0x0C90, 0x0CB0)) # Trophy Rare Game Kills
|
|
tracker_event_offsets.extend(range(0x1064 + 128, 0x1064 + 173)) # Hunt Progress
|
|
tracker_event_offsets.extend(range(0x0A03, 0x0A6C)) # Defeat flags
|
|
tracker_event_offsets.extend(range(0x06D7, 0x06DC)) # Visitor on Deck traveler aerodrome locations
|
|
tracker_event_offsets.extend(range(0x05AF, 0x05B6)) # Ann's letter delivery
|
|
|
|
MAX_PARTY_MEMBERS = 12
|
|
|
|
|
|
class FF12StateCache:
|
|
SAVE_DATA_LENGTH = 0xE200
|
|
ITEM_SECTION_LENGTH = 0x2000
|
|
BITFIELD_SECTION_LENGTH = 0x200
|
|
|
|
def __init__(self) -> None:
|
|
self.save_data_base: int = 0
|
|
self.save_data: bytes = b""
|
|
self.extra_segments: List[Tuple[int, bytes]] = []
|
|
self.party_address: int = 0
|
|
self.scenario_flag: int = -1
|
|
self.current_map: int = -1
|
|
self.current_game_state: int = -1
|
|
self.party_members: set[int] = set()
|
|
self.inventory_by_code: Dict[int, int] = {}
|
|
self.inventory_by_name: Dict[str, int] = {}
|
|
self.leviathan_progress: int = 0
|
|
self.escape_progress: int = 0
|
|
self.draklor_progress: int = 0
|
|
|
|
def read_range(self, absolute_address: int, size: int):
|
|
if self.save_data and self.save_data_base <= absolute_address and \
|
|
absolute_address + size <= self.save_data_base + len(self.save_data):
|
|
offset = absolute_address - self.save_data_base
|
|
return self.save_data[offset:offset + size]
|
|
|
|
for base, data in self.extra_segments:
|
|
if base <= absolute_address and absolute_address + size <= base + len(data):
|
|
offset = absolute_address - base
|
|
return data[offset:offset + size]
|
|
return None
|
|
|
|
def add_extra_segment(self, base: int, data: bytes):
|
|
if data:
|
|
self.extra_segments.append((base, data))
|
|
|
|
def save_byte(self, offset: int) -> int:
|
|
if 0 <= offset < len(self.save_data):
|
|
return self.save_data[offset]
|
|
return 0
|
|
|
|
def save_short(self, offset: int) -> int:
|
|
if 0 <= offset + 1 < len(self.save_data):
|
|
return int.from_bytes(self.save_data[offset:offset + 2], "little")
|
|
return 0
|
|
|
|
def save_bit(self, offset: int, bit: int) -> bool:
|
|
byte = self.save_byte(offset)
|
|
return ((byte >> bit) & 1) != 0
|
|
|
|
|
|
class FF12OpenWorldCommandProcessor(ClientCommandProcessor):
|
|
def __init__(self, ctx: CommonContext):
|
|
super().__init__(ctx)
|
|
|
|
def _cmd_list_processes(self):
|
|
"""List all processes found by pymem."""
|
|
for process in pymem.process.list_processes():
|
|
self.output(f"{process.szExeFile}: {process.th32ProcessID}")
|
|
|
|
def _cmd_set_process_by_id(self, process_id: str):
|
|
"""Set the process by ID (int)."""
|
|
int_id = int(process_id)
|
|
try:
|
|
self.ctx.ff12 = pymem.Pymem().open_process_from_id(int_id)
|
|
logger.info("You are now auto-tracking")
|
|
self.ctx.ff12connected = True
|
|
except Exception as e:
|
|
if self.ctx.ff12connected:
|
|
self.ctx.ff12connected = False
|
|
logger.info("Failed to set process by ID.")
|
|
logger.info(e)
|
|
|
|
|
|
# Copied from KH2 Client
|
|
class FF12OpenWorldContext(CommonContext):
|
|
command_processor = FF12OpenWorldCommandProcessor
|
|
game = "Final Fantasy 12 Open World"
|
|
items_handling = 0b111 # Indicates you get items sent from other worlds.
|
|
tags = ["AP"]
|
|
|
|
def __init__(self, server_address, password):
|
|
super(FF12OpenWorldContext, self).__init__(server_address, password)
|
|
|
|
self.last_big_batch_time = None
|
|
self.ff12_items_received: List[NetworkItem] = []
|
|
self.sending: List[int] = []
|
|
self.ff12slotdata = None
|
|
self.server_connected = False
|
|
self.ff12connected = False
|
|
self.stored_map_id = 0
|
|
self.event_flags = {}
|
|
# hooked object
|
|
self.ff12 = None
|
|
self.game_state_cache = FF12StateCache()
|
|
self.item_lock = asyncio.Lock()
|
|
if "localappdata" in os.environ:
|
|
self.game_communication_path = os.path.expandvars(r"%localappdata%\FF12OpenWorldAP")
|
|
else:
|
|
logger.info("Could not find localappdata environment variable")
|
|
self.game_communication_path = None
|
|
self.delete_communication_files()
|
|
|
|
async def get_username(self):
|
|
if not self.auth:
|
|
self.auth = self.username
|
|
if not self.auth:
|
|
logger.info('Enter slot name:')
|
|
self.auth = await self.console_input()
|
|
|
|
async def server_auth(self, password_requested: bool = False):
|
|
if password_requested and not self.password:
|
|
await super(FF12OpenWorldContext, self).server_auth(password_requested)
|
|
await self.get_username()
|
|
await self.send_connect()
|
|
|
|
async def connection_closed(self):
|
|
self.ff12connected = False
|
|
self.server_connected = False
|
|
self.delete_communication_files()
|
|
self.ff12_items_received.clear()
|
|
self.game_state_cache = FF12StateCache()
|
|
await super(FF12OpenWorldContext, self).connection_closed()
|
|
|
|
async def disconnect(self, allow_autoreconnect: bool = False):
|
|
self.ff12connected = False
|
|
self.server_connected = False
|
|
self.delete_communication_files()
|
|
self.ff12_items_received.clear()
|
|
self.game_state_cache = FF12StateCache()
|
|
await super(FF12OpenWorldContext, self).disconnect()
|
|
|
|
@property
|
|
def endpoints(self):
|
|
if self.server:
|
|
return [self.server]
|
|
else:
|
|
return []
|
|
|
|
async def shutdown(self):
|
|
await super(FF12OpenWorldContext, self).shutdown()
|
|
|
|
def ff12_story_address(self):
|
|
return self.ff12.base_address
|
|
|
|
def _compute_save_data_address(self) -> int:
|
|
if not self.ff12:
|
|
return 0
|
|
return self.ff12.base_address + 0x02044480
|
|
|
|
def _read_bytes(self, address: int, length: int, use_base: bool = True) -> bytes:
|
|
if not self.ff12:
|
|
raise RuntimeError("FF12 process not attached")
|
|
absolute = self.ff12.base_address + address if use_base else address
|
|
cached = self.game_state_cache.read_range(absolute, length)
|
|
if cached is not None and len(cached) == length:
|
|
return cached
|
|
return self.ff12.read_bytes(absolute, length)
|
|
|
|
def ff12_read_byte(self, address, use_base=True):
|
|
return int.from_bytes(self._read_bytes(address, 1, use_base), "little")
|
|
|
|
def ff12_read_bit(self, address, bit, use_base=True) -> bool:
|
|
return (self.ff12_read_byte(address, use_base) >> bit) & 1 == 1
|
|
|
|
def ff12_read_short(self, address, use_base=True):
|
|
return int.from_bytes(self._read_bytes(address, 2, use_base), "little")
|
|
|
|
def ff12_read_int(self, address, use_base=True):
|
|
return int.from_bytes(self._read_bytes(address, 4, use_base), "little")
|
|
|
|
def on_package(self, cmd: str, args: dict):
|
|
|
|
if cmd in {"Connected"}:
|
|
asyncio.create_task(self.send_msgs([{"cmd": "GetDataPackage", "games": ["Final Fantasy 12 Open World"]}]))
|
|
self.ff12slotdata = args['slot_data']
|
|
self.locations_checked = set(args["checked_locations"])
|
|
|
|
if cmd in {"ReceivedItems"}:
|
|
self.find_game()
|
|
if self.server_connected:
|
|
# Get the items past the start index in args items
|
|
for index, item in enumerate(args["items"], start=args["index"]):
|
|
if index >= len(self.ff12_items_received):
|
|
self.ff12_items_received.append(item)
|
|
else:
|
|
self.ff12_items_received[index] = item
|
|
|
|
if cmd in {"RoomUpdate"}:
|
|
if "checked_locations" in args:
|
|
new_locations = set(args["checked_locations"])
|
|
self.locations_checked |= new_locations
|
|
|
|
if cmd in {"DataPackage"}:
|
|
self.find_game()
|
|
self.server_connected = True
|
|
self.delete_communication_files()
|
|
asyncio.create_task(self.send_msgs([{'cmd': 'Sync'}]))
|
|
|
|
if cmd in {"RoomInfo"}:
|
|
if not os.path.exists(self.game_communication_path):
|
|
os.makedirs(self.game_communication_path)
|
|
|
|
super(FF12OpenWorldContext, self).on_package(cmd, args)
|
|
|
|
def find_game(self):
|
|
if not self.ff12connected:
|
|
try:
|
|
self.ff12 = pymem.Pymem(process_name="FFXII_TZA")
|
|
logger.info("You are now auto-tracking")
|
|
self.ff12connected = True
|
|
except Exception:
|
|
if self.ff12connected:
|
|
self.ff12connected = False
|
|
logger.info("Game is not open (Try running the client as an admin).")
|
|
|
|
def delete_communication_files(self):
|
|
if os.path.exists(self.game_communication_path):
|
|
for filename in os.listdir(self.game_communication_path):
|
|
file_path = os.path.join(self.game_communication_path, filename)
|
|
try:
|
|
if os.path.isfile(file_path):
|
|
os.remove(file_path)
|
|
except Exception as e:
|
|
logger.info(e)
|
|
|
|
async def update_game_state_cache(self):
|
|
if not self.ff12connected or not self.ff12:
|
|
return
|
|
|
|
new_cache = FF12StateCache()
|
|
try:
|
|
save_addr = self._compute_save_data_address()
|
|
if not save_addr:
|
|
return
|
|
|
|
new_cache.save_data_base = save_addr
|
|
new_cache.save_data = self.ff12.read_bytes(save_addr, FF12StateCache.SAVE_DATA_LENGTH)
|
|
if not new_cache.save_data:
|
|
return
|
|
new_cache.scenario_flag = int.from_bytes(new_cache.save_data[0:2], "little")
|
|
|
|
base = self.ff12.base_address
|
|
|
|
def capture_segment(offset: int, length: int) -> bytes:
|
|
data = self.ff12.read_bytes(base + offset, length)
|
|
new_cache.add_extra_segment(base + offset, data)
|
|
return data
|
|
|
|
normal_items = capture_segment(0x02097054, FF12StateCache.ITEM_SECTION_LENGTH)
|
|
equipment_items = capture_segment(0x020970D4, FF12StateCache.ITEM_SECTION_LENGTH)
|
|
loot_items = capture_segment(0x0209741C, FF12StateCache.ITEM_SECTION_LENGTH)
|
|
key_item_bits = capture_segment(0x0209784C, FF12StateCache.BITFIELD_SECTION_LENGTH)
|
|
esper_bits = capture_segment(0x0209788C, FF12StateCache.BITFIELD_SECTION_LENGTH)
|
|
magick_bits = capture_segment(0x0209781C, FF12StateCache.BITFIELD_SECTION_LENGTH)
|
|
technick_bits = capture_segment(0x02097828, FF12StateCache.BITFIELD_SECTION_LENGTH)
|
|
|
|
party_ptr = int.from_bytes(
|
|
self.ff12.read_bytes(self.ff12.base_address + 0x02D9F190, 4), "little")
|
|
new_cache.party_address = party_ptr + 0x08 if party_ptr else 0
|
|
|
|
pointer1 = int.from_bytes(
|
|
self.ff12.read_bytes(self.ff12.base_address + 0x01E5FFE0, 4), "little")
|
|
if pointer1:
|
|
new_cache.current_game_state = int.from_bytes(
|
|
self.ff12.read_bytes(pointer1 + 0x3A, 1), "little")
|
|
new_cache.current_map = int.from_bytes(
|
|
self.ff12.read_bytes(self.ff12.base_address + 0x020454C4, 2), "little")
|
|
|
|
def read_short_from(data: bytes, offset: int) -> int:
|
|
if 0 <= offset + 1 < len(data):
|
|
return int.from_bytes(data[offset:offset + 2], "little")
|
|
return 0
|
|
|
|
def read_flag_from(data: bytes, index: int) -> int:
|
|
byte_index = index // 8
|
|
bit_index = index % 8
|
|
if 0 <= byte_index < len(data):
|
|
return (data[byte_index] >> bit_index) & 1
|
|
return 0
|
|
|
|
inventory_by_code: Dict[int, int] = {}
|
|
inventory_by_name: Dict[str, int] = {}
|
|
for name, item in item_data_table.items():
|
|
code = item.code - 1
|
|
count = 0
|
|
if code < 0x1000:
|
|
count = read_short_from(normal_items, code * 2)
|
|
elif code < 0x2000:
|
|
count = read_short_from(equipment_items, (code - 0x1000) * 2)
|
|
elif 0x2000 <= code < 0x3000:
|
|
count = read_short_from(loot_items, (code - 0x2000) * 2)
|
|
elif 0x8000 <= code < 0x9000:
|
|
count = read_flag_from(key_item_bits, code - 0x8000)
|
|
elif 0xC000 <= code < 0xD000:
|
|
count = read_flag_from(esper_bits, code - 0xC000)
|
|
elif 0x3000 <= code < 0x4000:
|
|
count = read_flag_from(magick_bits, code - 0x3000)
|
|
elif 0x4000 <= code < 0x5000:
|
|
count = read_flag_from(technick_bits, code - 0x4000)
|
|
inventory_by_code[code] = count
|
|
inventory_by_name[name] = count
|
|
new_cache.inventory_by_code = inventory_by_code
|
|
new_cache.inventory_by_name = inventory_by_name
|
|
|
|
party_members = set()
|
|
if new_cache.party_address:
|
|
try:
|
|
party_blob = self.ff12.read_bytes(new_cache.party_address, 0x1C8 * MAX_PARTY_MEMBERS)
|
|
for chara in range(MAX_PARTY_MEMBERS):
|
|
base_index = chara * 0x1C8
|
|
if base_index < len(party_blob) and party_blob[base_index] & 0x10:
|
|
party_members.add(chara)
|
|
except Exception:
|
|
pass
|
|
new_cache.party_members = party_members
|
|
|
|
scenario_flag = new_cache.scenario_flag
|
|
if 0x37A <= scenario_flag <= 0x44C:
|
|
new_cache.leviathan_progress = scenario_flag
|
|
else:
|
|
lev_flag = new_cache.save_short(0xDFF7)
|
|
if lev_flag > 10000:
|
|
new_cache.leviathan_progress = lev_flag - 10000
|
|
elif lev_flag == 0:
|
|
new_cache.leviathan_progress = 0
|
|
else:
|
|
new_cache.leviathan_progress = lev_flag
|
|
|
|
esc_flag = new_cache.save_short(0xDFF4)
|
|
if new_cache.save_byte(0xA04) >= 2:
|
|
new_cache.escape_progress = 0x208
|
|
elif 0x11D < scenario_flag < 0x208:
|
|
new_cache.escape_progress = scenario_flag
|
|
elif 0x11D < esc_flag < 0x208:
|
|
new_cache.escape_progress = esc_flag
|
|
elif new_cache.save_byte(0xA06) >= 2:
|
|
new_cache.escape_progress = 0x11D
|
|
elif 6110 < scenario_flag <= 6110 + 70:
|
|
new_cache.escape_progress = scenario_flag - 6110
|
|
elif 6110 < esc_flag <= 6110 + 70:
|
|
new_cache.escape_progress = esc_flag - 6110
|
|
else:
|
|
new_cache.escape_progress = 0
|
|
|
|
if 0xD48 <= scenario_flag <= 0x1036:
|
|
new_cache.draklor_progress = scenario_flag
|
|
else:
|
|
darklor_flag = new_cache.save_short(0xDFF9)
|
|
new_cache.draklor_progress = 0 if darklor_flag == 0 else darklor_flag
|
|
|
|
self.game_state_cache = new_cache
|
|
except Exception as e:
|
|
if self.ff12connected:
|
|
self.ff12connected = False
|
|
logger.info(e)
|
|
|
|
def get_party_address(self) -> int:
|
|
if self.game_state_cache.party_address:
|
|
return self.game_state_cache.party_address
|
|
return self.ff12_read_int(0x02D9F190) + 0x08
|
|
|
|
def get_save_data_address(self) -> int:
|
|
if self.game_state_cache.save_data_base:
|
|
return self.game_state_cache.save_data_base
|
|
return self._compute_save_data_address()
|
|
|
|
def get_scenario_flag(self) -> int:
|
|
if self.game_state_cache.scenario_flag >= 0:
|
|
return self.game_state_cache.scenario_flag
|
|
return self.ff12_read_short(0x02044480)
|
|
|
|
def get_item_count_received(self, item_name: str) -> int:
|
|
return len([item for item in self.ff12_items_received[:self.get_item_index()] if
|
|
item.item == item_data_table[item_name].code])
|
|
|
|
def get_item_index(self) -> int:
|
|
return self.ff12_read_int(self.get_save_data_address() + 0x696, use_base=False)
|
|
|
|
def has_item_received(self, item_name: str) -> bool:
|
|
return self.get_item_count_received(item_name) > 0
|
|
|
|
def inventory_count(self, item_name: str) -> int:
|
|
return self.game_state_cache.inventory_by_name.get(item_name, 0)
|
|
|
|
def inventory_has(self, item_name: str) -> bool:
|
|
return self.inventory_count(item_name) > 0
|
|
|
|
def get_current_map(self) -> int:
|
|
if self.game_state_cache.current_map >= 0:
|
|
return self.game_state_cache.current_map
|
|
return self.ff12_read_short(0x20454C4)
|
|
|
|
def get_current_game_state(self) -> int:
|
|
if self.game_state_cache.current_game_state >= 0:
|
|
return self.game_state_cache.current_game_state
|
|
pointer1 = self.ff12_read_int(0x01E5FFE0)
|
|
return self.ff12_read_byte(pointer1 + 0x3A, False)
|
|
|
|
async def ff12_check_locations(self):
|
|
try:
|
|
self.sending.clear()
|
|
index = 0
|
|
for location_name, data in location_data_table.items():
|
|
index += 1
|
|
if data.address in self.locations_checked:
|
|
continue
|
|
|
|
# Check if the game is in a state where the location can be checked
|
|
map_id = self.get_current_map()
|
|
game_state = self.get_current_game_state()
|
|
scenario_flag = self.get_scenario_flag()
|
|
if (map_id == 0 or map_id > 0xFFFF or map_id <= 12 or
|
|
map_id == 274 or game_state != 0 or scenario_flag < 45):
|
|
break
|
|
|
|
if data.type == "inventory":
|
|
if int(data.str_id) in self.game_state_cache.party_members:
|
|
self.sending.append(data.address)
|
|
elif data.type == "reward":
|
|
if self.is_reward_met(data):
|
|
self.sending.append(data.address)
|
|
elif data.type == "treasure":
|
|
treasures: list[str] = self.ff12slotdata["treasures"]
|
|
if location_name not in treasures:
|
|
continue
|
|
treasure_index = treasures.index(location_name)
|
|
byte_index = treasure_index // 8
|
|
bit_index = treasure_index % 8
|
|
treasure_byte = self.game_state_cache.save_byte(0x14B4 + byte_index)
|
|
if (treasure_byte >> bit_index) & 1:
|
|
self.sending.append(data.address)
|
|
|
|
self.locations_checked |= set(self.sending)
|
|
|
|
# Victory, Final Boss
|
|
if self.game_state_cache.save_byte(0xA2E) >= 2 and not self.finished_game:
|
|
await self.send_msgs([{"cmd": "StatusUpdate", "status": ClientStatus.CLIENT_GOAL}])
|
|
self.finished_game = True
|
|
|
|
if len(self.sending) > 0:
|
|
message = [{"cmd": 'LocationChecks', "locations": self.sending}]
|
|
await self.send_msgs(message)
|
|
|
|
# Poptracker stuff
|
|
map_id = self.get_current_map()
|
|
if map_id != self.stored_map_id and map_id > 12 and map_id < 0xFFFF and map_id != 274:
|
|
# Send Bounce with new ID
|
|
self.stored_map_id = map_id
|
|
await self.send_msgs([{
|
|
"cmd": "Bounce",
|
|
"slots": [self.slot],
|
|
"data": {
|
|
"type": "MapUpdate",
|
|
"mapId": map_id,
|
|
},
|
|
}])
|
|
|
|
events_changed = False
|
|
for offset in tracker_event_offsets:
|
|
val = self.game_state_cache.save_byte(offset)
|
|
if self.event_flags.setdefault(offset, 0) != val:
|
|
self.event_flags[offset] = val
|
|
events_changed = True
|
|
|
|
if events_changed:
|
|
await self.send_msgs(
|
|
[
|
|
{
|
|
"cmd": "Set",
|
|
"key": f"ffxiiow_events_{self.team}_{self.slot}",
|
|
"default": {},
|
|
"want_reply": False,
|
|
"operations": [{"operation": "update", "value": self.event_flags}],
|
|
}
|
|
]
|
|
)
|
|
|
|
except Exception as e:
|
|
if self.ff12connected:
|
|
self.ff12connected = False
|
|
logger.info(e)
|
|
|
|
def is_reward_met(self, location_data: FF12OpenWorldLocationData) -> bool:
|
|
save = self.game_state_cache
|
|
scen = self.get_scenario_flag()
|
|
|
|
if location_data.str_id == "9000" or \
|
|
location_data.str_id == "916B" or \
|
|
location_data.str_id == "916C": # Tomaj Checks
|
|
return scen >= 6110
|
|
elif location_data.str_id == "9002": # Shadestone check
|
|
return save.save_bit(0xA42, 0)
|
|
elif location_data.str_id == "9001": # Sunstone check (if received Shadestone but the item is lost)
|
|
return self.has_item_received("Shadestone") and not self.inventory_has("Shadestone")
|
|
elif location_data.str_id == "905E": # Crescent Stone (if received Sunstone but the item is lost)
|
|
return self.has_item_received("Sunstone") and not self.inventory_has("Sunstone")
|
|
elif location_data.str_id == "905F": # Dalan SotO
|
|
return save.save_bit(0xA42, 1)
|
|
elif location_data.str_id == "911E": # SotO turn in
|
|
return self.has_item_received("Sword of the Order") and not self.inventory_has("Sword of the Order")
|
|
elif location_data.str_id == "9060": # Judges Boss
|
|
return save.save_byte(0xA27) >= 2
|
|
elif location_data.str_id == "9061": # Systems Access Key
|
|
return save.save_bit(0x14D4 + 4, 0)
|
|
elif location_data.str_id == "912C": # Manufacted Nethicite
|
|
return self.game_state_cache.leviathan_progress >= 0x3E8
|
|
elif location_data.str_id == "912D": # Eksir Berries
|
|
return save.save_bit(0xA42, 2)
|
|
elif location_data.str_id == "9190": # Belias Boss
|
|
return save.save_byte(0xA19) >= 2
|
|
elif location_data.str_id == "912E": # Dawn Shard
|
|
return save.save_bit(0xA42, 3)
|
|
elif location_data.str_id == "918E": # Vossler Boss
|
|
return save.save_byte(0xA3B) >= 2
|
|
elif location_data.str_id == "912F": # Goddess's Magicite
|
|
return self.game_state_cache.escape_progress >= 15
|
|
elif location_data.str_id == "9130": # Tube Fuse
|
|
return self.game_state_cache.escape_progress >= 0x13F
|
|
elif location_data.str_id == "911F": # Garif Reward
|
|
return save.save_bit(0xA42, 4)
|
|
elif location_data.str_id == "9131": # Lente's Tear (Tiamat Boss)
|
|
return save.save_byte(0xA08) >= 2
|
|
elif location_data.str_id == "9191": # Mateus Boss
|
|
return save.save_byte(0xA21) >= 2
|
|
elif location_data.str_id == "9132": # Sword of Kings
|
|
return save.save_bit(0xA42, 6)
|
|
elif location_data.str_id == "9133": # Start Mandragoras
|
|
# Kid or Dad
|
|
return self.game_state_cache.save_byte(0x684) == 1 or \
|
|
self.game_state_cache.save_byte(0x681) == 1
|
|
elif location_data.str_id == "9052": # Turn in Mandragoras
|
|
return self.game_state_cache.save_byte(0x683) == 1
|
|
elif location_data.str_id == "918D": # Cid 1 Boss
|
|
return self.game_state_cache.save_byte(0xA29) >= 2
|
|
elif 0x9134 <= int(location_data.str_id, 16) <= 0x914F: # Pinewood Chops
|
|
return (save.save_byte(0xDFF6) >
|
|
int(location_data.str_id, 16) - 0x9134)
|
|
elif location_data.str_id == "9150": # Sandalwood Chop
|
|
return save.save_bit(0xA42, 7)
|
|
elif location_data.str_id == "9151": # Lab Access Card
|
|
return self.game_state_cache.draklor_progress >= 0xD48
|
|
elif location_data.str_id == "9192": # Shemhazai Boss
|
|
return self.game_state_cache.save_byte(0xA20) >= 2
|
|
elif location_data.str_id == "9152": # Treaty Blade
|
|
return save.save_bit(0xDFFB, 0)
|
|
elif 0x9153 <= int(location_data.str_id, 16) <= 0x916A: # Black Orbs
|
|
return (save.save_byte(0xDFFC) >
|
|
int(location_data.str_id, 16) - 0x9153)
|
|
elif location_data.str_id == "9193": # Hashmal Boss
|
|
return self.game_state_cache.save_byte(0xA1F) >= 2
|
|
elif location_data.str_id == "918F": # Cid 2 Boss
|
|
return self.game_state_cache.save_byte(0xA2A) >= 2
|
|
elif location_data.str_id == "9003": # Hunt 1
|
|
return self.read_hunt_progress(0) >= 70
|
|
elif location_data.str_id == "9004": # Hunt 2
|
|
return self.read_hunt_progress(1) >= 70
|
|
elif location_data.str_id == "9005": # Hunt 3
|
|
return self.read_hunt_progress(2) >= 90
|
|
elif location_data.str_id == "9006": # Hunt 4
|
|
return self.read_hunt_progress(3) >= 100
|
|
elif location_data.str_id == "9007": # Hunt 5
|
|
return self.read_hunt_progress(4) >= 90
|
|
elif location_data.str_id == "9008": # Hunt 6
|
|
return self.read_hunt_progress(5) >= 100
|
|
elif location_data.str_id == "9009": # Hunt 7
|
|
return self.read_hunt_progress(6) >= 100
|
|
elif location_data.str_id == "900A": # Hunt 8
|
|
return self.read_hunt_progress(7) >= 100
|
|
elif location_data.str_id == "900B": # Hunt 9
|
|
return self.read_hunt_progress(8) >= 100
|
|
elif location_data.str_id == "900C": # Hunt 10
|
|
return self.read_hunt_progress(9) >= 100
|
|
elif location_data.str_id == "900D": # Hunt 11
|
|
return self.read_hunt_progress(10) >= 100
|
|
elif location_data.str_id == "900E": # Hunt 12
|
|
return self.read_hunt_progress(11) >= 100
|
|
elif location_data.str_id == "900F": # Hunt 13
|
|
return self.read_hunt_progress(12) >= 90
|
|
elif location_data.str_id == "9010": # Hunt 14
|
|
return self.read_hunt_progress(13) >= 100
|
|
elif location_data.str_id == "9011": # Hunt 15
|
|
return self.read_hunt_progress(14) >= 100
|
|
elif location_data.str_id == "9012": # Hunt 16
|
|
return self.read_hunt_progress(15) >= 90
|
|
elif location_data.str_id == "9013": # Hunt 17
|
|
return self.read_hunt_progress(16) >= 50
|
|
elif location_data.str_id == "9014": # Hunt 18
|
|
return self.read_hunt_progress(17) >= 50
|
|
elif location_data.str_id == "9015": # Hunt 19
|
|
return self.read_hunt_progress(18) >= 100
|
|
elif location_data.str_id == "9016": # Hunt 20
|
|
return self.read_hunt_progress(19) >= 150
|
|
elif location_data.str_id == "9017": # Hunt 21
|
|
return self.read_hunt_progress(20) >= 150
|
|
elif location_data.str_id == "9018": # Hunt 22
|
|
return self.read_hunt_progress(21) >= 150
|
|
elif location_data.str_id == "9019": # Hunt 23
|
|
return self.read_hunt_progress(22) >= 150
|
|
elif location_data.str_id == "901A": # Hunt 24
|
|
return self.read_hunt_progress(23) >= 50
|
|
elif location_data.str_id == "901B": # Hunt 25
|
|
return self.read_hunt_progress(24) >= 50
|
|
elif location_data.str_id == "901C": # Hunt 26
|
|
return self.read_hunt_progress(25) >= 90
|
|
elif location_data.str_id == "901D": # Hunt 27
|
|
return self.read_hunt_progress(26) >= 90
|
|
elif location_data.str_id == "901E": # Hunt 28
|
|
return self.read_hunt_progress(27) >= 90
|
|
elif location_data.str_id == "901F": # Hunt 29
|
|
return self.read_hunt_progress(28) >= 100
|
|
elif location_data.str_id == "9020": # Hunt 30
|
|
return self.read_hunt_progress(29) >= 100
|
|
elif location_data.str_id == "9021": # Hunt 31
|
|
return self.read_hunt_progress(30) >= 90
|
|
elif location_data.str_id == "9022": # Hunt 32
|
|
return self.read_hunt_progress(31) >= 150
|
|
elif location_data.str_id == "9023": # Hunt 33
|
|
return self.read_hunt_progress(32) >= 100
|
|
elif location_data.str_id == "9024": # Hunt 34
|
|
return self.read_hunt_progress(33) >= 90
|
|
elif location_data.str_id == "9025": # Hunt 35
|
|
return self.read_hunt_progress(34) >= 100
|
|
elif location_data.str_id == "9026": # Hunt 36
|
|
return self.read_hunt_progress(35) >= 100
|
|
elif location_data.str_id == "9027": # Hunt 37
|
|
return self.read_hunt_progress(36) >= 90
|
|
elif location_data.str_id == "9028": # Hunt 38
|
|
return self.read_hunt_progress(37) >= 110
|
|
elif location_data.str_id == "9029": # Hunt 39
|
|
return self.read_hunt_progress(38) >= 50
|
|
elif location_data.str_id == "902A": # Hunt 40
|
|
return self.read_hunt_progress(39) >= 130
|
|
elif location_data.str_id == "902B": # Hunt 42
|
|
return self.read_hunt_progress(40) >= 100
|
|
elif location_data.str_id == "902C": # Hunt 43
|
|
return self.read_hunt_progress(41) >= 150
|
|
elif location_data.str_id == "902D": # Hunt 44
|
|
return self.read_hunt_progress(42) >= 100
|
|
elif location_data.str_id == "902E": # Hunt 45
|
|
return self.read_hunt_progress(43) >= 100
|
|
elif location_data.str_id == "9122": # Hunt 41
|
|
return self.read_hunt_progress(44) >= 100
|
|
elif 0x902F <= int(location_data.str_id, 16) <= 0x903A: # Clan Rank Rewards
|
|
return (self.game_state_cache.save_byte(0x418) >
|
|
int(location_data.str_id, 16) - 0x902F)
|
|
elif location_data.str_id == "903B": # Clan Boss Flans
|
|
return self.game_state_cache.save_bit(0x419, 0)
|
|
elif location_data.str_id == "903C": # Clan Boss Firemane
|
|
return self.game_state_cache.save_bit(0x419, 1)
|
|
elif location_data.str_id == "903D": # Clan Boss Earth Tyrant
|
|
return self.game_state_cache.save_bit(0x419, 2)
|
|
elif location_data.str_id == "903E": # Clan Boss Mimic Queen
|
|
return self.game_state_cache.save_bit(0x419, 3)
|
|
elif location_data.str_id == "903F": # Clan Boss Demon Wall 1
|
|
return self.game_state_cache.save_bit(0x419, 4)
|
|
elif location_data.str_id == "9040": # Clan Boss Demon Wall 2
|
|
return self.game_state_cache.save_bit(0x419, 5)
|
|
elif location_data.str_id == "9041": # Clan Boss Elder Wyrm
|
|
return self.game_state_cache.save_bit(0x419, 6)
|
|
elif location_data.str_id == "9042": # Clan Boss Tiamat
|
|
return self.game_state_cache.save_bit(0x419, 7)
|
|
elif location_data.str_id == "9043": # Clan Boss Vinuskar
|
|
return self.game_state_cache.save_bit(0x41A, 0)
|
|
elif location_data.str_id == "9044": # Clan Boss King Bomb
|
|
return self.game_state_cache.save_bit(0x41A, 1)
|
|
elif location_data.str_id == "9045": # Clan Boss Mandragoras
|
|
return self.game_state_cache.save_bit(0x41A, 3)
|
|
elif location_data.str_id == "9046": # Clan Boss Ahriman
|
|
return self.game_state_cache.save_bit(0x41A, 2)
|
|
elif location_data.str_id == "9047": # Clan Boss Hell Wyrm
|
|
return self.game_state_cache.save_bit(0x41A, 4)
|
|
elif location_data.str_id == "9048": # Clan Boss Rafflesia
|
|
return self.game_state_cache.save_bit(0x41A, 5)
|
|
elif location_data.str_id == "9049": # Clan Boss Daedalus
|
|
return self.game_state_cache.save_bit(0x41A, 6)
|
|
elif location_data.str_id == "904A": # Clan Boss Tyrant
|
|
return self.game_state_cache.save_bit(0x41A, 7)
|
|
elif location_data.str_id == "904B": # Clan Boss Hydro
|
|
return self.game_state_cache.save_bit(0x41B, 0)
|
|
elif location_data.str_id == "904C": # Clan Boss Humbaba Mistant
|
|
return self.game_state_cache.save_bit(0x41B, 1)
|
|
elif location_data.str_id == "904D": # Clan Boss Fury
|
|
return self.game_state_cache.save_bit(0x41B, 2)
|
|
elif location_data.str_id == "905A": # Clan Boss Omega Mark XII
|
|
return self.game_state_cache.save_bit(0x41B, 3)
|
|
elif 0x904E <= int(location_data.str_id, 16) <= 0x9051: # Clan Espers (1,4,8,13)
|
|
return (self.game_state_cache.save_byte(0x41C) >
|
|
int(location_data.str_id, 16) - 0x904E)
|
|
elif location_data.str_id == "916D": # Flowering Cactoid Drop
|
|
return self.game_state_cache.save_byte(0x1064 + 130) >= 70
|
|
elif location_data.str_id == "916E": # Barheim Key
|
|
return self.game_state_cache.save_byte(0x68B) >= 11
|
|
elif location_data.str_id == "9081": # Deliver Cactus Flower
|
|
return self.game_state_cache.save_byte(0x68B) >= 3
|
|
elif location_data.str_id == "908A": # Cactus Family
|
|
return self.game_state_cache.save_byte(0x686) >= 7
|
|
elif location_data.str_id == "916F": # Get Stone of the Condemner
|
|
return self.game_state_cache.save_byte(0x680) >= 1
|
|
elif location_data.str_id == "9170": # Get Wind Globe
|
|
return self.game_state_cache.save_byte(0x1064 + 53) >= 50
|
|
elif location_data.str_id == "9171": # Get Windvane
|
|
return self.game_state_cache.save_byte(0x1064 + 53) >= 60
|
|
elif location_data.str_id == "9172": # White Mousse Drop
|
|
return self.game_state_cache.save_byte(0x1064 + 133) >= 50
|
|
elif location_data.str_id == "9173": # Sluice Gate Key
|
|
return self.game_state_cache.save_byte(0x1064 + 133) >= 120
|
|
elif location_data.str_id == "9174": # Enkelados Drop
|
|
return self.game_state_cache.save_byte(0x1064 + 137) >= 50
|
|
elif location_data.str_id == "9062": # Give Errmonea Leaf
|
|
return self.game_state_cache.save_byte(0x4AE) >= 1
|
|
elif location_data.str_id == "9175": # Merchant's Armband
|
|
return self.game_state_cache.save_byte(0x6FD) >= 2
|
|
elif location_data.str_id == "9176": # Get Pilika's Diary
|
|
return self.game_state_cache.save_byte(0x6FD) >= 3
|
|
elif location_data.str_id == "908D": # Give Pilika's Diary
|
|
return self.game_state_cache.save_byte(0x6FD) >= 4
|
|
elif location_data.str_id == "9177": # Vorpal Bunny Drop
|
|
return self.game_state_cache.save_byte(0x1064 + 141) >= 50
|
|
elif location_data.str_id == "9178": # Croakadile Drop
|
|
return self.game_state_cache.save_byte(0x1064 + 138) >= 50
|
|
elif location_data.str_id == "9179": # Lindwyrm Drop
|
|
return self.game_state_cache.save_byte(0x1064 + 149) >= 100
|
|
elif location_data.str_id == "917A": # Get Silent Urn
|
|
return self.game_state_cache.save_byte(0x1064 + 163) >= 50
|
|
elif location_data.str_id == "917B": # Orthros Drop
|
|
return self.game_state_cache.save_byte(0x1064 + 162) >= 70
|
|
elif location_data.str_id == "917D": # Site 3 Key
|
|
return self.game_state_cache.save_byte(0x1064 + 165) >= 50
|
|
elif location_data.str_id == "917E": # Site 11 Key
|
|
return self.game_state_cache.save_bit(0xDFFB, 2)
|
|
elif location_data.str_id == "917F": # Fafnir Drop
|
|
return self.game_state_cache.save_byte(0x1064 + 158) >= 70
|
|
elif location_data.str_id == "9180": # Marilith Drop
|
|
return self.game_state_cache.save_byte(0x1064 + 136) >= 70
|
|
elif location_data.str_id == "9181": # Vyraal Drop
|
|
return self.game_state_cache.save_byte(0x1064 + 148) >= 100
|
|
elif location_data.str_id == "9182": # Dragon Scale
|
|
return self.game_state_cache.save_byte(0x1064 + 148) >= 150
|
|
elif location_data.str_id == "9183": # Ageworn Key check (if received Dragon Scale but the item is lost)
|
|
return self.has_item_received("Dragon Scale") and not self.inventory_has("Dragon Scale")
|
|
elif location_data.str_id == "9184": # Ann's Letter
|
|
return self.game_state_cache.save_byte(0x5A6) >= 1
|
|
elif location_data.str_id == "906C": # Ann's Sisters
|
|
return self.game_state_cache.save_byte(0x5A6) >= 7
|
|
elif location_data.str_id == "9185": # Dusty Letter
|
|
return self.game_state_cache.save_bit(0x423, 2)
|
|
elif location_data.str_id == "917C": # Blackened Fragment
|
|
return self.game_state_cache.save_byte(0x1064 + 162) >= 100
|
|
elif location_data.str_id == "9186": # Dull Fragment
|
|
return self.game_state_cache.save_bit(0x423, 1)
|
|
elif location_data.str_id == "9187": # Grimy Fragment
|
|
return self.game_state_cache.save_byte(0x416) >= 7
|
|
elif location_data.str_id == "9188": # Moonsilver Medallion
|
|
return self.game_state_cache.save_byte(0x1064 + 59) >= 20
|
|
elif location_data.str_id == "9189" or \
|
|
location_data.str_id == "918A" or \
|
|
location_data.str_id == "918B": # Nabreus Medallions
|
|
return self.game_state_cache.save_byte(0x1064 + 57) >= 100
|
|
elif location_data.str_id == "918C": # Medallion of Might (Humbaba Mistant and Fury bosses)
|
|
return self.game_state_cache.save_byte(0xA0F) >= 2 and \
|
|
self.game_state_cache.save_byte(0xA10) >= 2
|
|
elif location_data.str_id == "9056": # Viera Rendevous
|
|
return self.game_state_cache.save_byte(0x40E) >= 6
|
|
elif location_data.str_id == "9058": # Ktjn Reward
|
|
return self.game_state_cache.save_bit(0x409, 0)
|
|
elif location_data.str_id == "906A": # Jovy Reward
|
|
return self.game_state_cache.save_byte(0x5B8) >= 6
|
|
elif location_data.str_id == "906E": # Outpost Glint 1
|
|
return self.game_state_cache.save_byte(0x691) >= 1
|
|
elif location_data.str_id == "906F": # Outpost Glint 2
|
|
return self.game_state_cache.save_byte(0x692) >= 1
|
|
elif location_data.str_id == "9057": # Outpost Glint 3
|
|
return self.game_state_cache.save_byte(0x693) >= 1
|
|
elif location_data.str_id == "9070": # Outpost Glint 4
|
|
return self.game_state_cache.save_byte(0x694) >= 1
|
|
elif location_data.str_id == "9059": # Outpost Glint 5
|
|
return self.game_state_cache.save_byte(0x695) >= 1
|
|
elif location_data.str_id == "908F": # Footrace
|
|
return self.game_state_cache.save_byte(0x73D) >= 1
|
|
elif location_data.str_id == "9194": # Adrammelech Boss
|
|
return save.save_byte(0xA25) >= 2
|
|
elif location_data.str_id == "9195": # Zalera Boss
|
|
return save.save_byte(0xA1D) >= 2
|
|
elif location_data.str_id == "9196": # Cuchulainn Boss
|
|
return save.save_byte(0xA1C) >= 2
|
|
elif location_data.str_id == "9197": # Zeromus Boss
|
|
return save.save_byte(0xA22) >= 2
|
|
elif location_data.str_id == "9198": # Exodus Boss
|
|
return save.save_byte(0xA23) >= 2
|
|
elif location_data.str_id == "9199": # Chaos Boss
|
|
return save.save_byte(0xA1A) >= 2
|
|
elif location_data.str_id == "919A": # Ultima Boss
|
|
return save.save_byte(0xA24) >= 2
|
|
elif location_data.str_id == "919B": # Zodiark Boss
|
|
return save.save_byte(0xA1B) >= 2
|
|
elif 0x9090 <= int(location_data.str_id, 16) <= 0x90AE: # Trophy Drops
|
|
trophy_index = int(location_data.str_id, 16) - 0x9090
|
|
return save.save_byte(0xC90 + trophy_index) >= 2
|
|
elif 0x90F9 <= int(location_data.str_id, 16) <= 0x90FE: # Rare Game Defeats (5,10,15,20,25,30)
|
|
return save.save_byte(0x725) > \
|
|
(int(location_data.str_id, 16) - 0x90F9) + 1
|
|
elif location_data.str_id == "90F3": # Atak >=16
|
|
if save.save_byte(0x1064 + 71) < 170:
|
|
return False
|
|
max_trophies = self.get_max_trophies()
|
|
return save.save_byte(0xB14) == max_trophies and \
|
|
max_trophies >= 16
|
|
elif location_data.str_id == "90F4": # Atak <16
|
|
if save.save_byte(0x1064 + 71) < 170:
|
|
return False
|
|
max_trophies = self.get_max_trophies()
|
|
return save.save_byte(0xB14) == max_trophies and \
|
|
max_trophies < 16
|
|
elif location_data.str_id == "90F5": # Blok >=16
|
|
if save.save_byte(0x1064 + 71) < 170:
|
|
return False
|
|
max_trophies = self.get_max_trophies()
|
|
return save.save_byte(0xB15) == max_trophies and \
|
|
max_trophies >= 16
|
|
elif location_data.str_id == "90F6": # Blok <16
|
|
if save.save_byte(0x1064 + 71) < 170:
|
|
return False
|
|
max_trophies = self.get_max_trophies()
|
|
return save.save_byte(0xB15) == max_trophies and \
|
|
max_trophies < 16
|
|
elif location_data.str_id == "90F7": # Stok >=16
|
|
if save.save_byte(0x1064 + 71) < 170:
|
|
return False
|
|
max_trophies = self.get_max_trophies()
|
|
return save.save_byte(0xB16) == max_trophies and \
|
|
max_trophies >= 16
|
|
elif location_data.str_id == "90F8": # Stok <16
|
|
if save.save_byte(0x1064 + 71) < 170:
|
|
return False
|
|
max_trophies = self.get_max_trophies()
|
|
return save.save_byte(0xB16) == max_trophies and \
|
|
max_trophies < 16
|
|
elif 0x90FF <= int(location_data.str_id, 16) <= 0x911D: # Hunt Club Outfitters
|
|
outfitter_index = int(location_data.str_id, 16) - 0x90FF
|
|
return save.save_byte(0xAF2 + outfitter_index) >= 1
|
|
raise Exception(f"Unknown reward location ID: {location_data.str_id}")
|
|
|
|
def read_hunt_progress(self, hunt_id: int) -> int:
|
|
value = self.game_state_cache.save_byte(0x1064 + 128 + hunt_id)
|
|
return value
|
|
|
|
def get_max_trophies(self):
|
|
return max(
|
|
self.game_state_cache.save_byte(0xB14),
|
|
self.game_state_cache.save_byte(0xB15),
|
|
self.game_state_cache.save_byte(0xB16))
|
|
|
|
async def give_items(self):
|
|
try:
|
|
# Write obtained items to txt files in the communication folder in format items_received_####.txt
|
|
cur_index = 0
|
|
for item in self.ff12_items_received:
|
|
file_path = os.path.join(
|
|
self.game_communication_path,
|
|
f"items_received_{cur_index:04d}.txt")
|
|
with open(file_path, "w") as f:
|
|
# Write the item ID and the amount of the item on new lines
|
|
item_id = item.item - 1
|
|
if item.item >= 1 + 98304: # Gil
|
|
item_id = 0xFFFE
|
|
item_count = item_data_table[inv_item_table[item.item]].amount
|
|
f.write(f"{item_id}\n{item_count}\n")
|
|
cur_index += 1
|
|
except Exception as e:
|
|
if self.ff12connected:
|
|
self.ff12connected = False
|
|
logger.info(e)
|
|
|
|
def make_gui(self):
|
|
"""Import kivy UI system and start running it as self.ui_task."""
|
|
ui = super().make_gui()
|
|
|
|
class FF12OpenWorldManager(ui):
|
|
logging_pairs = [
|
|
("Client", "Archipelago")
|
|
]
|
|
base_title = "Archipelago FF12 Open World Client"
|
|
|
|
return FF12OpenWorldManager
|
|
|
|
|
|
async def ff12_watcher(ctx: FF12OpenWorldContext):
|
|
while not ctx.exit_event.is_set():
|
|
try:
|
|
if ctx.ff12connected and ctx.server_connected:
|
|
await ctx.update_game_state_cache()
|
|
await ctx.ff12_check_locations()
|
|
await ctx.give_items()
|
|
elif not ctx.ff12connected and ctx.server_connected:
|
|
ctx.ff12 = None
|
|
last_check = time.time()
|
|
while not ctx.ff12connected and ctx.server_connected and not ctx.exit_event.is_set():
|
|
if time.time() - last_check > 15:
|
|
logger.info("Game Connection lost. Waiting 15 seconds until trying to reconnect.")
|
|
ctx.find_game()
|
|
last_check = time.time()
|
|
await asyncio.sleep(0.5)
|
|
except Exception as e:
|
|
if ctx.ff12connected:
|
|
ctx.ff12connected = False
|
|
logger.info(e)
|
|
await asyncio.sleep(0.5)
|
|
|
|
|
|
def launch(*launch_args):
|
|
async def main(args_in):
|
|
ctx = FF12OpenWorldContext(args_in.connect, args_in.password)
|
|
ctx.server_task = asyncio.create_task(server_loop(ctx), name="server loop")
|
|
if tracker_loaded:
|
|
ctx.run_generator()
|
|
if gui_enabled:
|
|
ctx.run_gui()
|
|
ctx.run_cli()
|
|
progression_watcher = asyncio.create_task(
|
|
ff12_watcher(ctx), name="FF12ProgressionWatcher")
|
|
|
|
await ctx.exit_event.wait()
|
|
ctx.server_address = None
|
|
|
|
await progression_watcher
|
|
|
|
await ctx.shutdown()
|
|
|
|
import colorama
|
|
|
|
parser = get_base_parser(description="FF12 Open World Client, for text interfacing.")
|
|
parser.add_argument("url", default="", type=str, nargs="?", help="Archipelago connection url")
|
|
|
|
args, rest = parser.parse_known_args(launch_args)
|
|
args = handle_url_arg(args, parser)
|
|
colorama.init()
|
|
asyncio.run(main(args))
|
|
colorama.deinit()
|