Initial commit

This commit is contained in:
2025-12-08 22:45:24 +01:00
commit 92289523ad
14 changed files with 1745 additions and 0 deletions

3
.gitignore vendored Normal file
View File

@@ -0,0 +1,3 @@
/build/target
.vscode/
__pycache__/

2
Sly3Client.py Normal file
View File

@@ -0,0 +1,2 @@
def launch_client():
pass

378
Sly3Interface.py Normal file
View File

@@ -0,0 +1,378 @@
from typing import Optional, Dict, NamedTuple
import struct
from logging import Logger
from enum import IntEnum
from .pcsx2_interface.pine import Pine
from .data.Constants import ADDRESSES, MENU_RETURN_DATA
class Sly3Episode(IntEnum):
Title_Screen = 0
An_Opera_of_Fear = 1
Rumble_Down_Under = 2
Flight_of_Fancy = 3
A_Cold_Alliance = 4
Dead_Men_Tell_No_Tales = 5
Honor_Among_Thieves = 6
class PowerUps(NamedTuple):
attack = False
binocucom = False
bombs = False
unknown = False
trigger_Bomb = False
fishing_pole = False
alarm_clock = False
adrenaline_burst = False
health_extractor = False
hover_pack = False
insanity_strike = False
grapple_cam = False
size_destabilizer = False
rage_bomb = False
reduction_bomb = False
ball_form = False
berserker_charge = False
juggernaut_throw = False
guttural_roar = False
fists_of_flame = False
temporal_lock = False
raging_inferno_flop = False
diablo_fire_slam = False
smoke_bomb = False
combat_dodge = False
paraglider = False
silent_obliteration = False
feral_pounce = False
mega_jump = False
knockout_dive = False
shadow_power_1 = False
thief_reflexes = False
shadow_power_2 = False
rocket_boots = False
treasure_map = False
shield = False
venice_disguise = False
photographer_disguise = False
pirate_disguise = False
spin_1 = False
spin_2 = False
spin_3 = False
jump_1 = False
jump_2 = False
jump_3 = False
push_1 = False
push_2 = False
push_3 = False
class GameInterface():
"""
Base class for connecting with a pcsx2 game
"""
pcsx2_interface: Pine = Pine()
logger: Logger
game_id_error: Optional[str] = None
current_game: Optional[str] = None
addresses: Dict = {}
def __init__(self, logger) -> None:
self.logger = logger
def _read8(self, address: int):
return self.pcsx2_interface.read_int8(address)
def _read16(self, address: int):
return self.pcsx2_interface.read_int16(address)
def _read32(self, address: int):
return self.pcsx2_interface.read_int32(address)
def _read_bytes(self, address: int, n: int):
return self.pcsx2_interface.read_bytes(address, n)
def _read_float(self, address: int):
return struct.unpack("f",self.pcsx2_interface.read_bytes(address, 4))[0]
def _write8(self, address: int, value: int):
self.pcsx2_interface.write_int8(address, value)
def _write16(self, address: int, value: int):
self.pcsx2_interface.write_int16(address, value)
def _write32(self, address: int, value: int):
self.pcsx2_interface.write_int32(address, value)
def _write_bytes(self, address: int, value: bytes):
self.pcsx2_interface.write_bytes(address, value)
def _write_float(self, address: int, value: float):
self.pcsx2_interface.write_bytes(address, struct.pack("f", value))
def connect_to_game(self):
"""
Initializes the connection to PCSX2 and verifies it is connected to the
right game
"""
if not self.pcsx2_interface.is_connected():
self.pcsx2_interface.connect()
if not self.pcsx2_interface.is_connected():
return
self.logger.info("Connected to PCSX2 Emulator")
try:
game_id = self.pcsx2_interface.get_game_id()
# The first read of the address will be null if the client is faster than the emulator
self.current_game = None
if game_id in ADDRESSES.keys():
self.current_game = game_id
self.addresses = ADDRESSES[game_id]
if self.current_game is None and self.game_id_error != game_id and game_id != b'\x00\x00\x00\x00\x00\x00':
self.logger.warning(
f"Connected to the wrong game ({game_id})")
self.game_id_error = game_id
except RuntimeError:
pass
except ConnectionError:
pass
def disconnect_from_game(self):
self.pcsx2_interface.disconnect()
self.current_game = None
self.logger.info("Disconnected from PCSX2 Emulator")
def get_connection_state(self) -> bool:
try:
connected = self.pcsx2_interface.is_connected()
return connected and self.current_game is not None
except RuntimeError:
return False
class Sly3Interface(GameInterface):
def _reload(self, reload_data: bytes):
self._write_bytes(
self.addresses["reload values"],
reload_data
)
self._write32(self.addresses["reload"], 1)
def _get_job_address(self, task: int) -> int:
pointer = self._read32(self.addresses["DAG root"])
for _ in range(task):
pointer = self._read32(pointer+0x20)
return pointer
def get_current_map(self) -> int:
return self._read32(self.addresses["map id"])
def get_current_job(self) -> int:
return self._read32(self.addresses["job id"])
def set_current_job(self, job: int) -> None:
self._write32(self.addresses["job id"], job)
def set_items_received(self, n:int) -> None:
self._write32(self.addresses["items received"], n)
def to_episode_menu(self) -> None:
self.logger.info("Skipping to episode menu")
if (
self.get_current_map() == 35 and
self.get_current_job() == 1797
):
self.set_current_job(0xffffffff)
# self.set_items_received(0)
self._reload(bytes.fromhex(MENU_RETURN_DATA))
def unlock_episodes(self) -> None:
self._write8(self.addresses["episode unlocks"], 8)
def in_cutscene(self) -> bool:
frame_counter = self._read16(self.addresses["frame counter"])
return frame_counter > 10
def skip_cutscene(self) -> None:
pressing_x = self._read8(self.addresses["x pressed"]) == 255
if self.in_cutscene() and pressing_x:
self._write32(self.addresses["skip cutscene"],0)
if __name__ == "__main__":
interf = Sly3Interface(Logger("test"))
interf.connect_to_game()
byte_list = [
[
False, # ???
False, # ???
True, # Sly/Bentley square attack
True, # Binocucom
True, # Bentley's bombs
False, # ???
True, # Trigger Bomb
True # Fishing Pole
],
[
True, # Alarm Clock
True, # Adrenaline Burst
True, # Health Extractor
True, # Hover Pack (Doesn't activate until you reload)
True, # Insanity Strike
True, # Grapple Came
True, # Size Destabilizer
True # Rage Bomb
],
[
True, # Reduction Bomb
True, # Ball Form
True, # Berskerker Charge
True, # Juggernaut Throw
True, # Gutteral Roar
True, # Fists of Flame
True, # Temporal Lock
True # Raging Inferno Flop
],
[
True, # Diablo Fire Slam
True, # Smoke Bomb
True, # Combat Dodge
True, # Paraglider
True, # Silent Obliteration
True, # Feral Pounce
True, # Mega Jump
True # Knockout Dive
],
[
True, # Shadow Power Lvl 1
True, # Thief Reflexes
True, # Shadow Power Lvl 2
True, # Rocket Boots
True, # Treasure Map
False, # ???
True, # Venice Disguise
True # Photographer Disguise
],
[
True, # Pirate Disguise
True, # Spin Attack lvl 1
True, # Spin Attack lvl 2
True, # Spin Attack lvl 3
True, # Jump Attack lvl 1
True, # Jump Attack lvl 2
True, # Jump Attack lvl 3
True # Push Attack lvl 1
],
[
True, # Push Attack lvl 1
True, # Push Attack lvl 1
False,
False,
False,
False,
False,
False
],
[
False,
False,
False,
False,
False,
False,
False,
False
],
]
# byte_list = [[False for _ in range(8)] for _ in range(8)]
data = b''.join(
int(''.join(str(int(i)) for i in byte[::-1]),2).to_bytes(1,"big")
for byte in byte_list
)
interf._write_bytes(0x468DCC,data)
# data = interf._read_bytes(0x468DCC, 8)
# bits = [
# bool(int(b))
# for byte in data
# for b in f"{byte:08b}"[::-1]
# ]
# print(bits)
interf._write32(0x468DDC, 10000)
# thiefnet_values = list(range(9)) + list(range(10,))
# def read_text(address: int):
# text = ""
# while True:
# character = interf._read_bytes(address,2)
# if character == b"\x00\x00":
# break
# text += character.decode("utf-16-le")
# address += 2
# return text
# def find_string_id(_id: int):
# string_table_address = interf._read32(0x47A2D8)
# i = 0
# while True:
# string_id = interf._read32(string_table_address+i*8)
# if string_id == _id:
# return interf._read32(string_table_address+i*8+4)
# i += 1
# print(" {")
# for i in range(44):
# address = 0x343208+i*0x3c
# interf._write32(address,i+1)
# interf._write32(address+0xC,0)
# name_id = interf._read32(address+0x14)
# name_address = find_string_id(name_id)
# name_text = read_text(name_address)
# description_id = interf._read32(address+0x18)
# description_address = find_string_id(description_id)
# print(
# " " +
# f"\"{name_text}\": "+
# f"({hex(name_address)},{hex(description_address)}),"
# )
# print(" }")
# string_table_address = interf._read32(0x47A2D8)
# for i in range(10):
# print("----")
# string_id = interf._read32(string_table_address+i*8)
# print(string_id)
# string_address = interf._read32(string_table_address+i*8+4)
# print(hex(string_address))
# print(read_text(string_address))
# print(interf._read32(0x6b4110+0x44))
print(interf._read32(0x1365be0+0x44))
print()
print(interf._read32(0x1357f80+0x44))
print(interf._read32(0x1350560+0x44))
print(interf._read32(0x135aba0+0x44))
print(interf._read32(0x36DB98))

