mirror of
https://github.com/MarioSpore/Grinch-AP.git
synced 2025-10-21 20:21:32 -06:00
Core remove legacy patch (#1047)
Co-authored-by: black-sliver <59490463+black-sliver@users.noreply.github.com>
This commit is contained in:
156
worlds/Files.py
Normal file
156
worlds/Files.py
Normal file
@@ -0,0 +1,156 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import zipfile
|
||||
|
||||
from typing import ClassVar, Dict, Tuple, Any, Optional, Union, BinaryIO
|
||||
|
||||
import bsdiff4
|
||||
|
||||
|
||||
class AutoPatchRegister(type):
|
||||
patch_types: ClassVar[Dict[str, AutoPatchRegister]] = {}
|
||||
file_endings: ClassVar[Dict[str, AutoPatchRegister]] = {}
|
||||
|
||||
def __new__(mcs, name: str, bases: Tuple[type, ...], dct: Dict[str, Any]) -> AutoPatchRegister:
|
||||
# construct class
|
||||
new_class = super().__new__(mcs, name, bases, dct)
|
||||
if "game" in dct:
|
||||
AutoPatchRegister.patch_types[dct["game"]] = new_class
|
||||
if not dct["patch_file_ending"]:
|
||||
raise Exception(f"Need an expected file ending for {name}")
|
||||
AutoPatchRegister.file_endings[dct["patch_file_ending"]] = new_class
|
||||
return new_class
|
||||
|
||||
@staticmethod
|
||||
def get_handler(file: str) -> Optional[AutoPatchRegister]:
|
||||
for file_ending, handler in AutoPatchRegister.file_endings.items():
|
||||
if file.endswith(file_ending):
|
||||
return handler
|
||||
return None
|
||||
|
||||
|
||||
current_patch_version: int = 5
|
||||
|
||||
|
||||
class APContainer:
|
||||
"""A zipfile containing at least archipelago.json"""
|
||||
version: int = current_patch_version
|
||||
compression_level: int = 9
|
||||
compression_method: int = zipfile.ZIP_DEFLATED
|
||||
game: Optional[str] = None
|
||||
|
||||
# instance attributes:
|
||||
path: Optional[str]
|
||||
player: Optional[int]
|
||||
player_name: str
|
||||
server: str
|
||||
|
||||
def __init__(self, path: Optional[str] = None, player: Optional[int] = None,
|
||||
player_name: str = "", server: str = ""):
|
||||
self.path = path
|
||||
self.player = player
|
||||
self.player_name = player_name
|
||||
self.server = server
|
||||
|
||||
def write(self, file: Optional[Union[str, BinaryIO]] = None) -> None:
|
||||
zip_file = file if file else self.path
|
||||
if not zip_file:
|
||||
raise FileNotFoundError(f"Cannot write {self.__class__.__name__} due to no path provided.")
|
||||
with zipfile.ZipFile(zip_file, "w", self.compression_method, True, self.compression_level) \
|
||||
as zf:
|
||||
if file:
|
||||
self.path = zf.filename
|
||||
self.write_contents(zf)
|
||||
|
||||
def write_contents(self, opened_zipfile: zipfile.ZipFile) -> None:
|
||||
manifest = self.get_manifest()
|
||||
try:
|
||||
manifest_str = json.dumps(manifest)
|
||||
except Exception as e:
|
||||
raise Exception(f"Manifest {manifest} did not convert to json.") from e
|
||||
else:
|
||||
opened_zipfile.writestr("archipelago.json", manifest_str)
|
||||
|
||||
def read(self, file: Optional[Union[str, BinaryIO]] = None) -> None:
|
||||
"""Read data into patch object. file can be file-like, such as an outer zip file's stream."""
|
||||
zip_file = file if file else self.path
|
||||
if not zip_file:
|
||||
raise FileNotFoundError(f"Cannot read {self.__class__.__name__} due to no path provided.")
|
||||
with zipfile.ZipFile(zip_file, "r") as zf:
|
||||
if file:
|
||||
self.path = zf.filename
|
||||
self.read_contents(zf)
|
||||
|
||||
def read_contents(self, opened_zipfile: zipfile.ZipFile) -> None:
|
||||
with opened_zipfile.open("archipelago.json", "r") as f:
|
||||
manifest = json.load(f)
|
||||
if manifest["compatible_version"] > self.version:
|
||||
raise Exception(f"File (version: {manifest['compatible_version']}) too new "
|
||||
f"for this handler (version: {self.version})")
|
||||
self.player = manifest["player"]
|
||||
self.server = manifest["server"]
|
||||
self.player_name = manifest["player_name"]
|
||||
|
||||
def get_manifest(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"server": self.server, # allow immediate connection to server in multiworld. Empty string otherwise
|
||||
"player": self.player,
|
||||
"player_name": self.player_name,
|
||||
"game": self.game,
|
||||
# minimum version of patch system expected for patching to be successful
|
||||
"compatible_version": 4,
|
||||
"version": current_patch_version,
|
||||
}
|
||||
|
||||
|
||||
class APDeltaPatch(APContainer, metaclass=AutoPatchRegister):
|
||||
"""An APContainer that additionally has delta.bsdiff4
|
||||
containing a delta patch to get the desired file, often a rom."""
|
||||
|
||||
hash: Optional[str] # base checksum of source file
|
||||
patch_file_ending: str = ""
|
||||
delta: Optional[bytes] = None
|
||||
result_file_ending: str = ".sfc"
|
||||
source_data: bytes
|
||||
|
||||
def __init__(self, *args: Any, patched_path: str = "", **kwargs: Any) -> None:
|
||||
self.patched_path = patched_path
|
||||
super(APDeltaPatch, self).__init__(*args, **kwargs)
|
||||
|
||||
def get_manifest(self) -> Dict[str, Any]:
|
||||
manifest = super(APDeltaPatch, self).get_manifest()
|
||||
manifest["base_checksum"] = self.hash
|
||||
manifest["result_file_ending"] = self.result_file_ending
|
||||
manifest["patch_file_ending"] = self.patch_file_ending
|
||||
return manifest
|
||||
|
||||
@classmethod
|
||||
def get_source_data(cls) -> bytes:
|
||||
"""Get Base data"""
|
||||
raise NotImplementedError()
|
||||
|
||||
@classmethod
|
||||
def get_source_data_with_cache(cls) -> bytes:
|
||||
if not hasattr(cls, "source_data"):
|
||||
cls.source_data = cls.get_source_data()
|
||||
return cls.source_data
|
||||
|
||||
def write_contents(self, opened_zipfile: zipfile.ZipFile):
|
||||
super(APDeltaPatch, self).write_contents(opened_zipfile)
|
||||
# write Delta
|
||||
opened_zipfile.writestr("delta.bsdiff4",
|
||||
bsdiff4.diff(self.get_source_data_with_cache(), open(self.patched_path, "rb").read()),
|
||||
compress_type=zipfile.ZIP_STORED) # bsdiff4 is a format with integrated compression
|
||||
|
||||
def read_contents(self, opened_zipfile: zipfile.ZipFile):
|
||||
super(APDeltaPatch, self).read_contents(opened_zipfile)
|
||||
self.delta = opened_zipfile.read("delta.bsdiff4")
|
||||
|
||||
def patch(self, target: str):
|
||||
"""Base + Delta -> Patched"""
|
||||
if not self.delta:
|
||||
self.read()
|
||||
result = bsdiff4.patch(self.get_source_data_with_cache(), self.delta)
|
||||
with open(target, "wb") as f:
|
||||
f.write(result)
|
||||
@@ -1,11 +1,12 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import Utils
|
||||
from Patch import read_rom
|
||||
import worlds.AutoWorld
|
||||
import worlds.Files
|
||||
|
||||
LTTPJPN10HASH = '03a63945398191337e896e5771f77173'
|
||||
RANDOMIZERBASEHASH = '9952c2a3ec1b421e408df0d20c8f0c7f'
|
||||
ROM_PLAYER_LIMIT = 255
|
||||
LTTPJPN10HASH: str = "03a63945398191337e896e5771f77173"
|
||||
RANDOMIZERBASEHASH: str = "9952c2a3ec1b421e408df0d20c8f0c7f"
|
||||
ROM_PLAYER_LIMIT: int = 255
|
||||
|
||||
import io
|
||||
import json
|
||||
@@ -34,7 +35,7 @@ from worlds.alttp.Text import KingsReturn_texts, Sanctuary_texts, Kakariko_texts
|
||||
DeathMountain_texts, \
|
||||
LostWoods_texts, WishingWell_texts, DesertPalace_texts, MountainTower_texts, LinksHouse_texts, Lumberjacks_texts, \
|
||||
SickKid_texts, FluteBoy_texts, Zora_texts, MagicShop_texts, Sahasrahla_names
|
||||
from Utils import local_path, user_path, int16_as_bytes, int32_as_bytes, snes_to_pc, is_frozen, parse_yaml
|
||||
from Utils import local_path, user_path, int16_as_bytes, int32_as_bytes, snes_to_pc, is_frozen, parse_yaml, read_snes_rom
|
||||
from worlds.alttp.Items import ItemFactory, item_table, item_name_groups, progression_items
|
||||
from worlds.alttp.EntranceShuffle import door_addresses
|
||||
from worlds.alttp.Options import smallkey_shuffle
|
||||
@@ -57,13 +58,13 @@ class LocalRom(object):
|
||||
self.orig_buffer = None
|
||||
|
||||
with open(file, 'rb') as stream:
|
||||
self.buffer = read_rom(stream)
|
||||
self.buffer = read_snes_rom(stream)
|
||||
if patch:
|
||||
self.patch_base_rom()
|
||||
self.orig_buffer = self.buffer.copy()
|
||||
if vanillaRom:
|
||||
with open(vanillaRom, 'rb') as vanillaStream:
|
||||
self.orig_buffer = read_rom(vanillaStream)
|
||||
self.orig_buffer = read_snes_rom(vanillaStream)
|
||||
|
||||
def read_byte(self, address: int) -> int:
|
||||
return self.buffer[address]
|
||||
@@ -123,29 +124,24 @@ class LocalRom(object):
|
||||
return expected == buffermd5.hexdigest()
|
||||
|
||||
def patch_base_rom(self):
|
||||
if os.path.isfile(local_path('basepatch.sfc')):
|
||||
with open(local_path('basepatch.sfc'), 'rb') as stream:
|
||||
if os.path.isfile(user_path('basepatch.sfc')):
|
||||
with open(user_path('basepatch.sfc'), 'rb') as stream:
|
||||
buffer = bytearray(stream.read())
|
||||
|
||||
if self.verify(buffer):
|
||||
self.buffer = buffer
|
||||
if not os.path.exists(local_path('data', 'basepatch.apbp')):
|
||||
Patch.create_patch_file(local_path('basepatch.sfc'))
|
||||
return
|
||||
|
||||
if not os.path.isfile(local_path('data', 'basepatch.apbp')):
|
||||
raise RuntimeError('Base patch unverified. Unable to continue.')
|
||||
with open(local_path("data", "basepatch.bsdiff4"), "rb") as f:
|
||||
delta = f.read()
|
||||
|
||||
if os.path.isfile(local_path('data', 'basepatch.apbp')):
|
||||
_, target, buffer = Patch.create_rom_bytes(local_path('data', 'basepatch.apbp'), ignore_version=True)
|
||||
if self.verify(buffer):
|
||||
self.buffer = bytearray(buffer)
|
||||
with open(user_path('basepatch.sfc'), 'wb') as stream:
|
||||
stream.write(buffer)
|
||||
return
|
||||
raise RuntimeError('Base patch unverified. Unable to continue.')
|
||||
|
||||
raise RuntimeError('Could not find Base Patch. Unable to continue.')
|
||||
buffer = bsdiff4.patch(get_base_rom_bytes(), delta)
|
||||
if self.verify(buffer):
|
||||
self.buffer = bytearray(buffer)
|
||||
with open(user_path('basepatch.sfc'), 'wb') as stream:
|
||||
stream.write(buffer)
|
||||
return
|
||||
raise RuntimeError('Base patch unverified. Unable to continue.')
|
||||
|
||||
def write_crc(self):
|
||||
crc = (sum(self.buffer[:0x7FDC] + self.buffer[0x7FE0:]) + 0x01FE) & 0xFFFF
|
||||
@@ -544,7 +540,7 @@ class Sprite():
|
||||
|
||||
def get_vanilla_sprite_data(self):
|
||||
file_name = get_base_rom_path()
|
||||
base_rom_bytes = bytes(read_rom(open(file_name, "rb")))
|
||||
base_rom_bytes = bytes(read_snes_rom(open(file_name, "rb")))
|
||||
Sprite.sprite = base_rom_bytes[0x80000:0x87000]
|
||||
Sprite.palette = base_rom_bytes[0xDD308:0xDD380]
|
||||
Sprite.glove_palette = base_rom_bytes[0xDEDF5:0xDEDF9]
|
||||
@@ -2906,7 +2902,7 @@ hash_alphabet = [
|
||||
]
|
||||
|
||||
|
||||
class LttPDeltaPatch(Patch.APDeltaPatch):
|
||||
class LttPDeltaPatch(worlds.Files.APDeltaPatch):
|
||||
hash = LTTPJPN10HASH
|
||||
game = "A Link to the Past"
|
||||
patch_file_ending = ".aplttp"
|
||||
@@ -2920,7 +2916,7 @@ def get_base_rom_bytes(file_name: str = "") -> bytes:
|
||||
base_rom_bytes = getattr(get_base_rom_bytes, "base_rom_bytes", None)
|
||||
if not base_rom_bytes:
|
||||
file_name = get_base_rom_path(file_name)
|
||||
base_rom_bytes = bytes(read_rom(open(file_name, "rb")))
|
||||
base_rom_bytes = bytes(read_snes_rom(open(file_name, "rb")))
|
||||
|
||||
basemd5 = hashlib.md5()
|
||||
basemd5.update(base_rom_bytes)
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import Utils
|
||||
from Patch import read_rom, APDeltaPatch
|
||||
from Utils import read_snes_rom
|
||||
from worlds.Files import APDeltaPatch
|
||||
from .Locations import lookup_id_to_name, all_locations
|
||||
from .Levels import level_list, level_dict
|
||||
|
||||
@@ -440,13 +441,13 @@ class LocalRom(object):
|
||||
self.orig_buffer = None
|
||||
|
||||
with open(file, 'rb') as stream:
|
||||
self.buffer = read_rom(stream)
|
||||
self.buffer = read_snes_rom(stream)
|
||||
#if patch:
|
||||
# self.patch_rom()
|
||||
# self.orig_buffer = self.buffer.copy()
|
||||
#if vanillaRom:
|
||||
# with open(vanillaRom, 'rb') as vanillaStream:
|
||||
# self.orig_buffer = read_rom(vanillaStream)
|
||||
# self.orig_buffer = read_snes_rom(vanillaStream)
|
||||
|
||||
def read_bit(self, address: int, bit_number: int) -> bool:
|
||||
bitflag = (1 << bit_number)
|
||||
@@ -724,7 +725,7 @@ def get_base_rom_bytes(file_name: str = "") -> bytes:
|
||||
base_rom_bytes = getattr(get_base_rom_bytes, "base_rom_bytes", None)
|
||||
if not base_rom_bytes:
|
||||
file_name = get_base_rom_path(file_name)
|
||||
base_rom_bytes = bytes(read_rom(open(file_name, "rb")))
|
||||
base_rom_bytes = bytes(read_snes_rom(open(file_name, "rb")))
|
||||
|
||||
basemd5 = hashlib.md5()
|
||||
basemd5.update(base_rom_bytes)
|
||||
|
||||
@@ -11,6 +11,8 @@ import shutil
|
||||
|
||||
import Utils
|
||||
import Patch
|
||||
import worlds.AutoWorld
|
||||
import worlds.Files
|
||||
from . import Options
|
||||
|
||||
from .Technologies import tech_table, recipes, free_sample_exclusions, progressive_technology_table, \
|
||||
@@ -57,7 +59,7 @@ recipe_time_ranges = {
|
||||
}
|
||||
|
||||
|
||||
class FactorioModFile(Patch.APContainer):
|
||||
class FactorioModFile(worlds.Files.APContainer):
|
||||
game = "Factorio"
|
||||
compression_method = zipfile.ZIP_DEFLATED # Factorio can't load LZMA archives
|
||||
|
||||
|
||||
@@ -3,7 +3,8 @@ import os
|
||||
|
||||
import json
|
||||
import Utils
|
||||
from Patch import read_rom, APDeltaPatch
|
||||
from Utils import read_snes_rom
|
||||
from worlds.Files import APDeltaPatch
|
||||
|
||||
SMJUHASH = '21f3e98df4780ee1c667b84e57d88675'
|
||||
ROM_PLAYER_LIMIT = 65535 # max archipelago player ID. note, SM ROM itself will only store 201 names+ids max
|
||||
@@ -22,7 +23,7 @@ def get_base_rom_bytes(file_name: str = "") -> bytes:
|
||||
base_rom_bytes = getattr(get_base_rom_bytes, "base_rom_bytes", None)
|
||||
if not base_rom_bytes:
|
||||
file_name = get_base_rom_path(file_name)
|
||||
base_rom_bytes = bytes(read_rom(open(file_name, "rb")))
|
||||
base_rom_bytes = bytes(read_snes_rom(open(file_name, "rb")))
|
||||
|
||||
basemd5 = hashlib.md5()
|
||||
basemd5.update(base_rom_bytes)
|
||||
|
||||
@@ -1,8 +1,7 @@
|
||||
import Utils
|
||||
from Patch import read_rom, APDeltaPatch
|
||||
from worlds.Files import APDeltaPatch
|
||||
from .Aesthetics import generate_shuffled_header_data
|
||||
from .Locations import lookup_id_to_name, all_locations
|
||||
from .Levels import level_info_dict, full_level_list, submap_level_list, location_id_to_level_id
|
||||
from .Levels import level_info_dict
|
||||
from .Names.TextBox import generate_goal_text, title_text_mapping, generate_text_box
|
||||
|
||||
USHASH = 'cdd3c8c37322978ca8669b34bc89c804'
|
||||
@@ -69,7 +68,7 @@ class LocalRom:
|
||||
self.orig_buffer = None
|
||||
|
||||
with open(file, 'rb') as stream:
|
||||
self.buffer = read_rom(stream)
|
||||
self.buffer = Utils.read_snes_rom(stream)
|
||||
|
||||
def read_bit(self, address: int, bit_number: int) -> bool:
|
||||
bitflag = (1 << bit_number)
|
||||
@@ -827,7 +826,7 @@ def get_base_rom_bytes(file_name: str = "") -> bytes:
|
||||
base_rom_bytes = getattr(get_base_rom_bytes, "base_rom_bytes", None)
|
||||
if not base_rom_bytes:
|
||||
file_name = get_base_rom_path(file_name)
|
||||
base_rom_bytes = bytes(read_rom(open(file_name, "rb")))
|
||||
base_rom_bytes = bytes(Utils.read_snes_rom(open(file_name, "rb")))
|
||||
|
||||
basemd5 = hashlib.md5()
|
||||
basemd5.update(base_rom_bytes)
|
||||
@@ -837,6 +836,7 @@ def get_base_rom_bytes(file_name: str = "") -> bytes:
|
||||
get_base_rom_bytes.base_rom_bytes = base_rom_bytes
|
||||
return base_rom_bytes
|
||||
|
||||
|
||||
def get_base_rom_path(file_name: str = "") -> str:
|
||||
options = Utils.get_options()
|
||||
if not file_name:
|
||||
|
||||
@@ -14,7 +14,6 @@ from ..generic.Rules import add_rule
|
||||
from .Names import ItemName, LocationName
|
||||
from ..AutoWorld import WebWorld, World
|
||||
from .Rom import LocalRom, patch_rom, get_base_rom_path, SMWDeltaPatch
|
||||
import Patch
|
||||
|
||||
|
||||
class SMWWeb(WebWorld):
|
||||
@@ -146,6 +145,7 @@ class SMWWorld(World):
|
||||
|
||||
|
||||
def generate_output(self, output_directory: str):
|
||||
rompath = "" # if variable is not declared finally clause may fail
|
||||
try:
|
||||
world = self.world
|
||||
player = self.player
|
||||
@@ -167,9 +167,9 @@ class SMWWorld(World):
|
||||
except:
|
||||
raise
|
||||
finally:
|
||||
self.rom_name_available_event.set() # make sure threading continues and errors are collected
|
||||
if os.path.exists(rompath):
|
||||
os.unlink(rompath)
|
||||
self.rom_name_available_event.set() # make sure threading continues and errors are collected
|
||||
|
||||
def modify_multidata(self, multidata: dict):
|
||||
import base64
|
||||
|
||||
@@ -2,7 +2,8 @@ import hashlib
|
||||
import os
|
||||
|
||||
import Utils
|
||||
from Patch import read_rom, APDeltaPatch
|
||||
from Utils import read_snes_rom
|
||||
from worlds.Files import APDeltaPatch
|
||||
|
||||
SMJUHASH = '21f3e98df4780ee1c667b84e57d88675'
|
||||
LTTPJPN10HASH = '03a63945398191337e896e5771f77173'
|
||||
@@ -23,7 +24,7 @@ def get_base_rom_bytes() -> bytes:
|
||||
base_rom_bytes = getattr(get_base_rom_bytes, "base_rom_bytes", None)
|
||||
if not base_rom_bytes:
|
||||
sm_file_name = get_sm_base_rom_path()
|
||||
sm_base_rom_bytes = bytes(read_rom(open(sm_file_name, "rb")))
|
||||
sm_base_rom_bytes = bytes(read_snes_rom(open(sm_file_name, "rb")))
|
||||
|
||||
basemd5 = hashlib.md5()
|
||||
basemd5.update(sm_base_rom_bytes)
|
||||
@@ -31,7 +32,7 @@ def get_base_rom_bytes() -> bytes:
|
||||
raise Exception('Supplied Base Rom does not match known MD5 for SM Japan+US release. '
|
||||
'Get the correct game and version, then dump it')
|
||||
lttp_file_name = get_lttp_base_rom_path()
|
||||
lttp_base_rom_bytes = bytes(read_rom(open(lttp_file_name, "rb")))
|
||||
lttp_base_rom_bytes = bytes(read_snes_rom(open(lttp_file_name, "rb")))
|
||||
|
||||
basemd5 = hashlib.md5()
|
||||
basemd5.update(lttp_base_rom_bytes)
|
||||
|
||||
@@ -2,7 +2,7 @@ import bsdiff4
|
||||
import yaml
|
||||
from typing import Optional
|
||||
import Utils
|
||||
from Patch import APDeltaPatch
|
||||
from worlds.Files import APDeltaPatch
|
||||
import os
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user