119
Sly3Options.py Normal file
View File

@@ -0,0 +1,119 @@
from Options import (
DeathLink,
StartInventoryPool,
PerGameCommonOptions,
Choice,
Toggle,
DefaultOnToggle,
Range,
OptionGroup
)
from dataclasses import dataclass
class StartingEpisode(Choice):
"""
Select Which episode to start with.
"""
display_name = "Starting Episode"
option_An_Opera_of_Fear = 0
option_Rumble_Down_Under = 1
option_Flight_of_Fancy = 2
option_A_Cold_Alliance = 3
option_Dead_Men_Tell_No_Tales = 4
default = 0
class Goal(Choice):
"""
Which boss you must defeat to goal.
"""
display_name = "Goal"
option_Don_Octavio = 0
option_Dark_Mask = 1
option_Black_Baron = 2
option_General_Tsao = 3
option_Captain_LeFwee = 4
option_Dr_M = 5
option_All_Bosses = 6
default = 5
class IncludeMegaJump(Toggle):
"""
Add the Mega Jump ability to the pool.
"""
display_name = "Include Mega Jump"
class CoinsMinimum(Range):
"""
The minimum number of coins you'll receive when you get a "Coins" filler
item.
"""
display_name = "Coins Minimum"
range_start = 0
range_end = 1000
default = 50
class CoinsMaximum(Range):
"""
The maximum number of coins you'll receive when you get a "Coins" filler
item.
"""
display_name = "Coins Maximum"
range_start = 0
range_end = 1000
default = 200
class ThiefNetCostMinimum(Range):
"""
The minimum number of coins items on ThiefNet will cost.
"""
display_name = "ThiefNet Cost Minimum"
range_start = 0
range_end = 9999
default = 200
class ThiefNetCostMaximum(Range):
"""
The maximum number of coins items on ThiefNet will cost.
"""
display_name = "ThiefNet Cost Maximum"
range_start = 0
range_end = 9999
default = 2000
@dataclass
class Sly3Options(PerGameCommonOptions):
start_inventory_from_pool: StartInventoryPool
death_link: DeathLink
starting_episode: StartingEpisode
goal: Goal
include_mega_jump: IncludeMegaJump
coins_minimum: CoinsMinimum
coins_maximum: CoinsMaximum
thiefnet_minimum: ThiefNetCostMinimum
thiefnet_maximum: ThiefNetCostMaximum
sly3_option_groups = [
OptionGroup("Goal",[
Goal
]),
OptionGroup("Items",[
IncludeMegaJump,
CoinsMinimum,
CoinsMaximum
]),
OptionGroup("Locations",[
ThiefNetCostMinimum,
ThiefNetCostMaximum
])
]

68
Sly3Pool.py Normal file
View File

@@ -0,0 +1,68 @@
import typing
from BaseClasses import Item
from .data.Constants import EPISODES
from .data.Items import item_groups
if typing.TYPE_CHECKING:
from . import Sly3World
def gen_powerups(world: "Sly3World") -> list[Item]:
"""Generate the power-ups for the item pool"""
powerups = []
for item_name in item_groups["Power-Up"]:
if item_name == "Mega Jump" and not world.options.include_mega_jump:
continue
elif item_name == "Progressive Shadow Power":
powerups.append(world.create_item(item_name))
powerups.append(world.create_item(item_name))
elif item_name[:11] == "Progressive":
powerups.append(world.create_item(item_name))
powerups.append(world.create_item(item_name))
powerups.append(world.create_item(item_name))
else:
powerups.append(world.create_item(item_name))
return powerups
def gen_crew(world: "Sly3World") -> list[Item]:
"""Generate the crew for the item pool"""
crew = []
for item_name in item_groups["Crew"]:
crew.append(world.create_item(item_name))
return crew
def gen_episodes(world: "Sly3World") -> list[Item]:
"""Generate the progressive episodes items for the item pool"""
all_episodes = [
item_name for item_name in item_groups["Episode"]
for _ in range(4)
]
# Make sure the starting episode is precollected
starting_episode_n = world.options.starting_episode.value
starting_episode = f"Progressive {list(EPISODES.keys())[starting_episode_n]}"
all_episodes.remove(starting_episode)
world.multiworld.push_precollected(world.create_item(starting_episode))
return [world.create_item(e) for e in all_episodes]
def gen_pool_sly3(world: "Sly3World") -> list[Item]:
"""Generate the item pool for the world"""
item_pool = []
item_pool += gen_powerups(world)
item_pool += gen_episodes(world)
item_pool += gen_crew(world)
unfilled_locations = world.multiworld.get_unfilled_locations(world.player)
remaining = len(unfilled_locations)-len(item_pool)
if world.options.goal.value < 6:
remaining -= 1
assert remaining >= 0, f"There are more items than locations ({len(item_pool)} items; {len(unfilled_locations)} locations)"
item_pool += [world.create_item(world.get_filler_item_name()) for _ in range(remaining)]
return item_pool

88
Sly3Regions.py Normal file
View File

@@ -0,0 +1,88 @@
import typing
from BaseClasses import Region, CollectionState, Location
from .data.Locations import location_dict
from .data.Constants import EPISODES, CHALLENGES
if typing.TYPE_CHECKING:
from . import Sly3World
from .Sly3Options import Sly3Options
def create_access_rule(episode: str, n: int, options: "Sly3Options", player: int):
"""Returns a function that checks if the player has access to a specific region"""
def rule(state: CollectionState):
access = True
item_name = f"Progressive {episode}"
if episode == "Honor Among Thieves":
access = access and state.count_group("Crew", player) == 7
else:
access = access and state.count(item_name, player) >= n
if n > 1:
requirements = sum({
"An Opera of Fear": [
[],
["Binocucom", "Bentley"],
["Carmelita", "Murray", "Ball Form", "Disguise (Venice)"]
],
"Rumble Down Under": [
[],
["Murray", "Guru"],
["Bentley"]
],
"Flight of Fancy": [
[],
["Murray", "Bentley", "Guru", "Fishing Pole", "Penelope"],
["Hover Pack", "Carmelita", "Binocucom"]
],
"A Cold Alliance": [
["Bentley", "Murray", "Guru", "Penelope", "Binocucom"],
["Disguise (Photographer)", "Grapple-Cam", "Panda King"],
["Carmelita"]
],
"Dead Men Tell No Tales": [
[],
["Bentley", "Penelope", "Grapple-Cam", "Murray", "Silent Obliteration", "Treasure Map"],
["Panda King", "Dimitri"]
]
}[episode][:n-2], [])
access = access and all(state.has(i, player) for i in requirements)
return access
return rule
def create_regions_sly3(world: "Sly3World"):
"""Creates a region for each chapter of each episode"""
menu = Region("Menu", world.player, world.multiworld)
menu.add_locations({
f"ThiefNet {i+1:02}": location_dict[f"ThiefNet {i+1:02}"].code
for i in range(37)
})
world.multiworld.regions.append(menu)
for i, episode in enumerate(EPISODES.keys()):
for n in range(1,5):
if n == 2 and episode == "Honor Among Thieves":
break
region = Region(f"Episode {i+1} ({n})", world.player, world.multiworld)
region.add_locations({
f"{episode} - {job}": location_dict[f"{episode} - {job}"].code
for job in EPISODES[episode][n-1]
})
region.add_locations({
f"{episode} - {challenge}": location_dict[f"{episode} - {challenge}"].code
for challenge in CHALLENGES[episode][n-1]
})
world.multiworld.regions.append(region)
menu.connect(
region,
None,
create_access_rule(episode, n, world.options, world.player)
)

208
Sly3Rules.py Normal file
View File

@@ -0,0 +1,208 @@
import typing
from math import ceil
from BaseClasses import CollectionState
from worlds.generic.Rules import add_rule
from .data.Constants import EPISODES
if typing.TYPE_CHECKING:
from . import Sly3World
def set_rules_sly3(world: "Sly3World"):
player = world.player
# Putting ThiefNet stuff out of logic, to make early game less slow.
# Divides the items into 8 groups of 3. First groups requires 2 episodes
# items to be in logic, second group requires 4, etc.
for i in range(1,35):
episode_items_n = ceil(i/4)*2
add_rule(
world.get_location(f"ThiefNet {i:02}"),
lambda state, n=episode_items_n: (
state.has_group("Episode", player, n)
)
)
def require(location: str, item: str|list[str]):
if isinstance(item,str):
add_rule(
world.get_location(location),
lambda state, i=item: (
state.has(i, player)
)
)
else:
add_rule(
world.get_location(location),
lambda state, i=item: (
all(state.has(j, player) for j in i)
)
)
### Job requirements
## An Opera of fear
# An Opera of Fear - Police HQ
require("An Opera of Fear - Octavio Snap", "Binocucom")
# An Opera of Fear - Into the Depths
require("An Opera of Fear - Canal Chase", "Bentley")
require("An Opera of Fear - Turf War!", "Carmelita")
require("An Opera of Fear - Tar Ball", ["Murray", "Ball Form"])
# An Opera of Fear - Run 'n Bomb
require("An Opera of Fear - Guard Duty", "Disguise (Venice)")
require("An Opera of Fear - Operation: Tar-Be Gone!", "Bombs")
## Rumble Down Under
# Rumble Down Under - Search for the Guru
require("Rumble Down Under - Spelunking", "Murray")
# Rumble Down Under - Dark Caves
# Rumble Down Under - Big Truck
require("Rumble Down Under - Unleash the Guru", "Guru")
# Rumble Down Under - The Claw
require("Rumble Down Under - Lemon Rage", "Bentley")
# Rumble Down Under - Hungry Croc
# Rumble Down Under - Operation: Moon Crash
## Flight of Fancy
# Flight of Fancy - Hidden Flight Roster
require("Flight of Fancy - Frame Team Belgium", ["Murray", "Bentley", "Guru", "Fishing Pole"])
require("Flight of Fancy - Frame Team Iceland", "Murray")
require("Flight of Fancy - Cooper Hangar Defense", "Penelope")
require("Flight of Fancy - ACES Semifinals", ["Murray", "Bentley", "Guru", "Fishing Pole", "Penelope"])
require("Flight of Fancy - Giant Wolf Massacre", "Binocucom")
require("Flight of Fancy - Windmill Firewall", "Hover Pack")
require("Flight of Fancy - Beauty and the Beast", "Carmelita")
require("Flight of Fancy - Operation: Turbo Dominant Eagle", "Paraglider")
## A Cold Alliance
require("A Cold Alliance - King of Fire", ["Bentley", "Murray", "Guru", "Penelope", "Binocucom"])
require("A Cold Alliance - Get a Job", "Disguise (Photographer)")
require("A Cold Alliance - Tearful Reunion", "Panda King")
require("A Cold Alliance - Grapple-Cam Break-In", "Grapple-Cam")
require("A Cold Alliance - Laptop Retrieval", ["Disguise (Photographer)", "Panda King", "Grapple-Cam"])
# A Cold Alliance - Vampiric Defense
# A Cold Alliance - Down the Line
require("A Cold Alliance - A Battery of Peril", "Carmelita")
# A Cold Alliance - Operation: Wedding Crasher
## Dead Men Tell No Tales
require("Dead Men Tell No Tales - The Talk of Pirates", "Disguise (Pirate)")
require("Dead Men Tell No Tales - Dynamic Duo", ["Bentley", "Penelope", "Grapple-Cam"])
require("Dead Men Tell No Tales - Jollyboat of Destruction", "Murray")
require("Dead Men Tell No Tales - X Marks the Spot", ["Bentley", "Penelope", "Grapple-Cam", "Murray", "Silent Obliteration", "Treasure Map"])
require("Dead Men Tell No Tales - Crusher from the Depths", "Panda King")
require("Dead Men Tell No Tales - Deep Sea Danger", "Dimitri")
# Dead Men Tell No Tales - Battle on the High Seas
require("Dead Men Tell No Tales - Operation: Reverse Double-Cross", "Guru")
## Honor Among Thieves
# Honor Among Thieves - Carmelita to the Rescue
# Honor Among Thieves - A Deadly Bite
# Honor Among Thieves - The Dark Current
# Honor Among Thieves - Bump-Charge-Jump
# Honor Among Thieves - Danger in the Skie
# Honor Among Thieves - The Ancestors' Gauntlet
# Honor Among Thieves - Stand your Ground
# Honor Among Thieves - Final Legacy
### Challenge requirements
## An Opera of Fear
require("An Opera of Fear - Canal Chase - Expert Course", "Bentley")
require("An Opera of Fear - Air Time", ["Murray", "Ball Form"])
# An Opera of Fear - Tower Scramble
# An Opera of Fear - Coin Chase
require("An Opera of Fear - Speed Bombing", "Bombs")
# An Opera of Fear - Octavio Canal Challenge
# An Opera of Fear - Octavio's Last Stand
require("An Opera of Fear - Venice Treasure Hunt", "Treasure Map")
## Rumble Down Under
# Rumble Down Under - Rock Run
# Rumble Down Under - Cave Sprint
# Rumble Down Under - Cave Mayhem
# Rumble Down Under - Scaling the Drill
require("Rumble Down Under - Guard Swappin'", "Guru")
# Rumble Down Under - Quick Claw
require("Rumble Down Under - Pressure Brawl", "Bentley")
# Rumble Down Under - Croc and Coins
# Rumble Down Under - Carmelita Climb
require("Rumble Down Under - Outback Treasure Hunt", "Treasure Map")
## Flight of Fancy
# Flight of Fancy - Castle Quick Climb
require("Flight of Fancy - Muggshot Goon Attack", "Penelope")
require("Flight of Fancy - Security Breach", "Penelope")
require("Flight of Fancy - Defend the Hangar", "Penelope")
require("Flight of Fancy - Precision Air Duel", ["Murray", "Bentley", "Guru", "Fishing Pole", "Penelope"])
require("Flight of Fancy - Wolf Rampage", "Guru")
require("Flight of Fancy - One Woman Army", "Carmelita")
require("Flight of Fancy - Going Out On A Wing", "Paraglider")
require("Flight of Fancy - Holland Treasure Hunt", "Treasure Map")
## A Cold Alliance
require("A Cold Alliance - Big Air in China", ["Bentley", "Murray", "Guru", "Penelope", "Binocucom"])
require("A Cold Alliance - Sharpshooter", "Panda King")
# A Cold Alliance - Treetop Tangle
# A Cold Alliance - Tsao Showdown
require("A Cold Alliance - China Treasure Hunt", "Treasure Map")
## Dead Men Tell No Tales
# Dead Men Tell No Tales - Patch Grab
# Dead Men Tell No Tales - Stealth Challenge
require("Dead Men Tell No Tales - Boat Bash", "Murray")
require("Dead Men Tell No Tales - Last Ship Sailing", ["Bentley", "Penelope", "Grapple-Cam", "Murray", "Silent Obliteration", "Treasure Map"])
# Dead Men Tell No Tales - Pirate Treasure Hunt
## Honor Among Thieves
# Beauty versus the Beast
# Road Rage
# Dr. M Dogfight
# Ultimate Gauntlet
# Battle Against Time
if world.options.goal.value < 6:
victory_condition = [
"An Opera of Fear - Operation: Tar-Be Gone!",
"Rumble Down Under - Operation: Moon Crash",
"Flight of Fancy - Operation: Turbo Dominant Eagle",
"A Cold Alliance - Operation: Wedding Crasher",
"Dead Men Tell No Tales - Operation: Reverse Double-Cross",
"Honor Among Thieves - Final Legacy"
][world.options.goal.value]
victory_location = world.multiworld.get_location(victory_condition, world.player)
victory_location.address = None
victory_location.place_locked_item(world.create_event("Victory"))
world.multiworld.completion_condition[world.player] = lambda state: state.has("Victory", world.player)
elif world.options.goal.value == 6:
def access_rule(state: CollectionState):
victory_conditions = [
"An Opera of Fear - Operation: Tar-Be Gone!",
"Rumble Down Under - Operation: Moon Crash",
"Flight of Fancy - Operation: Turbo Dominant Eagle",
"A Cold Alliance - Operation: Wedding Crasher",
"Dead Men Tell No Tales - Operation: Reverse Double-Cross",
"Honor Among Thieves - Final Legacy"
]
return all(
world.multiworld.get_location(cond,world.player).access_rule(state)
for cond in victory_conditions
)
world.multiworld.completion_condition[world.player] = access_rule

167
__init__.py Normal file
View File

@@ -0,0 +1,167 @@
from typing import Dict, List, Any, Optional, Mapping
import logging
from BaseClasses import Item, ItemClassification
from worlds.AutoWorld import World, WebWorld
from worlds.LauncherComponents import (
Component,
Type,
components,
launch,
icon_paths,
)
from .Sly3Options import sly3_option_groups, Sly3Options
from .Sly3Regions import create_regions_sly3
from .Sly3Pool import gen_pool_sly3
from .Sly3Rules import set_rules_sly3
from .data.Items import item_dict, item_groups, Sly3Item
from .data.Locations import location_dict, location_groups
from .data.Constants import EPISODES
## Client stuff
def run_client():
from .Sly3Client import launch_client
launch(launch_client, name="Sly3Client")
icon_paths["sly3_ico"] = f"ap:{__name__}/icon.png"
components.append(
Component("Sly 3 Client", func=run_client, component_type=Type.CLIENT, icon="sly3_ico")
)
## UT Stuff
def map_page_index(episode: str) -> int:
mapping = {k: i for i,k in enumerate(EPISODES.keys())}
return mapping.get(episode,0)
## The world
class Sly3Web(WebWorld):
game = "Sly 3: Honor Among Thieves"
option_groups = sly3_option_groups
class Sly3World(World):
"""
Sly 3: Honor Among Thieves is a 2004 stealth action video game developed by
Sucker Punch Productions and published by Sony Computer Entertainment for
the PlayStation 2.
"""
game = "Sly 3: Honor Among Thieves"
web = Sly3Web()
options_dataclass = Sly3Options
options: Sly3Options
topology_present = True
item_name_to_id = {item.name: item.code for item in item_dict.values()}
item_name_groups = item_groups
location_name_to_id = {
location.name: location.code for location in location_dict.values()
}
location_name_groups = location_groups
thiefnet_costs: List[int] = []
# this is how we tell the Universal Tracker we want to use re_gen_passthrough
@staticmethod
def interpret_slot_data(slot_data: Dict[str, Any]) -> Dict[str, Any]:
return slot_data
# and this is how we tell Universal Tracker we don't need the yaml
ut_can_gen_without_yaml = True
def validate_options(self, opt: Sly3Options):
if opt.coins_maximum < opt.coins_minimum:
logging.warning(
f"{self.player_name}: " +
f"Coins minimum cannot be larger than maximum (min: {opt.coins_minimum}, max: {opt.coins_maximum}). Swapping values."
)
temp = opt.coins_minimum.value
opt.coins_minimum.value = opt.coins_maximum.value
opt.coins_maximum.value = temp
if opt.thiefnet_maximum < opt.thiefnet_minimum:
logging.warning(
f"{self.player_name}: " +
f"Thiefnet minimum cannot be larger than maximum (min: {opt.thiefnet_minimum}, max: {opt.thiefnet_maximum}). Swapping values."
)
temp = opt.thiefnet_minimum.value
opt.thiefnet_minimum.value = opt.thiefnet_maximum.value
opt.thiefnet_maximum.value = temp
def generate_early(self) -> None:
# implement .yaml-less Universal Tracker support
if hasattr(self.multiworld, "generation_is_fake"):
if hasattr(self.multiworld, "re_gen_passthrough"):
# I'm doing getattr purely so pylance stops being mad at me
re_gen_passthrough = getattr(self.multiworld, "re_gen_passthrough")
if "Sly 3: Honor Among Thieves" in re_gen_passthrough:
slot_data = re_gen_passthrough["Sly 3: Honor Among Thieves"]
self.thiefnet_costs = slot_data["thiefnet_costs"]
self.options.starting_episode.value = slot_data["starting_episode"]
self.options.goal.value = slot_data["goal"]
self.options.include_mega_jump.value = slot_data["include_mega_jump"]
self.options.coins_minimum.value = slot_data["coins_minimum"]
self.options.coins_maximum.value = slot_data["coins_maximum"]
self.options.thiefnet_minimum.value = slot_data["thiefnet_minimum"]
self.options.thiefnet_maximum.value = slot_data["thiefnet_maximum"]
return
self.validate_options(self.options)
thiefnet_min = self.options.thiefnet_minimum.value
thiefnet_max = self.options.thiefnet_maximum.value
self.thiefnet_costs = sorted([
self.random.randint(thiefnet_min,thiefnet_max)
for _ in range(37)
])
def create_regions(self) -> None:
create_regions_sly3(self)
def get_filler_item_name(self) -> str:
# Currently just coins
return self.random.choice(list(self.item_name_groups["Filler"]))
def create_item(
self, name: str, override: Optional[ItemClassification] = None
) -> Item:
item = item_dict[name]
if override is not None:
return Sly3Item(name, override, item.code, self.player)
return Sly3Item(name, item.classification, item.code, self.player)
def create_event(self, name: str):
return Sly3Item(name, ItemClassification.progression, None, self.player)
def create_items(self) -> None:
items_to_add = gen_pool_sly3(self)
self.multiworld.itempool += items_to_add
def set_rules(self) -> None:
set_rules_sly3(self)
def get_options_as_dict(self) -> Dict[str, Any]:
return self.options.as_dict(
"death_link",
"starting_episode",
"goal",
"include_mega_jump",
"coins_minimum",
"coins_maximum",
"thiefnet_minimum",
"thiefnet_maximum",
)
def fill_slot_data(self) -> Mapping[str, Any]:
slot_data = self.get_options_as_dict()
slot_data["thiefnet_costs"] = self.thiefnet_costs
return slot_data

303
data/Constants.py Normal file
View File

@@ -0,0 +1,303 @@
EPISODES = {
"An Opera of Fear": [
[
"Police HQ"
],
[
"Octavio Snap",
"Into the Depths",
"Canal Chase"
],
[
"Turf War!",
"Tar Ball",
"Run 'n Bomb",
"Guard Duty",
],
[
"Operation: Tar-Be Gone!"
]
],
"Rumble Down Under": [
[
"Search for the Guru"
],
[
"Spelunking",
"Dark Caves",
"Big Truck",
"Unleash the Guru"
],
[
"The Claw",
"Lemon Rage",
"Hungry Croc"
],
[
"Operation: Moon Crash"
]
],
"Flight of Fancy": [
[
"Hidden Flight Roster"
],
[
"Frame Team Belgium",
"Frame Team Iceland",
"Cooper Hangar Defense",
"ACES Semifinals"
],
[
"Giant Wolf Massacre",
"Windmill Firewall",
"Beauty and the Beast"
],
[
"Operation: Turbo Dominant Eagle"
]
],
"A Cold Alliance": [
[
"King of Fire"
],
[
"Get a Job",
"Tearful Reunion",
"Grapple-Cam Break-In",
"Laptop Retrieval"
],
[
"Vampiric Demise",
"Down the Line",
"A Battery of Peril"
],
[
"Operation: Wedding Crasher"
]
],
"Dead Men Tell No Tales": [
[
"The Talk of Pirates"
],
[
"Dynamic Duo",
"Jollyboat of Destruction",
"X Marks the Spot"
],
[
"Crusher from the Depths",
"Deep Sea Danger",
"Battle on the High Seas"
],
[
"Operation: Reverse Double-Cross"
]
],
"Honor Among Thieves": [
[
"Carmelita to the Rescue",
"A Deadly Bite",
"The Dark Current",
"Bump-Charge-Jump",
"Danger in the Skies",
"The Ancestor's Gauntlet",
"Stand Your Ground",
"Final Legacy"
]
]
}
CHALLENGES = {
"An Opera of Fear": [
[],
[
"Canal Chase - Expert Course"
],
[
"Air Time",
"Tower Scramble",
"Coin Chase",
],
[
"Speed Bombing",
"Octavio Canal Challenge",
"Octavio's Last Stand",
"Venice Treasure Hunt"
]
],
"Rumble Down Under": [
[
"Rock Run"
],
[
"Cave Sprint",
"Cave Mayhem",
"Scaling the Drill",
"Guard Swappin'"
],
[
"Quick Claw",
"Pressure Brawl",
"Croc and Coins"
],
[
"Carmelita Climb",
"Outback Treasure Hunt"
]
],
"Flight of Fancy": [
[
"Castle Quick Climb"
],
[
"Muggshot Goon Attack",
"Security Breach",
"Defend the Hangar",
"Precision Air Duel"
],
[
"Wolf Rampage",
"One Woman Army",
],
[
"Going Out On A Wing",
"Holland Treasure Hunt"
]
],
"A Cold Alliance": [
[
"Big Air in China"
],
[
"Sharpshooter",
"Treetop Tangle",
"Tsao Showdown"
],
[],
[
"China Treasure Hunt"
]
],
"Dead Men Tell No Tales": [
[
"Patch Grab",
"Stealth Challenge"
],
[
"Boat Bash",
"Last Ship Sailing"
],
[],
[
"Pirate Treasure Hunt"
]
],
"Honor Among Thieves": [
[
"Beauty versus the Beast",
"Road Rage",
"Dr. M Dogfight",
"Ultimate Gauntlet",
"Battle Against Time"
]
]
}
ADDRESSES = {
"SCUS-97464" : {
"map id": 0x47989C,
"job id": 0x36DB98,
"reload": 0x4797C4,
"reload values": 0x4797CC,
"episode unlocks": 0x56AEC8,
"frame counter": 0x389BE0,
"x pressed": 0x36E78E,
"skip cutscene": 0x389C20,
"jobs": [
[
[0x1335d10]
],
[
[0x1350560,0x1357f80,0x135aba0]
],
[],
[],
[],
[]
],
"text": {
"powerups": [
{},
{},
{
"Trigger Bomb": (0x592c40,0x592e00),
"Fishing Pole": (0x59b000,0x59b2d0),
"Alarm Clock": (0x5962b0,0x5964c0),
"Adrenaline Burst": (0x593410,0x5934f0),
"Health Extractor": (0x5935c0,0x593660),
"Hover Pack": (0x593750,0x593840),
"Insanity Strike": (0x598480,0x598690),
"Grapple-Cam": (0x59acd0,0x59ae50),
"Size Destabilizer": (0x592f30,0x593010),
"Rage Bomb": (0x599250,0x599420),
"Reduction Bomb": (0x5939b0,0x593b00),
"Be The Ball": (0x59a9c0,0x59ab70),
"Berserker Charge": (0x5955a0,0x595700),
"Juggernaut Throw": (0x594c50,0x594dd0),
"Guttural Roar": (0x595830,0x595920),
"Fists of Flame": (0x5944a0,0x594610),
"Temporal Lock": (0x593c40,0x593e60),
"Raging Inferno Flop": (0x595a50,0x595bd0),
"Diablo Fire Slam": (0x595260,0x595450),
"Smoke Bomb": (0x595d60,0x595ee0),
"Combat Dodge": (0x596050,0x596190),
"Paraglide": (0x5966d0,0x5968d0),
"Silent Obliteration": (0x596ba0,0x596df0),
"Feral Pounce": (0x597290,0x5973f0),
"Mega Jump": (0x5975f0,0x597780),
"Knockout Dive": (0x597e10,0x598130),
"Shadow Power Level 1": (0x599c30,0x599eb0),
"Thief Reflexes": (0x596f70,0x597110),
"Shadow Power Level 2": (0x59a140,0x59a310),
"Rocket Boots": (0x57aa00,0x57ace0),
"Treasure Map": (0x57a3e0,0x57a780),
"ENGLISHpowerup_shield_name": (0x59b550,0x579c50),
"Venice Disguise": (0x57ae40,0x57b040),
"Photographer Disguise": (0x57b220,0x57b3b0),
"Pirate Disguise": (0x57b5d0,0x57b7b0),
"Spin Attack Level 1": (0x57b9e0,0x57bc40),
"Spin Attack Level 2": (0x57be80,0x57c130),
"Spin Attack Level 3": (0x57c300,0x57c5b0),
"Jump Attack Level 1": (0x57c7c0,0x57c970),
"Jump Attack Level 2": (0x57cb00,0x57cc30),
"Jump Attack Level 3": (0x57cdf0,0x57d010),
"Push Attack Level 1": (0x57d370,0x57d680),
"Push Attack Level 2": (0x57d940,0x57dc80),
"Push Attack Level 3": (0x57e070,0x57e3b0),
},
{},
{},
{}
]
}
}
}
MENU_RETURN_DATA = (
"794C15EE"+
"419A69B1"+
"FA2319BC"+
"FF2E5E8A"+
"ACD1E787"+
"3A2B7DB0"+
"B94681B3"+
"95777951"+
"CE8FEAA9"+
"07FB6D94"+
"F890094F"+
"3BFA55F6"+
"A0310D22"+
"F93E1EEE"+
"7F2319BC"+
"7B8274B1"
)

107
data/Items.py Normal file
View File

@@ -0,0 +1,107 @@
from typing import NamedTuple
from BaseClasses import Item, ItemClassification
from .Constants import EPISODES
class Sly3Item(Item):
game: str = "Sly 3: Honor Among Thieves"
class Sly3ItemData(NamedTuple):
name: str
code: int
category: str
classification: ItemClassification
filler_list = [
("Coins", ItemClassification.filler, "Filler"),
]
powerup_list = [
("Binocucom", ItemClassification.progression, "Power-Up"),
("Smoke Bomb", ItemClassification.useful, "Power-Up"),
("Knockout Dive", ItemClassification.useful, "Power-Up"),
("Combat Dodge", ItemClassification.useful, "Power-Up"),
("Paraglider", ItemClassification.progression, "Power-Up"),
("Rocket Boots", ItemClassification.useful, "Power-Up"),
("Silent Obliteration", ItemClassification.progression, "Power-Up"),
("Feral Pounce", ItemClassification.useful, "Power-Up"),
("Thief Reflexes", ItemClassification.useful, "Power-Up"),
("Progressive Shadow Power", ItemClassification.useful, "Power-Up"),
("Treasure Map", ItemClassification.progression, "Power-Up"),
("Disguise (Venice)", ItemClassification.progression, "Power-Up"),
("Disguise (Photographer)", ItemClassification.progression, "Power-Up"),
("Disguise (Pirate)", ItemClassification.progression, "Power-Up"),
("Progressive Spin Attack", ItemClassification.useful, "Power-Up"),
("Progressive Jump Attack", ItemClassification.useful, "Power-Up"),
("Progressive Push Attack", ItemClassification.useful, "Power-Up"),
("Mega Jump", ItemClassification.useful, "Power-Up"),
("Bombs", ItemClassification.progression, "Power-Up"),
("Trigger Bomb", ItemClassification.useful, "Power-Up"),
("Fishing Pole", ItemClassification.progression, "Power-Up"),
("Alarm Clock", ItemClassification.useful, "Power-Up"),
("Adrenaline Burst", ItemClassification.useful, "Power-Up"),
("Health Extractor", ItemClassification.useful, "Power-Up"),
("Insanity Strike", ItemClassification.useful, "Power-Up"),
("Grapple-Cam", ItemClassification.progression, "Power-Up"),
("Size Destabilizer", ItemClassification.useful, "Power-Up"),
("Rage Bomb", ItemClassification.useful, "Power-Up"),
("Reduction Bomb", ItemClassification.useful, "Power-Up"),
("Hover Pack", ItemClassification.progression, "Power-Up"),
("Ball Form", ItemClassification.progression, "Power-Up"),
("Berserker Charge", ItemClassification.useful, "Power-Up"),
("Juggernaut Throw", ItemClassification.useful, "Power-Up"),
("Guttural Roar", ItemClassification.useful, "Power-Up"),
("Fists of Flame", ItemClassification.useful, "Power-Up"),
("Temporal Lock", ItemClassification.useful, "Power-Up"),
("Raging Inferno Flop", ItemClassification.useful, "Power-Up"),
("Diablo Fire Slam", ItemClassification.useful, "Power-Up")
]
crew_list = [
("Bentley", ItemClassification.progression, "Crew"),
("Murray", ItemClassification.progression, "Crew"),
("Guru", ItemClassification.progression, "Crew"),
("Penelope", ItemClassification.progression, "Crew"),
("Panda King", ItemClassification.progression, "Crew"),
("Dimitri", ItemClassification.progression, "Crew"),
("Carmelita", ItemClassification.progression, "Crew")
]
progressive_episode_list = [
(f"Progressive {e}", ItemClassification.progression, "Episode")
for e in list(EPISODES.keys())[:-1]
]
item_list = (
filler_list +
powerup_list +
crew_list +
progressive_episode_list
)
base_code = 5318008
item_dict = {
name: Sly3ItemData(name, base_code+code, category, classification)
for code, (name, classification, category) in enumerate(item_list)
}
item_groups = {
key: {item.name for item in item_dict.values() if item.category == key}
for key in [
"Filler",
"Power-Up",
"Episode",
"Crew"
]
}
def from_id(item_id: int) -> Sly3ItemData:
matching = [item for item in item_dict.values() if item.code == item_id]
if len(matching) == 0:
raise ValueError(f"No item data for item id '{item_id}'")
assert len(matching) < 2, f"Multiple item data with id '{item_id}'. Please report."
return matching[0]

52
data/Locations.py Normal file
View File

@@ -0,0 +1,52 @@
from typing import NamedTuple
from .Constants import EPISODES, CHALLENGES
class Sly3LocationData(NamedTuple):
name: str
code: int
category: str
jobs_list = [
(f"{ep} - {job}", "Job")
for ep, chapters in EPISODES.items()
for jobs in chapters
for job in jobs
]
purchases_list = [
(f"ThiefNet {i+1:02}", "Purchase")
for i in range(37)
]
challenges_list = [
(f"{ep} - {challenge}", "Challenge")
for ep, chapters in CHALLENGES.items()
for challenges in chapters
for challenge in challenges
]
location_list = jobs_list + purchases_list + challenges_list
base_code = 8008135
location_dict = {
name: Sly3LocationData(name, base_code+code, category)
for code, (name, category) in enumerate(location_list)
}
location_groups = {
key: {location.name for location in location_dict.values() if location.category == key}
for key in [
"Job",
"Purchase",
"Challenge"
]
}
def from_id(location_id: int) -> Sly3LocationData:
matching = [location for location in location_dict.values() if location.code == location_id]
if len(matching) == 0:
raise ValueError(f"No location data for location id '{location_id}'")
assert len(matching) < 2, f"Multiple locations data with id '{location_id}'. Please report."
return matching[0]

BIN
icon.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 223 KiB

View File

250
pcsx2_interface/pine.py Normal file
View File

@@ -0,0 +1,250 @@
"""
The PINE API.
This is the client side implementation of the PINE protocol.
It allows for a three-way communication between the emulated game, the emulator and an external
tool, using the external tool as a relay for all communication. It is a socket based IPC that
is _very_ fast.
If you want to draw comparisons you can think of this as an equivalent of the BizHawk LUA API,
although with the logic out of the core and in an external tool. While BizHawk would run a lua
script at each frame in the core of the emulator we opt instead to keep the entire logic out of
the emulator to make it more easily extensible, more portable, require less code and be more
performant.
"""
import os
import struct
from enum import IntEnum
from platform import system
import socket
class Pine:
""" Exposes PS2 memory within a running instance of the PCSX2 emulator using the Pine IPC Protocol. """
""" Maximum memory used by an IPC message request. Equivalent to 50,000 Write64 requests. """
MAX_IPC_SIZE: int = 650000
""" Maximum memory used by an IPC message reply. Equivalent to 50,000 Read64 replies. """
MAX_IPC_RETURN_SIZE: int = 450000
""" Maximum number of commands sent in a batch message. """
MAX_BATCH_REPLY_COUNT: int = 50000
class IPCResult(IntEnum):
""" IPC result codes. A list of possible result codes the IPC can send back. Each one of them is what we call an
"opcode" or "tag" and is the first byte sent by the IPC to differentiate between results.
"""
IPC_OK = 0, # IPC command successfully completed.
IPC_FAIL = 0xFF # IPC command failed to complete.
class IPCCommand(IntEnum):
READ8 = 0,
READ16 = 1,
READ32 = 2,
READ64 = 3,
WRITE8 = 4,
WRITE16 = 5,
WRITE32 = 6,
WRITE64 = 7,
VERSION = 8,
SAVE_STATE = 9,
LOAD_STATE = 0xA,
TITLE = 0xB,
ID = 0xC,
UUID = 0xD,
GAME_VERSION = 0xE,
STATUS = 0xF,
UNIMPLEMENTED = 0xFF,
class DataSize(IntEnum):
INT8 = 1,
INT16 = 2,
INT32 = 4,
INT64 = 8,
def __init__(self, slot: int = 28011):
if not 0 < slot <= 65536:
raise ValueError("Provided slot number is outside valid range")
self._slot: int = slot
self._sock: socket.socket = socket.socket()
self._sock_state: bool = False
# self._init_socket()
def _init_socket(self) -> None:
if system() == "Windows":
socket_family = socket.AF_INET
socket_name = ("127.0.0.1", self._slot)
elif system() == "Linux":
socket_family = socket.AF_UNIX
socket_name = os.environ.get("XDG_RUNTIME_DIR", "/tmp")
socket_name += "/pcsx2.sock"
elif system() == "Darwin":
socket_family = socket.AF_UNIX
socket_name = os.environ.get("TMPDIR", "/tmp")
socket_name += "/pcsx2.sock"
else:
socket_family = socket.AF_UNIX
socket_name = "/tmp/pcsx2.sock"
try:
self._sock = socket.socket(socket_family, socket.SOCK_STREAM)
self._sock.settimeout(5.0)
self._sock.connect(socket_name)
except socket.error:
self._sock.close()
self._sock_state = False
return
self._sock_state = True
def connect(self) -> None:
if not self._sock_state:
self._init_socket()
def disconnect(self) -> None:
if self._sock_state:
self._sock.close()
def is_connected(self) -> bool:
return self._sock_state
def read_int8(self, address: int) -> int:
request = Pine._create_request(Pine.IPCCommand.READ8, address, 9)
return Pine.from_bytes(self._send_request(request)[-1:])
def read_int16(self, address) -> int:
request = Pine._create_request(Pine.IPCCommand.READ16, address, 9)
return Pine.from_bytes(self._send_request(request)[-2:])
def read_int32(self, address) -> int:
request = Pine._create_request(Pine.IPCCommand.READ32, address, 9)
return Pine.from_bytes(self._send_request(request)[-4:])
def read_int64(self, address) -> int:
request = Pine._create_request(Pine.IPCCommand.READ64, address, 9)
return Pine.from_bytes(self._send_request(request)[-8:])
def read_bytes(self, address: int, length: int) -> bytes:
"""Careful! This can be quite slow for large reads"""
data = b''
while len(data) < length:
if length - len(data) >= 8:
data += self._send_request(Pine._create_request(Pine.IPCCommand.READ64, address + len(data), 9))[-8:]
elif length - len(data) >= 4:
data += self._send_request(Pine._create_request(Pine.IPCCommand.READ32, address + len(data), 9))[-4:]
elif length - len(data) >= 2:
data += self._send_request(Pine._create_request(Pine.IPCCommand.READ16, address + len(data), 9))[-2:]
elif length - len(data) >= 1:
data += self._send_request(Pine._create_request(Pine.IPCCommand.READ8, address + len(data), 9))[-1:]
return data
def write_int8(self, address: int, value: int) -> None:
request = Pine._create_request(Pine.IPCCommand.WRITE8, address, 9 + Pine.DataSize.INT8)
request += value.to_bytes(length=1, byteorder="little")
self._send_request(request)
def write_int16(self, address: int, value: int) -> None:
request = Pine._create_request(Pine.IPCCommand.WRITE16, address, 9 + Pine.DataSize.INT16)
request += value.to_bytes(length=2, byteorder="little")
self._send_request(request)
def write_int32(self, address: int, value: int) -> None:
request = Pine._create_request(Pine.IPCCommand.WRITE32, address, 9 + Pine.DataSize.INT32)
request += value.to_bytes(length=4, byteorder="little")
self._send_request(request)
def write_int64(self, address: int, value: int) -> None:
request = Pine._create_request(Pine.IPCCommand.WRITE64, address, 9 + Pine.DataSize.INT64)
request += value.to_bytes(length=8, byteorder="little")
self._send_request(request)
def write_float(self, address: int, value: float) -> None:
request = Pine._create_request(Pine.IPCCommand.WRITE32, address, 9 + Pine.DataSize.INT32)
request + struct.pack("<f", value)
self._send_request(request)
def write_bytes(self, address: int, data: bytes) -> None:
"""Careful! This can be quite slow for large writes"""
bytes_written = 0
while bytes_written < len(data):
if len(data) - bytes_written >= 8:
request = self._create_request(Pine.IPCCommand.WRITE64, address + bytes_written, 9 + Pine.DataSize.INT64)
request += data[bytes_written:bytes_written + 8]
self._send_request(request)
bytes_written += 8
elif len(data) - bytes_written >= 4:
request = self._create_request(Pine.IPCCommand.WRITE32, address + bytes_written, 9 + Pine.DataSize.INT32)
request += data[bytes_written:bytes_written + 4]
self._send_request(request)
bytes_written += 4
elif len(data) - bytes_written >= 2:
request = self._create_request(Pine.IPCCommand.WRITE16, address + bytes_written, 9 + Pine.DataSize.INT16)
request += data[bytes_written:bytes_written + 2]
self._send_request(request)
bytes_written += 2
elif len(data) - bytes_written >= 1:
request = self._create_request(Pine.IPCCommand.WRITE8, address + bytes_written, 9 + Pine.DataSize.INT8)
request += data[bytes_written:bytes_written + 1]
self._send_request(request)
bytes_written += 1
def get_game_id(self) -> str:
request = Pine.to_bytes(5, 4) + Pine.to_bytes(Pine.IPCCommand.ID, 1)
response = self._send_request(request)
return response[9:-1].decode("ascii")
def _send_request(self, request: bytes) -> bytes:
if not self._sock_state:
self._init_socket()
try:
self._sock.sendall(request)
except socket.error:
self._sock.close()
self._sock_state = False
raise ConnectionError("Lost connection to PCSX2.")
end_length = 4
result: bytes = b''
while len(result) < end_length:
try:
response = self._sock.recv(4096)
except TimeoutError:
raise TimeoutError("Response timed out. "
"This might be caused by having two PINE connections open on the same slot")
if len(response) <= 0:
result = b''
break
result += response
if end_length == 4 and len(response) >= 4:
end_length = Pine.from_bytes(result[0:4])
if end_length > Pine.MAX_IPC_SIZE:
result = b''
break
if len(result) == 0:
raise ConnectionError("Invalid response from PCSX2.")
if result[4] == Pine.IPCResult.IPC_FAIL:
raise ConnectionError("Failure indicated in PCSX2 response.")
return result
@staticmethod
def _create_request(command: IPCCommand, address: int, size: int = 0) -> bytes:
ipc = Pine.to_bytes(size, 4)
ipc += Pine.to_bytes(command, 1)
ipc += Pine.to_bytes(address, 4)
return ipc
@staticmethod
def to_bytes(value: int, size: int) -> bytes:
return value.to_bytes(length=size, byteorder="little")
@staticmethod
def from_bytes(arr: bytes) -> int:
return int.from_bytes(arr, byteorder="little")