alttpr-gwaa-kiwi/alttpr-python
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Merge remote-tracking branch 'remotes/multi/multiworld_31' into multidoors

Browse Source
This commit is contained in:
compiling
2020-01-19 18:18:43 +11:00
parent c08fb4bd5e 7f800de8d0
commit e0eff76b9b
12 changed files with 583 additions and 512 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
AdjusterMain.py
View File

@@ -2,7 +2,7 @@ import os
import time
import logging
from Utils import output_path, parse_names_string
from Utils import output_path
from Rom import LocalRom, apply_rom_settings
@@ -21,7 +21,7 @@ def adjust(args):
else:
raise RuntimeError('Provided Rom is not a valid Link to the Past Randomizer Rom. Please provide one for adjusting.')
apply_rom_settings(rom, args.heartbeep, args.heartcolor, args.quickswap, args.fastmenu, args.disablemusic, args.sprite, args.ow_palettes, args.uw_palettes, parse_names_string(args.names))
apply_rom_settings(rom, args.heartbeep, args.heartcolor, args.quickswap, args.fastmenu, args.disablemusic, args.sprite, args.ow_palettes, args.uw_palettes)
rom.write_to_file(output_path('%s.sfc' % outfilebase))

119
BaseClasses.py
View File

@@ -14,6 +14,7 @@ class World(object):
def __init__(self, players, shuffle, doorShuffle, logic, mode, swords, difficulty, difficulty_adjustments, timer, progressive, goal, algorithm, accessibility, shuffle_ganon, retro, custom, customitemarray, hints):
self.players = players
self.teams = 1
self.shuffle = shuffle.copy()
self.doorShuffle = doorShuffle.copy()
self.logic = logic.copy()
@@ -73,6 +74,8 @@ class World(object):
def set_player_attr(attr, val):
self.__dict__.setdefault(attr, {})[player] = val
set_player_attr('_region_cache', {})
set_player_attr('player_names', [])
set_player_attr('remote_items', False)
set_player_attr('required_medallions', ['Ether', 'Quake'])
set_player_attr('swamp_patch_required', False)
set_player_attr('powder_patch_required', False)
@@ -105,6 +108,12 @@ class World(object):
set_player_attr('treasure_hunt_icon', 'Triforce Piece')
set_player_attr('treasure_hunt_count', 0)
def get_name_string_for_object(self, obj):
return obj.name if self.players == 1 else f'{obj.name} ({self.get_player_names(obj.player)})'
def get_player_names(self, player):
return ", ".join([name for i, name in enumerate(self.player_names[player]) if self.player_names[player].index(name) == i])
def initialize_regions(self, regions=None):
for region in regions if regions else self.regions:
region.world = self
@@ -279,6 +288,7 @@ class World(object):
return [location for location in self.get_locations() if location.item is not None and location.item.name == item and location.item.player == player]
def push_precollected(self, item):
item.world = self
if (item.smallkey and self.keyshuffle[item.player]) or (item.bigkey and self.bigkeyshuffle[item.player]):
item.advancement = True
self.precollected_items.append(item)
@@ -291,6 +301,7 @@ class World(object):
if location.can_fill(self.state, item, False):
location.item = item
item.location = location
item.world = self
if collect:
self.state.collect(item, location.event, location)
@@ -876,10 +887,7 @@ class Region(object):
return str(self.__unicode__())
def __unicode__(self):
if self.world and self.world.players == 1:
return self.name
else:
return '%s (Player %d)' % (self.name, self.player)
return self.world.get_name_string_for_object(self) if self.world else f'{self.name} (Player {self.player})'
class Entrance(object):
@@ -916,11 +924,8 @@ class Entrance(object):
return str(self.__unicode__())
def __unicode__(self):
if self.parent_region and self.parent_region.world and self.parent_region.world.players == 1:
return self.name
else:
return '%s (Player %d)' % (self.name, self.player)
world = self.parent_region.world if self.parent_region else None
return world.get_name_string_for_object(self) if world else f'{self.name} (Player {self.player})'
class Dungeon(object):
@@ -969,10 +974,7 @@ class Dungeon(object):
return str(self.__unicode__())
def __unicode__(self):
if self.world and self.world.players==1:
return self.name
else:
return '%s (Player %d)' % (self.name, self.player)
return self.world.get_name_string_for_object(self) if self.world else f'{self.name} (Player {self.player})'
@unique
@@ -1314,10 +1316,8 @@ class Location(object):
return str(self.__unicode__())
def __unicode__(self):
if self.parent_region and self.parent_region.world and self.parent_region.world.players == 1:
return self.name
else:
return '%s (Player %d)' % (self.name, self.player)
world = self.parent_region.world if self.parent_region and self.parent_region.world else None
return world.get_name_string_for_object(self) if world else f'{self.name} (Player {self.player})'
class Item(object):
@@ -1336,6 +1336,7 @@ class Item(object):
self.hint_text = hint_text
self.code = code
self.location = None
self.world = None
self.player = player
@property
@@ -1362,10 +1363,7 @@ class Item(object):
return str(self.__unicode__())
def __unicode__(self):
if self.location and self.location.parent_region and self.location.parent_region.world and self.location.parent_region.world.players == 1:
return self.name
else:
return '%s (Player %d)' % (self.name, self.player)
return self.world.get_name_string_for_object(self) if self.world else f'{self.name} (Player {self.player})'
# have 6 address that need to be filled
@@ -1438,6 +1436,7 @@ class Spoiler(object):
def __init__(self, world):
self.world = world
self.hashes = {}
self.entrances = OrderedDict()
self.doors = OrderedDict()
self.doorTypes = OrderedDict()
@@ -1476,8 +1475,8 @@ class Spoiler(object):
self.medallions['Turtle Rock'] = self.world.required_medallions[1][1]
else:
for player in range(1, self.world.players + 1):
self.medallions['Misery Mire (Player %d)' % player] = self.world.required_medallions[player][0]
self.medallions['Turtle Rock (Player %d)' % player] = self.world.required_medallions[player][1]
self.medallions[f'Misery Mire ({self.world.get_player_names(player)})'] = self.world.required_medallions[player][0]
self.medallions[f'Turtle Rock ({self.world.get_player_names(player)})'] = self.world.required_medallions[player][1]
self.startinventory = list(map(str, self.world.precollected_items))
@@ -1566,7 +1565,8 @@ class Spoiler(object):
'enemy_shuffle': self.world.enemy_shuffle,
'enemy_health': self.world.enemy_health,
'enemy_damage': self.world.enemy_damage,
'players': self.world.players
'players': self.world.players,
'teams': self.world.teams
}
def to_json(self):
@@ -1578,6 +1578,8 @@ class Spoiler(object):
out.update(self.locations)
out['Starting Inventory'] = self.startinventory
out['Special'] = self.medallions
if self.hashes:
out['Hashes'] = {f"{self.world.player_names[player][team]} (Team {team+1})": hash for (player, team), hash in self.hashes.items()}
if self.shops:
out['Shops'] = self.shops
out['playthrough'] = self.playthrough
@@ -1591,30 +1593,35 @@ class Spoiler(object):
self.parse_data()
with open(filename, 'w') as outfile:
outfile.write('ALttP Entrance Randomizer Version %s - Seed: %s\n\n' % (self.metadata['version'], self.world.seed))
outfile.write('Players: %d\n' % self.world.players)
outfile.write('Filling Algorithm: %s\n' % self.world.algorithm)
outfile.write('Logic: %s\n' % self.metadata['logic'])
outfile.write('Mode: %s\n' % self.metadata['mode'])
outfile.write('Retro: %s\n' % {k: 'Yes' if v else 'No' for k, v in self.metadata['retro'].items()})
outfile.write('Swords: %s\n' % self.metadata['weapons'])
outfile.write('Goal: %s\n' % self.metadata['goal'])
outfile.write('Difficulty: %s\n' % self.metadata['item_pool'])
outfile.write('Item Functionality: %s\n' % self.metadata['item_functionality'])
outfile.write('Entrance Shuffle: %s\n' % self.metadata['shuffle'])
outfile.write('Door Shuffle: %s\n' % self.metadata['door_shuffle'])
outfile.write('Crystals required for GT: %s\n' % self.metadata['gt_crystals'])
outfile.write('Crystals required for Ganon: %s\n' % self.metadata['ganon_crystals'])
outfile.write('Pyramid hole pre-opened: %s\n' % {k: 'Yes' if v else 'No' for k, v in self.metadata['open_pyramid'].items()})
outfile.write('Accessibility: %s\n' % self.metadata['accessibility'])
outfile.write('Map shuffle: %s\n' % {k: 'Yes' if v else 'No' for k, v in self.metadata['mapshuffle'].items()})
outfile.write('Compass shuffle: %s\n' % {k: 'Yes' if v else 'No' for k, v in self.metadata['compassshuffle'].items()})
outfile.write('Small Key shuffle: %s\n' % {k: 'Yes' if v else 'No' for k, v in self.metadata['keyshuffle'].items()})
outfile.write('Big Key shuffle: %s\n' % {k: 'Yes' if v else 'No' for k, v in self.metadata['bigkeyshuffle'].items()})
outfile.write('Boss shuffle: %s\n' % self.metadata['boss_shuffle'])
outfile.write('Enemy shuffle: %s\n' % self.metadata['enemy_shuffle'])
outfile.write('Enemy health: %s\n' % self.metadata['enemy_health'])
outfile.write('Enemy damage: %s\n' % self.metadata['enemy_damage'])
outfile.write('Hints: %s\n' % {k: 'Yes' if v else 'No' for k, v in self.metadata['hints'].items()})
outfile.write('Players: %d\n' % self.world.players)
outfile.write('Teams: %d\n' % self.world.teams)
for player in range(1, self.world.players + 1):
if self.world.players > 1:
outfile.write('\nPlayer %d: %s\n' % (player, self.world.get_player_names(player)))
for team in range(self.world.teams):
outfile.write('%s%s\n' % (f"Hash - {self.world.player_names[player][team]} (Team {team+1}): " if self.world.teams > 1 else 'Hash: ', self.hashes[player, team]))
outfile.write('Logic: %s\n' % self.metadata['logic'][player])
outfile.write('Mode: %s\n' % self.metadata['mode'][player])
outfile.write('Retro: %s\n' % ('Yes' if self.metadata['retro'][player] else 'No'))
outfile.write('Swords: %s\n' % self.metadata['weapons'][player])
outfile.write('Goal: %s\n' % self.metadata['goal'][player])
outfile.write('Difficulty: %s\n' % self.metadata['item_pool'][player])
outfile.write('Item Functionality: %s\n' % self.metadata['item_functionality'][player])
outfile.write('Entrance Shuffle: %s\n' % self.metadata['shuffle'][player])
outfile.write('Crystals required for GT: %s\n' % self.metadata['gt_crystals'][player])
outfile.write('Crystals required for Ganon: %s\n' % self.metadata['ganon_crystals'][player])
outfile.write('Pyramid hole pre-opened: %s\n' % ('Yes' if self.metadata['open_pyramid'][player] else 'No'))
outfile.write('Accessibility: %s\n' % self.metadata['accessibility'][player])
outfile.write('Map shuffle: %s\n' % ('Yes' if self.metadata['mapshuffle'][player] else 'No'))
outfile.write('Compass shuffle: %s\n' % ('Yes' if self.metadata['compassshuffle'][player] else 'No'))
outfile.write('Small Key shuffle: %s\n' % ('Yes' if self.metadata['keyshuffle'][player] else 'No'))
outfile.write('Big Key shuffle: %s\n' % ('Yes' if self.metadata['bigkeyshuffle'][player] else 'No'))
outfile.write('Boss shuffle: %s\n' % self.metadata['boss_shuffle'][player])
outfile.write('Enemy shuffle: %s\n' % self.metadata['enemy_shuffle'][player])
outfile.write('Enemy health: %s\n' % self.metadata['enemy_health'][player])
outfile.write('Enemy damage: %s\n' % self.metadata['enemy_damage'][player])
outfile.write('Hints: %s\n' % ('Yes' if self.metadata['hints'][player] else 'No'))
if self.doors:
outfile.write('\n\nDoors:\n\n')
outfile.write('\n'.join(['%s%s %s %s' % ('Player {0}: '.format(entry['player']) if self.world.players > 1 else '', entry['entrance'], '<=>' if entry['direction'] == 'both' else '<=' if entry['direction'] == 'exit' else '=>', entry['exit']) for entry in self.doors.values()]))
@@ -1623,17 +1630,13 @@ class Spoiler(object):
outfile.write('\n'.join(['%s%s %s' % ('Player {0}: '.format(entry['player']) if self.world.players > 1 else '', entry['doorNames'], entry['type']) for entry in self.doorTypes.values()]))
if self.entrances:
outfile.write('\n\nEntrances:\n\n')
outfile.write('\n'.join(['%s%s %s %s' % ('Player {0}: '.format(entry['player']) if self.world.players > 1 else '', entry['entrance'], '<=>' if entry['direction'] == 'both' else '<=' if entry['direction'] == 'exit' else '=>', entry['exit']) for entry in self.entrances.values()]))
outfile.write('\n\nMedallions\n')
if self.world.players == 1:
outfile.write('\nMisery Mire Medallion: %s' % (self.medallions['Misery Mire']))
outfile.write('\nTurtle Rock Medallion: %s' % (self.medallions['Turtle Rock']))
else:
for player in range(1, self.world.players + 1):
outfile.write('\nMisery Mire Medallion (Player %d): %s' % (player, self.medallions['Misery Mire (Player %d)' % player]))
outfile.write('\nTurtle Rock Medallion (Player %d): %s' % (player, self.medallions['Turtle Rock (Player %d)' % player]))
outfile.write('\n\nStarting Inventory:\n\n')
outfile.write('\n'.join(self.startinventory))
outfile.write('\n'.join(['%s%s %s %s' % (f'{self.world.get_player_names(entry["player"])}: ' if self.world.players > 1 else '', entry['entrance'], '<=>' if entry['direction'] == 'both' else '<=' if entry['direction'] == 'exit' else '=>', entry['exit']) for entry in self.entrances.values()]))
outfile.write('\n\nMedallions:\n')
for dungeon, medallion in self.medallions.items():
outfile.write(f'\n{dungeon}: {medallion}')
if self.startinventory:
outfile.write('\n\nStarting Inventory:\n\n')
outfile.write('\n'.join(self.startinventory))
outfile.write('\n\nLocations:\n\n')
outfile.write('\n'.join(['%s: %s' % (location, item) for grouping in self.locations.values() for (location, item) in grouping.items()]))
outfile.write('\n\nShops:\n\n')

5
DungeonRandomizer.py
View File

@@ -278,8 +278,10 @@ def parse_arguments(argv, no_defaults=False):
parser.add_argument('--enemy_damage', default=defval('default'), choices=['default', 'shuffled', 'chaos'])
parser.add_argument('--shufflepots', default=defval(False), action='store_true')
parser.add_argument('--beemizer', default=defval(0), type=lambda value: min(max(int(value), 0), 4))
parser.add_argument('--remote_items', default=defval(False), action='store_true')
parser.add_argument('--multi', default=defval(1), type=lambda value: min(max(int(value), 1), 255))
parser.add_argument('--names', default=defval(''))
parser.add_argument('--teams', default=defval(1), type=lambda value: max(int(value), 1))
parser.add_argument('--outputpath')
parser.add_argument('--race', default=defval(False), action='store_true')
parser.add_argument('--outputname')
@@ -302,7 +304,8 @@ def parse_arguments(argv, no_defaults=False):
'mapshuffle', 'compassshuffle', 'keyshuffle', 'bigkeyshuffle', 'startinventory',
'retro', 'accessibility', 'hints', 'beemizer',
'shufflebosses', 'shuffleenemies', 'enemy_health', 'enemy_damage', 'shufflepots',
'ow_palettes', 'uw_palettes', 'sprite', 'disablemusic', 'quickswap', 'fastmenu', 'heartcolor', 'heartbeep']:
'ow_palettes', 'uw_palettes', 'sprite', 'disablemusic', 'quickswap', 'fastmenu', 'heartcolor', 'heartbeep',
'remote_items']:
value = getattr(defaults, name) if getattr(playerargs, name) is None else getattr(playerargs, name)
if player == 1:
setattr(ret, name, {1: value})

24
Gui.py
View File

@@ -15,7 +15,7 @@ from DungeonRandomizer import parse_arguments
from GuiUtils import ToolTips, set_icon, BackgroundTaskProgress
from Main import main, __version__ as ESVersion
from Rom import Sprite
from Utils import is_bundled, local_path, output_path, open_file, parse_names_string
from Utils import is_bundled, local_path, output_path, open_file
def guiMain(args=None):
@@ -509,11 +509,7 @@ def guiMain(args=None):
logging.exception(e)
messagebox.showerror(title="Error while creating seed", message=str(e))
else:
msgtxt = "Rom patched successfully"
if guiargs.names:
for player, name in parse_names_string(guiargs.names).items():
msgtxt += "\nPlayer %d => %s" % (player, name)
messagebox.showinfo(title="Success", message=msgtxt)
messagebox.showinfo(title="Success", message="Rom patched successfully")
generateButton = Button(bottomFrame, text='Generate Patched Rom', command=generateRom)
@@ -613,20 +609,11 @@ def guiMain(args=None):
uwPalettesLabel2 = Label(uwPalettesFrame2, text='Dungeon palettes')
uwPalettesLabel2.pack(side=LEFT)
namesFrame2 = Frame(drowDownFrame2)
namesLabel2 = Label(namesFrame2, text='Player names')
namesVar2 = StringVar()
namesEntry2 = Entry(namesFrame2, textvariable=namesVar2)
namesLabel2.pack(side=LEFT)
namesEntry2.pack(side=LEFT)
heartbeepFrame2.pack(expand=True, anchor=E)
heartcolorFrame2.pack(expand=True, anchor=E)
fastMenuFrame2.pack(expand=True, anchor=E)
owPalettesFrame2.pack(expand=True, anchor=E)
uwPalettesFrame2.pack(expand=True, anchor=E)
namesFrame2.pack(expand=True, anchor=E)
bottomFrame2 = Frame(topFrame2)
@@ -642,18 +629,13 @@ def guiMain(args=None):
guiargs.rom = romVar2.get()
guiargs.baserom = romVar.get()
guiargs.sprite = sprite
guiargs.names = namesEntry2.get()
try:
adjust(args=guiargs)
except Exception as e:
logging.exception(e)
messagebox.showerror(title="Error while creating seed", message=str(e))
else:
msgtxt = "Rom patched successfully"
if guiargs.names:
for player, name in parse_names_string(guiargs.names).items():
msgtxt += "\nPlayer %d => %s" % (player, name)
messagebox.showinfo(title="Success", message=msgtxt)
messagebox.showinfo(title="Success", message="Rom patched successfully")
adjustButton = Button(bottomFrame2, text='Adjust Rom', command=adjustRom)

116
Main.py
View File

@@ -13,7 +13,7 @@ from Items import ItemFactory
from Regions import create_regions, create_shops, mark_light_world_regions
from InvertedRegions import create_inverted_regions, mark_dark_world_regions
from EntranceShuffle import link_entrances, link_inverted_entrances
from Rom import patch_rom, patch_race_rom, patch_enemizer, apply_rom_settings, LocalRom, JsonRom
from Rom import patch_rom, patch_race_rom, patch_enemizer, apply_rom_settings, LocalRom, JsonRom, get_hash_string
from Doors import create_doors
from DoorShuffle import link_doors
from RoomData import create_rooms
@@ -21,7 +21,7 @@ from Rules import set_rules
from Dungeons import create_dungeons, fill_dungeons, fill_dungeons_restrictive
from Fill import distribute_items_cutoff, distribute_items_staleness, distribute_items_restrictive, flood_items, balance_multiworld_progression
from ItemList import generate_itempool, difficulties, fill_prizes
from Utils import output_path, parse_names_string
from Utils import output_path, parse_player_names
__version__ = '0.0.1-pre'
@@ -42,6 +42,7 @@ def main(args, seed=None):
world.seed = int(seed)
random.seed(world.seed)
world.remote_items = args.remote_items.copy()
world.mapshuffle = args.mapshuffle.copy()
world.compassshuffle = args.compassshuffle.copy()
world.keyshuffle = args.keyshuffle.copy()
@@ -57,7 +58,16 @@ def main(args, seed=None):
world.rom_seeds = {player: random.randint(0, 999999999) for player in range(1, world.players + 1)}
logger.info('ALttP Door Randomizer Version %s - Seed: %s\n\n', __version__, world.seed)
logger.info('ALttP Door Randomizer Version %s - Seed: %s\n', __version__, world.seed)
parsed_names = parse_player_names(args.names, world.players, args.teams)
world.teams = len(parsed_names)
for i, team in enumerate(parsed_names, 1):
if world.players > 1:
logger.info('%s%s', 'Team%d: ' % i if world.teams > 1 else 'Players: ', ', '.join(team))
for player, name in enumerate(team, 1):
world.player_names[player].append(name)
logger.info('')
for player in range(1, world.players + 1):
world.difficulty_requirements[player] = difficulties[world.difficulty[player]]
@@ -144,62 +154,68 @@ def main(args, seed=None):
logger.info('Patching ROM.')
player_names = parse_names_string(args.names)
outfilebase = 'DR_%s' % (args.outputname if args.outputname else world.seed)
rom_names = []
jsonout = {}
if not args.suppress_rom:
for player in range(1, world.players + 1):
sprite_random_on_hit = type(args.sprite[player]) is str and args.sprite[player].lower() == 'randomonhit'
use_enemizer = (world.boss_shuffle[player] != 'none' or world.enemy_shuffle[player] != 'none'
or world.enemy_health[player] != 'default' or world.enemy_damage[player] != 'default'
or args.shufflepots[player] or sprite_random_on_hit)
for team in range(world.teams):
for player in range(1, world.players + 1):
sprite_random_on_hit = type(args.sprite[player]) is str and args.sprite[player].lower() == 'randomonhit'
use_enemizer = (world.boss_shuffle[player] != 'none' or world.enemy_shuffle[player] != 'none'
or world.enemy_health[player] != 'default' or world.enemy_damage[player] != 'default'
or args.shufflepots[player] or sprite_random_on_hit)
rom = JsonRom() if args.jsonout or use_enemizer else LocalRom(args.rom)
rom = JsonRom() if args.jsonout or use_enemizer else LocalRom(args.rom)
patch_rom(world, player, rom, use_enemizer)
rom_names.append((player, list(rom.name)))
patch_rom(world, rom, player, team, use_enemizer)
if use_enemizer and (args.enemizercli or not args.jsonout):
patch_enemizer(world, player, rom, args.rom, args.enemizercli, args.shufflepots[player], sprite_random_on_hit)
if not args.jsonout:
patches = rom.patches
rom = LocalRom(args.rom)
rom.merge_enemizer_patches(patches)
if use_enemizer and (args.enemizercli or not args.jsonout):
patch_enemizer(world, player, rom, args.rom, args.enemizercli, args.shufflepots[player], sprite_random_on_hit)
if not args.jsonout:
rom = LocalRom.fromJsonRom(rom, args.rom, 0x400000)
if args.race:
patch_race_rom(rom)
if args.race:
patch_race_rom(rom)
apply_rom_settings(rom, args.heartbeep[player], args.heartcolor[player], args.quickswap[player], args.fastmenu[player], args.disablemusic[player], args.sprite[player], args.ow_palettes[player], args.uw_palettes[player], player_names)
rom_names.append((player, team, list(rom.name)))
world.spoiler.hashes[(player, team)] = get_hash_string(rom.hash)
if args.jsonout:
jsonout[f'patch{player}'] = rom.patches
else:
mcsb_name = ''
if all([world.mapshuffle[player], world.compassshuffle[player], world.keyshuffle[player], world.bigkeyshuffle[player]]):
mcsb_name = '-keysanity'
elif [world.mapshuffle[player], world.compassshuffle[player], world.keyshuffle[player], world.bigkeyshuffle[player]].count(True) == 1:
mcsb_name = '-mapshuffle' if world.mapshuffle[player] else '-compassshuffle' if world.compassshuffle[player] else '-keyshuffle' if world.keyshuffle[player] else '-bigkeyshuffle'
elif any([world.mapshuffle[player], world.compassshuffle[player], world.keyshuffle[player], world.bigkeyshuffle[player]]):
mcsb_name = '-%s%s%s%sshuffle' % (
'M' if world.mapshuffle[player] else '', 'C' if world.compassshuffle[player] else '',
'S' if world.keyshuffle[player] else '', 'B' if world.bigkeyshuffle[player] else '')
apply_rom_settings(rom, args.heartbeep[player], args.heartcolor[player], args.quickswap[player], args.fastmenu[player], args.disablemusic[player], args.sprite[player], args.ow_palettes[player], args.uw_palettes[player])
playername = f"{f'_P{player}' if world.players > 1 else ''}{f'_{player_names[player]}' if player in player_names else ''}"
outfilesuffix = ('_%s_%s-%s-%s-%s%s_%s-%s-%s%s%s%s%s' % (world.logic[player], world.difficulty[player], world.difficulty_adjustments[player],
world.mode[player], world.goal[player],
"" if world.timer in ['none', 'display'] else "-" + world.timer,
world.shuffle[player], world.doorShuffle[player], world.algorithm, mcsb_name,
"-retro" if world.retro[player] else "",
"-prog_" + world.progressive if world.progressive in ['off', 'random'] else "",
"-nohints" if not world.hints[player] else "")) if not args.outputname else ''
rom.write_to_file(output_path(f'{outfilebase}{playername}{outfilesuffix}.sfc'))
if args.jsonout:
jsonout[f'patch_t{team}_p{player}'] = rom.patches
else:
mcsb_name = ''
if all([world.mapshuffle[player], world.compassshuffle[player], world.keyshuffle[player], world.bigkeyshuffle[player]]):
mcsb_name = '-keysanity'
elif [world.mapshuffle[player], world.compassshuffle[player], world.keyshuffle[player], world.bigkeyshuffle[player]].count(True) == 1:
mcsb_name = '-mapshuffle' if world.mapshuffle[player] else '-compassshuffle' if world.compassshuffle[player] else '-keyshuffle' if world.keyshuffle[player] else '-bigkeyshuffle'
elif any([world.mapshuffle[player], world.compassshuffle[player], world.keyshuffle[player], world.bigkeyshuffle[player]]):
mcsb_name = '-%s%s%s%sshuffle' % (
'M' if world.mapshuffle[player] else '', 'C' if world.compassshuffle[player] else '',
'S' if world.keyshuffle[player] else '', 'B' if world.bigkeyshuffle[player] else '')
multidata = zlib.compress(json.dumps((world.players,
rom_names,
[((location.address, location.player), (location.item.code, location.item.player)) for location in world.get_filled_locations() if type(location.address) is int])
).encode("utf-8"))
outfilepname = f'_T{team+1}' if world.teams > 1 else ''
if world.players > 1:
outfilepname += f'_P{player}'
if world.players > 1 or world.teams > 1:
outfilepname += f"_{world.player_names[player][team].replace(' ', '_')}" if world.player_names[player][team] != 'Player %d' % player else ''
outfilesuffix = ('_%s_%s-%s-%s-%s%s_%s-%s%s%s%s%s' % (world.logic[player], world.difficulty[player], world.difficulty_adjustments[player],
world.mode[player], world.goal[player],
"" if world.timer in ['none', 'display'] else "-" + world.timer,
world.shuffle[player], world.algorithm, mcsb_name,
"-retro" if world.retro[player] else "",
"-prog_" + world.progressive if world.progressive in ['off', 'random'] else "",
"-nohints" if not world.hints[player] else "")) if not args.outputname else ''
rom.write_to_file(output_path(f'{outfilebase}{outfilepname}{outfilesuffix}.sfc'))
multidata = zlib.compress(json.dumps({"names": parsed_names,
"roms": rom_names,
"remote_items": [player for player in range(1, world.players + 1) if world.remote_items[player]],
"locations": [((location.address, location.player), (location.item.code, location.item.player))
for location in world.get_filled_locations() if type(location.address) is int]
}).encode("utf-8"))
if args.jsonout:
jsonout["multidata"] = list(multidata)
else:
@@ -230,6 +246,9 @@ def main(args, seed=None):
def copy_world(world):
# ToDo: Not good yet
ret = World(world.players, world.shuffle, world.doorShuffle, world.logic, world.mode, world.swords, world.difficulty, world.difficulty_adjustments, world.timer, world.progressive, world.goal, world.algorithm, world.accessibility, world.shuffle_ganon, world.retro, world.custom, world.customitemarray, world.hints)
ret.teams = world.teams
ret.player_names = copy.deepcopy(world.player_names)
ret.remote_items = world.remote_items.copy()
ret.required_medallions = world.required_medallions.copy()
ret.swamp_patch_required = world.swamp_patch_required.copy()
ret.ganon_at_pyramid = world.ganon_at_pyramid.copy()
@@ -297,6 +316,7 @@ def copy_world(world):
item = Item(location.item.name, location.item.advancement, location.item.priority, location.item.type, player = location.item.player)
ret.get_location(location.name, location.player).item = item
item.location = ret.get_location(location.name, location.player)
item.world = ret
if location.event:
ret.get_location(location.name, location.player).event = True
if location.locked:
@@ -306,9 +326,11 @@ def copy_world(world):
for item in world.itempool:
ret.itempool.append(Item(item.name, item.advancement, item.priority, item.type, player = item.player))
for item in world.precollected_items:
ret.push_precollected(ItemFactory(item.name, item.player))
# copy progress items in state
ret.state.prog_items = world.state.prog_items.copy()
ret.precollected_items = world.precollected_items.copy()
ret.state.stale = {player: True for player in range(1, world.players + 1)}
ret.doors = world.doors

440
MultiClient.py
View File

@@ -1,58 +1,38 @@
import aioconsole
import argparse
import asyncio
import colorama
import json
import logging
import re
import subprocess
import sys
import shlex
import urllib.parse
import websockets
import Items
import Regions
while True:
try:
import aioconsole
break
except ImportError:
aioconsole = None
print('Required python module "aioconsole" not found, press enter to install it')
input()
subprocess.call([sys.executable, '-m', 'pip', 'install', '--upgrade', 'aioconsole'])
while True:
try:
import websockets
break
except ImportError:
websockets = None
print('Required python module "websockets" not found, press enter to install it')
input()
subprocess.call([sys.executable, '-m', 'pip', 'install', '--upgrade', 'websockets'])
try:
import colorama
except ImportError:
colorama = None
class ReceivedItem:
def __init__(self, item, location, player_id, player_name):
def __init__(self, item, location, player):
self.item = item
self.location = location
self.player_id = player_id
self.player_name = player_name
self.player = player
class Context:
def __init__(self, snes_address, server_address, password, name, team, slot):
def __init__(self, snes_address, server_address, password):
self.snes_address = snes_address
self.server_address = server_address
self.exit_event = asyncio.Event()
self.watcher_event = asyncio.Event()
self.input_queue = asyncio.Queue()
self.input_requests = 0
self.snes_socket = None
self.snes_state = SNES_DISCONNECTED
self.snes_attached_device = None
self.snes_reconnect_address = None
self.snes_recv_queue = asyncio.Queue()
self.snes_request_lock = asyncio.Lock()
self.is_sd2snes = False
@@ -62,15 +42,16 @@ class Context:
self.socket = None
self.password = password
self.name = name
self.team = team
self.slot = slot
self.team = None
self.slot = None
self.player_names = {}
self.locations_checked = set()
self.locations_scouted = set()
self.items_received = []
self.last_rom = None
self.expected_rom = None
self.rom_confirmed = False
self.locations_info = {}
self.awaiting_rom = False
self.rom = None
self.auth = None
def color_code(*args):
codes = {'reset': 0, 'bold': 1, 'underline': 4, 'black': 30, 'red': 31, 'green': 32, 'yellow': 33, 'blue': 34,
@@ -81,6 +62,7 @@ def color_code(*args):
def color(text, *args):
return color_code(*args) + text + color_code('reset')
RECONNECT_DELAY = 30
ROM_START = 0x000000
WRAM_START = 0xF50000
@@ -95,11 +77,15 @@ INGAME_MODES = {0x07, 0x09, 0x0b}
SAVEDATA_START = WRAM_START + 0xF000
SAVEDATA_SIZE = 0x500
RECV_PROGRESS_ADDR = SAVEDATA_START + 0x4D0 # 2 bytes
RECV_ITEM_ADDR = SAVEDATA_START + 0x4D2 # 1 byte
RECV_ITEM_PLAYER_ADDR = SAVEDATA_START + 0x4D3 # 1 byte
ROOMID_ADDR = SAVEDATA_START + 0x4D4 # 2 bytes
ROOMDATA_ADDR = SAVEDATA_START + 0x4D6 # 1 byte
RECV_PROGRESS_ADDR = SAVEDATA_START + 0x4D0 # 2 bytes
RECV_ITEM_ADDR = SAVEDATA_START + 0x4D2 # 1 byte
RECV_ITEM_PLAYER_ADDR = SAVEDATA_START + 0x4D3 # 1 byte
ROOMID_ADDR = SAVEDATA_START + 0x4D4 # 2 bytes
ROOMDATA_ADDR = SAVEDATA_START + 0x4D6 # 1 byte
SCOUT_LOCATION_ADDR = SAVEDATA_START + 0x4D7 # 1 byte
SCOUTREPLY_LOCATION_ADDR = SAVEDATA_START + 0x4D8 # 1 byte
SCOUTREPLY_ITEM_ADDR = SAVEDATA_START + 0x4D9 # 1 byte
SCOUTREPLY_PLAYER_ADDR = SAVEDATA_START + 0x4DA # 1 byte
location_table_uw = {"Blind's Hideout - Top": (0x11d, 0x10),
"Blind's Hideout - Left": (0x11d, 0x20),
@@ -323,18 +309,17 @@ SNES_CONNECTING = 1
SNES_CONNECTED = 2
SNES_ATTACHED = 3
async def snes_connect(ctx : Context, address = None):
async def snes_connect(ctx : Context, address):
if ctx.snes_socket is not None:
print('Already connected to snes')
logging.error('Already connected to snes')
return
ctx.snes_state = SNES_CONNECTING
recv_task = None
if address is None:
address = 'ws://' + ctx.snes_address
address = f"ws://{address}" if "://" not in address else address
print("Connecting to QUsb2snes at %s ..." % address)
logging.info("Connecting to QUsb2snes at %s ..." % address)
try:
ctx.snes_socket = await websockets.connect(address, ping_timeout=None, ping_interval=None)
@@ -352,24 +337,32 @@ async def snes_connect(ctx : Context, address = None):
if not devices:
raise Exception('No device found')
print("Available devices:")
logging.info("Available devices:")
for id, device in enumerate(devices):
print("[%d] %s" % (id + 1, device))
logging.info("[%d] %s" % (id + 1, device))
device = None
while True:
print("Enter a number:")
choice = await console_input(ctx)
if choice is None:
raise Exception('Abort input')
if not choice.isdigit() or int(choice) < 1 or int(choice) > len(devices):
print("Invalid choice (%s)" % choice)
continue
if len(devices) == 1:
device = devices[0]
elif ctx.snes_reconnect_address:
if ctx.snes_attached_device[1] in devices:
device = ctx.snes_attached_device[1]
else:
device = devices[ctx.snes_attached_device[0]]
else:
while True:
logging.info("Select a device:")
choice = await console_input(ctx)
if choice is None:
raise Exception('Abort input')
if not choice.isdigit() or int(choice) < 1 or int(choice) > len(devices):
logging.warning("Invalid choice (%s)" % choice)
continue
device = devices[int(choice) - 1]
break
device = devices[int(choice) - 1]
break
print("Attaching to " + device)
logging.info("Attaching to " + device)
Attach_Request = {
"Opcode" : "Attach",
@@ -378,21 +371,22 @@ async def snes_connect(ctx : Context, address = None):
}
await ctx.snes_socket.send(json.dumps(Attach_Request))
ctx.snes_state = SNES_ATTACHED
ctx.snes_attached_device = (devices.index(device), device)
if 'SD2SNES'.lower() in device.lower() or (len(device) == 4 and device[:3] == 'COM'):
print("SD2SNES Detected")
logging.info("SD2SNES Detected")
ctx.is_sd2snes = True
await ctx.snes_socket.send(json.dumps({"Opcode" : "Info", "Space" : "SNES"}))
reply = json.loads(await ctx.snes_socket.recv())
if reply and 'Results' in reply:
print(reply['Results'])
logging.info(reply['Results'])
else:
ctx.is_sd2snes = False
ctx.snes_reconnect_address = address
recv_task = asyncio.create_task(snes_recv_loop(ctx))
except Exception as e:
print("Error connecting to snes (%s)" % e)
if recv_task is not None:
if not ctx.snes_socket.closed:
await ctx.snes_socket.close()
@@ -402,16 +396,26 @@ async def snes_connect(ctx : Context, address = None):
await ctx.snes_socket.close()
ctx.snes_socket = None
ctx.snes_state = SNES_DISCONNECTED
if not ctx.snes_reconnect_address:
logging.error("Error connecting to snes (%s)" % e)
else:
logging.error(f"Error connecting to snes, attempt again in {RECONNECT_DELAY}s")
asyncio.create_task(snes_autoreconnect(ctx))
async def snes_autoreconnect(ctx: Context):
await asyncio.sleep(RECONNECT_DELAY)
if ctx.snes_reconnect_address and ctx.snes_socket is None:
await snes_connect(ctx, ctx.snes_reconnect_address)
async def snes_recv_loop(ctx : Context):
try:
async for msg in ctx.snes_socket:
ctx.snes_recv_queue.put_nowait(msg)
print("Snes disconnected, type /snes to reconnect")
logging.warning("Snes disconnected")
except Exception as e:
print("Lost connection to the snes, type /snes to reconnect")
if not isinstance(e, websockets.WebSocketException):
logging.exception(e)
logging.error("Lost connection to the snes, type /snes to reconnect")
finally:
socket, ctx.snes_socket = ctx.snes_socket, None
if socket is not None and not socket.closed:
@@ -421,8 +425,11 @@ async def snes_recv_loop(ctx : Context):
ctx.snes_recv_queue = asyncio.Queue()
ctx.hud_message_queue = []
ctx.rom_confirmed = False
ctx.last_rom = None
ctx.rom = None
if ctx.snes_reconnect_address:
logging.info(f"...reconnecting in {RECONNECT_DELAY}s")
asyncio.create_task(snes_autoreconnect(ctx))
async def snes_read(ctx : Context, address, size):
try:
@@ -449,9 +456,9 @@ async def snes_read(ctx : Context, address, size):
break
if len(data) != size:
print('Error reading %s, requested %d bytes, received %d' % (hex(address), size, len(data)))
logging.error('Error reading %s, requested %d bytes, received %d' % (hex(address), size, len(data)))
if len(data):
print(str(data))
logging.error(str(data))
if ctx.snes_socket is not None and not ctx.snes_socket.closed:
await ctx.snes_socket.close()
return None
@@ -477,7 +484,7 @@ async def snes_write(ctx : Context, write_list):
for address, data in write_list:
if (address < WRAM_START) or ((address + len(data)) > (WRAM_START + WRAM_SIZE)):
print("SD2SNES: Write out of range %s (%d)" % (hex(address), len(data)))
logging.error("SD2SNES: Write out of range %s (%d)" % (hex(address), len(data)))
return False
for ptr, byte in enumerate(data, address + 0x7E0000 - WRAM_START):
cmd += b'\xA9' # LDA
@@ -534,100 +541,100 @@ async def send_msgs(websocket, msgs):
except websockets.ConnectionClosed:
pass
async def server_loop(ctx : Context):
async def server_loop(ctx : Context, address = None):
if ctx.socket is not None:
print('Already connected')
logging.error('Already connected')
return
while not ctx.server_address:
print('Enter multiworld server address')
ctx.server_address = await console_input(ctx)
if address is None:
address = ctx.server_address
address = f"ws://{ctx.server_address}" if "://" not in ctx.server_address else ctx.server_address
while not address:
logging.info('Enter multiworld server address')
address = await console_input(ctx)
print('Connecting to multiworld server at %s' % address)
address = f"ws://{address}" if "://" not in address else address
port = urllib.parse.urlparse(address).port or 38281
logging.info('Connecting to multiworld server at %s' % address)
try:
ctx.socket = await websockets.connect(address, ping_timeout=None, ping_interval=None)
print('Connected')
ctx.socket = await websockets.connect(address, port=port, ping_timeout=None, ping_interval=None)
logging.info('Connected')
ctx.server_address = address
async for data in ctx.socket:
for msg in json.loads(data):
cmd, args = (msg[0], msg[1]) if len(msg) > 1 else (msg, None)
await process_server_cmd(ctx, cmd, args)
print('Disconnected from multiworld server, type /connect to reconnect')
logging.warning('Disconnected from multiworld server, type /connect to reconnect')
except ConnectionRefusedError:
print('Connection refused by the multiworld server')
logging.error('Connection refused by the multiworld server')
except (OSError, websockets.InvalidURI):
print('Failed to connect to the multiworld server')
logging.error('Failed to connect to the multiworld server')
except Exception as e:
print('Lost connection to the multiworld server, type /connect to reconnect')
logging.error('Lost connection to the multiworld server, type /connect to reconnect')
if not isinstance(e, websockets.WebSocketException):
logging.exception(e)
finally:
ctx.name = None
ctx.team = None
ctx.slot = None
ctx.expected_rom = None
ctx.rom_confirmed = False
ctx.awaiting_rom = False
ctx.auth = None
ctx.items_received = []
ctx.locations_info = {}
socket, ctx.socket = ctx.socket, None
if socket is not None and not socket.closed:
await socket.close()
ctx.server_task = None
if ctx.server_address:
logging.info(f"... reconnecting in {RECONNECT_DELAY}s")
asyncio.create_task(server_autoreconnect(ctx))
async def server_autoreconnect(ctx: Context):
await asyncio.sleep(RECONNECT_DELAY)
if ctx.server_address and ctx.server_task is None:
ctx.server_task = asyncio.create_task(server_loop(ctx))
async def process_server_cmd(ctx : Context, cmd, args):
if cmd == 'RoomInfo':
print('--------------------------------')
print('Room Information:')
print('--------------------------------')
logging.info('--------------------------------')
logging.info('Room Information:')
logging.info('--------------------------------')
if args['password']:
print('Password required')
print('%d players seed' % args['slots'])
logging.info('Password required')
if len(args['players']) < 1:
print('No player connected')
logging.info('No player connected')
else:
args['players'].sort(key=lambda player: ('' if not player[1] else player[1].lower(), player[2]))
args['players'].sort()
current_team = 0
print('Connected players:')
for name, team, slot in args['players']:
logging.info('Connected players:')
logging.info(' Team #1')
for team, slot, name in args['players']:
if team != current_team:
print(' Default team' if not team else ' Team: %s' % team)
logging.info(f' Team #{team + 1}')
current_team = team
print(' %s (Player %d)' % (name, slot))
logging.info(' %s (Player %d)' % (name, slot))
await server_auth(ctx, args['password'])
if cmd == 'ConnectionRefused':
password_requested = False
if 'InvalidPassword' in args:
print('Invalid password')
logging.error('Invalid password')
ctx.password = None
password_requested = True
if 'InvalidName' in args:
print('Invalid name')
ctx.name = None
if 'NameAlreadyTaken' in args:
print('Name already taken')
ctx.name = None
if 'InvalidTeam' in args:
print('Invalid team name')
ctx.team = None
if 'InvalidSlot' in args:
print('Invalid player slot')
ctx.slot = None
await server_auth(ctx, True)
if 'InvalidRom' in args:
raise Exception('Invalid ROM detected, please verify that you have loaded the correct rom and reconnect your snes')
if 'SlotAlreadyTaken' in args:
print('Player slot already in use for that team')
ctx.team = None
ctx.slot = None
await server_auth(ctx, password_requested)
raise Exception('Player slot already in use for that team')
raise Exception('Connection refused by the multiworld host')
if cmd == 'Connected':
ctx.expected_rom = args
if ctx.last_rom is not None:
if ctx.last_rom[:len(args)] == ctx.expected_rom:
rom_confirmed(ctx)
if ctx.locations_checked:
await send_msgs(ctx.socket, [['LocationChecks', [Regions.location_table[loc][0] for loc in ctx.locations_checked]]])
else:
raise Exception('Different ROM expected from server')
ctx.team, ctx.slot = args[0]
ctx.player_names = {p: n for p, n in args[1]}
msgs = []
if ctx.locations_checked:
msgs.append(['LocationChecks', [Regions.location_table[loc][0] for loc in ctx.locations_checked]])
if ctx.locations_scouted:
msgs.append(['LocationScouts', list(ctx.locations_scouted)])
if msgs:
await send_msgs(ctx.socket, msgs)
if cmd == 'ReceivedItems':
start_index, items = args
@@ -640,39 +647,54 @@ async def process_server_cmd(ctx : Context, cmd, args):
await send_msgs(ctx.socket, sync_msg)
if start_index == len(ctx.items_received):
for item in items:
ctx.items_received.append(ReceivedItem(item[0], item[1], item[2], item[3]))
ctx.items_received.append(ReceivedItem(*item))
ctx.watcher_event.set()
if cmd == 'LocationInfo':
for location, item, player in args:
if location not in ctx.locations_info:
replacements = {0xA2: 'Small Key', 0x9D: 'Big Key', 0x8D: 'Compass', 0x7D: 'Map'}
item_name = replacements.get(item, get_item_name_from_id(item))
logging.info(f"Saw {color(item_name, 'red', 'bold')} at {list(Regions.location_table.keys())[location - 1]}")
ctx.locations_info[location] = (item, player)
ctx.watcher_event.set()
if cmd == 'ItemSent':
player_sent, player_recvd, item, location = args
item = color(get_item_name_from_id(item), 'cyan' if player_sent != ctx.name else 'green')
player_sent = color(player_sent, 'yellow' if player_sent != ctx.name else 'magenta')
player_recvd = color(player_recvd, 'yellow' if player_recvd != ctx.name else 'magenta')
print('(%s) %s sent %s to %s (%s)' % (ctx.team if ctx.team else 'Team', player_sent, item, player_recvd, get_location_name_from_address(location)))
player_sent, location, player_recvd, item = args
item = color(get_item_name_from_id(item), 'cyan' if player_sent != ctx.slot else 'green')
player_sent = color(ctx.player_names[player_sent], 'yellow' if player_sent != ctx.slot else 'magenta')
player_recvd = color(ctx.player_names[player_recvd], 'yellow' if player_recvd != ctx.slot else 'magenta')
logging.info('%s sent %s to %s (%s)' % (player_sent, item, player_recvd, get_location_name_from_address(location)))
if cmd == 'Print':
print(args)
logging.info(args)
async def server_auth(ctx : Context, password_requested):
if password_requested and not ctx.password:
print('Enter the password required to join this game:')
logging.info('Enter the password required to join this game:')
ctx.password = await console_input(ctx)
while not ctx.name or not re.match(r'\w{1,10}', ctx.name):
print('Enter your name (10 characters):')
ctx.name = await console_input(ctx)
if not ctx.team:
print('Enter your team name (optional):')
ctx.team = await console_input(ctx)
if ctx.team == '': ctx.team = None
if not ctx.slot:
print('Choose your player slot (optional):')
slot = await console_input(ctx)
ctx.slot = int(slot) if slot.isdigit() else None
await send_msgs(ctx.socket, [['Connect', {'password': ctx.password, 'name': ctx.name, 'team': ctx.team, 'slot': ctx.slot}]])
if ctx.rom is None:
ctx.awaiting_rom = True
logging.info('No ROM detected, awaiting snes connection to authenticate to the multiworld server')
return
ctx.awaiting_rom = False
ctx.auth = ctx.rom.copy()
await send_msgs(ctx.socket, [['Connect', {'password': ctx.password, 'rom': ctx.auth}]])
async def console_input(ctx : Context):
ctx.input_requests += 1
return await ctx.input_queue.get()
async def disconnect(ctx: Context):
if ctx.socket is not None and not ctx.socket.closed:
await ctx.socket.close()
if ctx.server_task is not None:
await ctx.server_task
async def connect(ctx: Context, address=None):
await disconnect(ctx)
ctx.server_task = asyncio.create_task(server_loop(ctx, address))
async def console_loop(ctx : Context):
while not ctx.exit_event.is_set():
input = await aioconsole.ainput()
@@ -682,70 +704,53 @@ async def console_loop(ctx : Context):
ctx.input_queue.put_nowait(input)
continue
command = input.split()
command = shlex.split(input)
if not command:
continue
if command[0] == '/exit':
ctx.exit_event.set()
if command[0] == '/installcolors' and 'colorama' not in sys.modules:
subprocess.call([sys.executable, '-m', 'pip', 'install', '--upgrade', 'colorama'])
global colorama
import colorama
colorama.init()
if command[0] == '/snes':
asyncio.create_task(snes_connect(ctx, command[1] if len(command) > 1 else None))
ctx.snes_reconnect_address = None
asyncio.create_task(snes_connect(ctx, command[1] if len(command) > 1 else ctx.snes_address))
if command[0] in ['/snes_close', '/snes_quit']:
ctx.snes_reconnect_address = None
if ctx.snes_socket is not None and not ctx.snes_socket.closed:
await ctx.snes_socket.close()
async def disconnect():
if ctx.socket is not None and not ctx.socket.closed:
await ctx.socket.close()
if ctx.server_task is not None:
await ctx.server_task
async def connect():
await disconnect()
ctx.server_task = asyncio.create_task(server_loop(ctx))
if command[0] in ['/connect', '/reconnect']:
if len(command) > 1:
ctx.server_address = command[1]
asyncio.create_task(connect())
ctx.server_address = None
asyncio.create_task(connect(ctx, command[1] if len(command) > 1 else None))
if command[0] == '/disconnect':
asyncio.create_task(disconnect())
ctx.server_address = None
asyncio.create_task(disconnect(ctx))
if command[0][:1] != '/':
asyncio.create_task(send_msgs(ctx.socket, [['Say', input]]))
if command[0] == '/received':
print('Received items:')
logging.info('Received items:')
for index, item in enumerate(ctx.items_received, 1):
print('%s from %s (%s) (%d/%d in list)' % (
color(get_item_name_from_id(item.item), 'red', 'bold'), color(item.player_name, 'yellow'),
logging.info('%s from %s (%s) (%d/%d in list)' % (
color(get_item_name_from_id(item.item), 'red', 'bold'), color(ctx.player_names[item.player], 'yellow'),
get_location_name_from_address(item.location), index, len(ctx.items_received)))
if command[0] == '/missing':
for location in [k for k, v in Regions.location_table.items() if type(v[0]) is int]:
if location not in ctx.locations_checked:
print('Missing: ' + location)
logging.info('Missing: ' + location)
if command[0] == '/getitem' and len(command) > 1:
item = input[9:]
item_id = Items.item_table[item][3] if item in Items.item_table else None
if type(item_id) is int and item_id in range(0x100):
print('Sending item: ' + item)
logging.info('Sending item: ' + item)
snes_buffered_write(ctx, RECV_ITEM_ADDR, bytes([item_id]))
snes_buffered_write(ctx, RECV_ITEM_PLAYER_ADDR, bytes([0]))
else:
print('Invalid item: ' + item)
logging.info('Invalid item: ' + item)
await snes_flush_writes(ctx)
def rom_confirmed(ctx : Context):
ctx.rom_confirmed = True
print('ROM hash Confirmed')
def get_item_name_from_id(code):
items = [k for k, i in Items.item_table.items() if type(i[3]) is int and i[3] == code]
return items[0] if items else 'Unknown item'
@@ -761,7 +766,7 @@ async def track_locations(ctx : Context, roomid, roomdata):
new_locations = []
def new_check(location):
ctx.locations_checked.add(location)
print("New check: %s (%d/216)" % (location, len(ctx.locations_checked)))
logging.info("New check: %s (%d/216)" % (location, len(ctx.locations_checked)))
new_locations.append(Regions.location_table[location][0])
for location, (loc_roomid, loc_mask) in location_table_uw.items():
@@ -820,68 +825,82 @@ async def track_locations(ctx : Context, roomid, roomdata):
async def game_watcher(ctx : Context):
while not ctx.exit_event.is_set():
await asyncio.sleep(2)
try:
await asyncio.wait_for(ctx.watcher_event.wait(), 2)
except asyncio.TimeoutError:
pass
ctx.watcher_event.clear()
if not ctx.rom_confirmed:
if not ctx.rom:
rom = await snes_read(ctx, ROMNAME_START, ROMNAME_SIZE)
if rom is None or rom == bytes([0] * ROMNAME_SIZE):
continue
if list(rom) != ctx.last_rom:
ctx.last_rom = list(rom)
ctx.locations_checked = set()
if ctx.expected_rom is not None:
if ctx.last_rom[:len(ctx.expected_rom)] != ctx.expected_rom:
print("Wrong ROM detected")
await ctx.snes_socket.close()
continue
else:
rom_confirmed(ctx)
ctx.rom = list(rom)
ctx.locations_checked = set()
ctx.locations_scouted = set()
if ctx.awaiting_rom:
await server_auth(ctx, False)
if ctx.auth and ctx.auth != ctx.rom:
logging.warning("ROM change detected, please reconnect to the multiworld server")
await disconnect(ctx)
gamemode = await snes_read(ctx, WRAM_START + 0x10, 1)
if gamemode is None or gamemode[0] not in INGAME_MODES:
continue
data = await snes_read(ctx, RECV_PROGRESS_ADDR, 7)
data = await snes_read(ctx, RECV_PROGRESS_ADDR, 8)
if data is None:
continue
recv_index = data[0] | (data[1] << 8)
assert(RECV_ITEM_ADDR == RECV_PROGRESS_ADDR + 2)
assert RECV_ITEM_ADDR == RECV_PROGRESS_ADDR + 2
recv_item = data[2]
assert(ROOMID_ADDR == RECV_PROGRESS_ADDR + 4)
assert ROOMID_ADDR == RECV_PROGRESS_ADDR + 4
roomid = data[4] | (data[5] << 8)
assert(ROOMDATA_ADDR == RECV_PROGRESS_ADDR + 6)
assert ROOMDATA_ADDR == RECV_PROGRESS_ADDR + 6
roomdata = data[6]
await track_locations(ctx, roomid, roomdata)
assert SCOUT_LOCATION_ADDR == RECV_PROGRESS_ADDR + 7
scout_location = data[7]
if recv_index < len(ctx.items_received) and recv_item == 0:
item = ctx.items_received[recv_index]
print('Received %s from %s (%s) (%d/%d in list)' % (
color(get_item_name_from_id(item.item), 'red', 'bold'), color(item.player_name, 'yellow'),
logging.info('Received %s from %s (%s) (%d/%d in list)' % (
color(get_item_name_from_id(item.item), 'red', 'bold'), color(ctx.player_names[item.player], 'yellow'),
get_location_name_from_address(item.location), recv_index + 1, len(ctx.items_received)))
recv_index += 1
snes_buffered_write(ctx, RECV_PROGRESS_ADDR, bytes([recv_index & 0xFF, (recv_index >> 8) & 0xFF]))
snes_buffered_write(ctx, RECV_ITEM_ADDR, bytes([item.item]))
snes_buffered_write(ctx, RECV_ITEM_PLAYER_ADDR, bytes([item.player_id]))
snes_buffered_write(ctx, RECV_ITEM_PLAYER_ADDR, bytes([item.player if item.player != ctx.slot else 0]))
if scout_location > 0 and scout_location in ctx.locations_info:
snes_buffered_write(ctx, SCOUTREPLY_LOCATION_ADDR, bytes([scout_location]))
snes_buffered_write(ctx, SCOUTREPLY_ITEM_ADDR, bytes([ctx.locations_info[scout_location][0]]))
snes_buffered_write(ctx, SCOUTREPLY_PLAYER_ADDR, bytes([ctx.locations_info[scout_location][1]]))
await snes_flush_writes(ctx)
if scout_location > 0 and scout_location not in ctx.locations_scouted:
ctx.locations_scouted.add(scout_location)
logging.info(f'Scouting item at {list(Regions.location_table.keys())[scout_location - 1]}')
await send_msgs(ctx.socket, [['LocationScouts', [scout_location]]])
await track_locations(ctx, roomid, roomdata)
async def main():
parser = argparse.ArgumentParser()
parser.add_argument('--snes', default='localhost:8080', help='Address of the QUsb2snes server.')
parser.add_argument('--connect', default=None, help='Address of the multiworld host.')
parser.add_argument('--password', default=None, help='Password of the multiworld host.')
parser.add_argument('--name', default=None)
parser.add_argument('--team', default=None)
parser.add_argument('--slot', default=None, type=int)
parser.add_argument('--loglevel', default='info', choices=['debug', 'info', 'warning', 'error', 'critical'])
args = parser.parse_args()
ctx = Context(args.snes, args.connect, args.password, args.name, args.team, args.slot)
logging.basicConfig(format='%(message)s', level=getattr(logging, args.loglevel.upper(), logging.INFO))
ctx = Context(args.snes, args.connect, args.password)
input_task = asyncio.create_task(console_loop(ctx))
await snes_connect(ctx)
await snes_connect(ctx, ctx.snes_address)
if ctx.server_task is None:
ctx.server_task = asyncio.create_task(server_loop(ctx))
@@ -890,7 +909,8 @@ async def main():
await ctx.exit_event.wait()
ctx.server_address = None
ctx.snes_reconnect_address = None
await watcher_task
@@ -909,13 +929,9 @@ async def main():
await input_task
if __name__ == '__main__':
if 'colorama' in sys.modules:
colorama.init()
colorama.init()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.run_until_complete(asyncio.gather(*asyncio.Task.all_tasks()))
loop.close()
if 'colorama' in sys.modules:
colorama.deinit()
colorama.deinit()

259
MultiServer.py
View File

@@ -5,6 +5,7 @@ import functools
import json
import logging
import re
import shlex
import urllib.request
import websockets
import zlib
@@ -27,29 +28,18 @@ class Context:
self.data_filename = None
self.save_filename = None
self.disable_save = False
self.players = 0
self.player_names = {}
self.rom_names = {}
self.remote_items = set()
self.locations = {}
self.host = host
self.port = port
self.password = password
self.server = None
self.countdown_timer = 0
self.clients = []
self.received_items = {}
def get_room_info(ctx : Context):
return {
'password': ctx.password is not None,
'slots': ctx.players,
'players': [(client.name, client.team, client.slot) for client in ctx.clients if client.auth]
}
def same_name(lhs, rhs):
return lhs.lower() == rhs.lower()
def same_team(lhs, rhs):
return (type(lhs) is type(rhs)) and ((not lhs and not rhs) or (lhs.lower() == rhs.lower()))
async def send_msgs(websocket, msgs):
if not websocket or not websocket.open or websocket.closed:
return
@@ -65,21 +55,21 @@ def broadcast_all(ctx : Context, msgs):
def broadcast_team(ctx : Context, team, msgs):
for client in ctx.clients:
if client.auth and same_team(client.team, team):
if client.auth and client.team == team:
asyncio.create_task(send_msgs(client.socket, msgs))
def notify_all(ctx : Context, text):
print("Notice (all): %s" % text)
logging.info("Notice (all): %s" % text)
broadcast_all(ctx, [['Print', text]])
def notify_team(ctx : Context, team : str, text : str):
print("Team notice (%s): %s" % ("Default" if not team else team, text))
def notify_team(ctx : Context, team : int, text : str):
logging.info("Notice (Team #%d): %s" % (team+1, text))
broadcast_team(ctx, team, [['Print', text]])
def notify_client(client : Client, text : str):
if not client.auth:
return
print("Player notice (%s): %s" % (client.name, text))
logging.info("Notice (Player %s in team %d): %s" % (client.name, client.team+1, text))
asyncio.create_task(send_msgs(client.socket, [['Print', text]]))
async def server(websocket, path, ctx : Context):
@@ -105,54 +95,54 @@ async def server(websocket, path, ctx : Context):
ctx.clients.remove(client)
async def on_client_connected(ctx : Context, client : Client):
await send_msgs(client.socket, [['RoomInfo', get_room_info(ctx)]])
await send_msgs(client.socket, [['RoomInfo', {
'password': ctx.password is not None,
'players': [(client.team, client.slot, client.name) for client in ctx.clients if client.auth]
}]])
async def on_client_disconnected(ctx : Context, client : Client):
if client.auth:
await on_client_left(ctx, client)
async def on_client_joined(ctx : Context, client : Client):
notify_all(ctx, "%s has joined the game as player %d for %s" % (client.name, client.slot, "the default team" if not client.team else "team %s" % client.team))
notify_all(ctx, "%s (Team #%d) has joined the game" % (client.name, client.team + 1))
async def on_client_left(ctx : Context, client : Client):
notify_all(ctx, "%s (Player %d, %s) has left the game" % (client.name, client.slot, "Default team" if not client.team else "Team %s" % client.team))
notify_all(ctx, "%s (Team #%d) has left the game" % (client.name, client.team + 1))
async def countdown(ctx : Context, timer):
notify_all(ctx, f'[Server]: Starting countdown of {timer}s')
if ctx.countdown_timer:
ctx.countdown_timer = timer
return
ctx.countdown_timer = timer
while ctx.countdown_timer > 0:
notify_all(ctx, f'[Server]: {ctx.countdown_timer}')
ctx.countdown_timer -= 1
await asyncio.sleep(1)
notify_all(ctx, f'[Server]: GO')
def get_connected_players_string(ctx : Context):
auth_clients = [c for c in ctx.clients if c.auth]
if not auth_clients:
return 'No player connected'
auth_clients.sort(key=lambda c: ('' if not c.team else c.team.lower(), c.slot))
auth_clients.sort(key=lambda c: (c.team, c.slot))
current_team = 0
text = ''
text = 'Team #1: '
for c in auth_clients:
if c.team != current_team:
text += '::' + ('default team' if not c.team else c.team) + ':: '
text += f':: Team #{c.team + 1}: '
current_team = c.team
text += '%d:%s ' % (c.slot, c.name)
text += f'{c.name} '
return 'Connected players: ' + text[:-1]
def get_player_name_in_team(ctx : Context, team, slot):
for client in ctx.clients:
if client.auth and same_team(team, client.team) and client.slot == slot:
return client.name
return "Player %d" % slot
def get_client_from_name(ctx : Context, name):
for client in ctx.clients:
if client.auth and same_name(name, client.name):
return client
return None
def get_received_items(ctx : Context, team, player):
for (c_team, c_id), items in ctx.received_items.items():
if c_id == player and same_team(c_team, team):
return items
ctx.received_items[(team, player)] = []
return ctx.received_items[(team, player)]
return ctx.received_items.setdefault((team, player), [])
def tuplize_received_items(items):
return [(item.item, item.location, item.player_id, item.player_name) for item in items]
return [(item.item, item.location, item.player) for item in items]
def send_new_items(ctx : Context):
for client in ctx.clients:
@@ -163,37 +153,36 @@ def send_new_items(ctx : Context):
asyncio.create_task(send_msgs(client.socket, [['ReceivedItems', (client.send_index, tuplize_received_items(items)[client.send_index:])]]))
client.send_index = len(items)
def forfeit_player(ctx : Context, team, slot, name):
def forfeit_player(ctx : Context, team, slot):
all_locations = [values[0] for values in Regions.location_table.values() if type(values[0]) is int]
notify_all(ctx, "%s (Player %d) in team %s has forfeited" % (name, slot, team if team else 'default'))
register_location_checks(ctx, name, team, slot, all_locations)
notify_all(ctx, "%s (Team #%d) has forfeited" % (ctx.player_names[(team, slot)], team + 1))
register_location_checks(ctx, team, slot, all_locations)
def register_location_checks(ctx : Context, name, team, slot, locations):
def register_location_checks(ctx : Context, team, slot, locations):
found_items = False
for location in locations:
if (location, slot) in ctx.locations:
target_item, target_player = ctx.locations[(location, slot)]
if target_player != slot:
if target_player != slot or slot in ctx.remote_items:
found = False
recvd_items = get_received_items(ctx, team, target_player)
for recvd_item in recvd_items:
if recvd_item.location == location and recvd_item.player_id == slot:
if recvd_item.location == location and recvd_item.player == slot:
found = True
break
if not found:
new_item = ReceivedItem(target_item, location, slot, name)
new_item = ReceivedItem(target_item, location, slot)
recvd_items.append(new_item)
target_player_name = get_player_name_in_team(ctx, team, target_player)
broadcast_team(ctx, team, [['ItemSent', (name, target_player_name, target_item, location)]])
print('(%s) %s sent %s to %s (%s)' % (team if team else 'Team', name, get_item_name_from_id(target_item), target_player_name, get_location_name_from_address(location)))
if slot != target_player:
broadcast_team(ctx, team, [['ItemSent', (slot, location, target_player, target_item)]])
logging.info('(Team #%d) %s sent %s to %s (%s)' % (team+1, ctx.player_names[(team, slot)], get_item_name_from_id(target_item), ctx.player_names[(team, target_player)], get_location_name_from_address(location)))
found_items = True
send_new_items(ctx)
if found_items and not ctx.disable_save:
try:
with open(ctx.save_filename, "wb") as f:
jsonstr = json.dumps((ctx.players,
[(k, v) for k, v in ctx.rom_names.items()],
jsonstr = json.dumps((list(ctx.rom_names.items()),
[(k, [i.__dict__ for i in v]) for k, v in ctx.received_items.items()]))
f.write(zlib.compress(jsonstr.encode("utf-8")))
except Exception as e:
@@ -207,50 +196,30 @@ async def process_client_cmd(ctx : Context, client : Client, cmd, args):
if cmd == 'Connect':
if not args or type(args) is not dict or \
'password' not in args or type(args['password']) not in [str, type(None)] or \
'name' not in args or type(args['name']) is not str or \
'team' not in args or type(args['team']) not in [str, type(None)] or \
'slot' not in args or type(args['slot']) not in [int, type(None)]:
'rom' not in args or type(args['rom']) is not list:
await send_msgs(client.socket, [['InvalidArguments', 'Connect']])
return
errors = set()
if ctx.password is not None and ('password' not in args or args['password'] != ctx.password):
if ctx.password is not None and args['password'] != ctx.password:
errors.add('InvalidPassword')
if 'name' not in args or not args['name'] or not re.match(r'\w{1,10}', args['name']):
errors.add('InvalidName')
elif any([same_name(c.name, args['name']) for c in ctx.clients if c.auth]):
errors.add('NameAlreadyTaken')
if tuple(args['rom']) not in ctx.rom_names:
errors.add('InvalidRom')
else:
client.name = args['name']
if 'team' in args and args['team'] is not None and not re.match(r'\w{1,15}', args['team']):
errors.add('InvalidTeam')
else:
client.team = args['team'] if 'team' in args else None
if 'slot' in args and any([c.slot == args['slot'] for c in ctx.clients if c.auth and same_team(c.team, client.team)]):
errors.add('SlotAlreadyTaken')
elif 'slot' not in args or not args['slot']:
for slot in range(1, ctx.players + 1):
if slot not in [c.slot for c in ctx.clients if c.auth and same_team(c.team, client.team)]:
client.slot = slot
break
elif slot == ctx.players:
errors.add('SlotAlreadyTaken')
elif args['slot'] not in range(1, ctx.players + 1):
errors.add('InvalidSlot')
else:
client.slot = args['slot']
team, slot = ctx.rom_names[tuple(args['rom'])]
if any([c.slot == slot and c.team == team for c in ctx.clients if c.auth]):
errors.add('SlotAlreadyTaken')
else:
client.name = ctx.player_names[(team, slot)]
client.team = team
client.slot = slot
if errors:
client.name = None
client.team = None
client.slot = None
await send_msgs(client.socket, [['ConnectionRefused', list(errors)]])
else:
client.auth = True
reply = [['Connected', ctx.rom_names[client.slot]]]
reply = [['Connected', [(client.team, client.slot), [(p, n) for (t, p), n in ctx.player_names.items() if t == client.team]]]]
items = get_received_items(ctx, client.team, client.slot)
if items:
reply.append(['ReceivedItems', (0, tuplize_received_items(items))])
@@ -265,13 +234,35 @@ async def process_client_cmd(ctx : Context, client : Client, cmd, args):
items = get_received_items(ctx, client.team, client.slot)
if items:
client.send_index = len(items)
await send_msgs(client.socket, ['ReceivedItems', (0, tuplize_received_items(items))])
await send_msgs(client.socket, [['ReceivedItems', (0, tuplize_received_items(items))]])
if cmd == 'LocationChecks':
if type(args) is not list:
await send_msgs(client.socket, [['InvalidArguments', 'LocationChecks']])
return
register_location_checks(ctx, client.name, client.team, client.slot, args)
register_location_checks(ctx, client.team, client.slot, args)
if cmd == 'LocationScouts':
if type(args) is not list:
await send_msgs(client.socket, [['InvalidArguments', 'LocationScouts']])
return
locs = []
for location in args:
if type(location) is not int or 0 >= location > len(Regions.location_table):
await send_msgs(client.socket, [['InvalidArguments', 'LocationScouts']])
return
loc_name = list(Regions.location_table.keys())[location - 1]
target_item, target_player = ctx.locations[(Regions.location_table[loc_name][0], client.slot)]
replacements = {'SmallKey': 0xA2, 'BigKey': 0x9D, 'Compass': 0x8D, 'Map': 0x7D}
item_type = [i[2] for i in Items.item_table.values() if type(i[3]) is int and i[3] == target_item]
if item_type:
target_item = replacements.get(item_type[0], target_item)
locs.append([loc_name, location, target_item, target_player])
logging.info(f"{client.name} in team {client.team+1} scouted {', '.join([l[0] for l in locs])}")
await send_msgs(client.socket, [['LocationInfo', [l[1:] for l in locs]]])
if cmd == 'Say':
if type(args) is not str or not args.isprintable():
@@ -280,20 +271,26 @@ async def process_client_cmd(ctx : Context, client : Client, cmd, args):
notify_all(ctx, client.name + ': ' + args)
if args[:8] == '!players':
if args.startswith('!players'):
notify_all(ctx, get_connected_players_string(ctx))
if args[:8] == '!forfeit':
forfeit_player(ctx, client.team, client.slot, client.name)
if args.startswith('!forfeit'):
forfeit_player(ctx, client.team, client.slot)
if args.startswith('!countdown'):
try:
timer = int(args.split()[1])
except (IndexError, ValueError):
timer = 10
asyncio.create_task(countdown(ctx, timer))
def set_password(ctx : Context, password):
ctx.password = password
print('Password set to ' + password if password is not None else 'Password disabled')
logging.warning('Password set to ' + password if password is not None else 'Password disabled')
async def console(ctx : Context):
while True:
input = await aioconsole.ainput()
command = input.split()
command = shlex.split(input)
if not command:
continue
@@ -302,34 +299,41 @@ async def console(ctx : Context):
break
if command[0] == '/players':
print(get_connected_players_string(ctx))
logging.info(get_connected_players_string(ctx))
if command[0] == '/password':
set_password(ctx, command[1] if len(command) > 1 else None)
if command[0] == '/kick' and len(command) > 1:
client = get_client_from_name(ctx, command[1])
if client and client.socket and not client.socket.closed:
await client.socket.close()
team = int(command[2]) - 1 if len(command) > 2 and command[2].isdigit() else None
for client in ctx.clients:
if client.auth and client.name.lower() == command[1].lower() and (team is None or team == client.team):
if client.socket and not client.socket.closed:
await client.socket.close()
if command[0] == '/forfeitslot' and len(command) == 3 and command[2].isdigit():
team = command[1] if command[1] != 'default' else None
slot = int(command[2])
name = get_player_name_in_team(ctx, team, slot)
forfeit_player(ctx, team, slot, name)
if command[0] == '/forfeitslot' and len(command) > 1 and command[1].isdigit():
if len(command) > 2 and command[2].isdigit():
team = int(command[1]) - 1
slot = int(command[2])
else:
team = 0
slot = int(command[1])
forfeit_player(ctx, team, slot)
if command[0] == '/forfeitplayer' and len(command) > 1:
client = get_client_from_name(ctx, command[1])
if client:
forfeit_player(ctx, client.team, client.slot, client.name)
team = int(command[2]) - 1 if len(command) > 2 and command[2].isdigit() else None
for client in ctx.clients:
if client.auth and client.name.lower() == command[1].lower() and (team is None or team == client.team):
if client.socket and not client.socket.closed:
forfeit_player(ctx, client.team, client.slot)
if command[0] == '/senditem' and len(command) > 2:
[(player, item)] = re.findall(r'\S* (\S*) (.*)', input)
if item in Items.item_table:
client = get_client_from_name(ctx, player)
if client:
new_item = ReceivedItem(Items.item_table[item][3], "cheat console", 0, "server")
get_received_items(ctx, client.team, client.slot).append(new_item)
notify_all(ctx, 'Cheat console: sending "' + item + '" to ' + client.name)
for client in ctx.clients:
if client.auth and client.name.lower() == player.lower():
new_item = ReceivedItem(Items.item_table[item][3], "cheat console", client.slot)
get_received_items(ctx, client.team, client.slot).append(new_item)
notify_all(ctx, 'Cheat console: sending "' + item + '" to ' + client.name)
send_new_items(ctx)
else:
print("Unknown item: " + item)
logging.warning("Unknown item: " + item)
if command[0][0] != '/':
notify_all(ctx, '[Server]: ' + input)
@@ -342,8 +346,11 @@ async def main():
parser.add_argument('--multidata', default=None)
parser.add_argument('--savefile', default=None)
parser.add_argument('--disable_save', default=False, action='store_true')
parser.add_argument('--loglevel', default='info', choices=['debug', 'info', 'warning', 'error', 'critical'])
args = parser.parse_args()
logging.basicConfig(format='[%(asctime)s] %(message)s', level=getattr(logging, args.loglevel.upper(), logging.INFO))
ctx = Context(args.host, args.port, args.password)
ctx.data_filename = args.multidata
@@ -358,15 +365,18 @@ async def main():
with open(ctx.data_filename, 'rb') as f:
jsonobj = json.loads(zlib.decompress(f.read()).decode("utf-8"))
ctx.players = jsonobj[0]
ctx.rom_names = {k: v for k, v in jsonobj[1]}
ctx.locations = {tuple(k): tuple(v) for k, v in jsonobj[2]}
for team, names in enumerate(jsonobj['names']):
for player, name in enumerate(names, 1):
ctx.player_names[(team, player)] = name
ctx.rom_names = {tuple(rom): (team, slot) for slot, team, rom in jsonobj['roms']}
ctx.remote_items = set(jsonobj['remote_items'])
ctx.locations = {tuple(k): tuple(v) for k, v in jsonobj['locations']}
except Exception as e:
print('Failed to read multiworld data (%s)' % e)
logging.error('Failed to read multiworld data (%s)' % e)
return
ip = urllib.request.urlopen('https://v4.ident.me').read().decode('utf8') if not ctx.host else ctx.host
print('Hosting game of %d players (%s) at %s:%d' % (ctx.players, 'No password' if not ctx.password else 'Password: %s' % ctx.password, ip, ctx.port))
logging.info('Hosting game at %s:%d (%s)' % (ip, ctx.port, 'No password' if not ctx.password else 'Password: %s' % ctx.password))
ctx.disable_save = args.disable_save
if not ctx.disable_save:
@@ -375,17 +385,16 @@ async def main():
try:
with open(ctx.save_filename, 'rb') as f:
jsonobj = json.loads(zlib.decompress(f.read()).decode("utf-8"))
players = jsonobj[0]
rom_names = {k: v for k, v in jsonobj[1]}
received_items = {tuple(k): [ReceivedItem(**i) for i in v] for k, v in jsonobj[2]}
if players != ctx.players or rom_names != ctx.rom_names:
rom_names = jsonobj[0]
received_items = {tuple(k): [ReceivedItem(**i) for i in v] for k, v in jsonobj[1]}
if not all([ctx.rom_names[tuple(rom)] == (team, slot) for rom, (team, slot) in rom_names]):
raise Exception('Save file mismatch, will start a new game')
ctx.received_items = received_items
print('Loaded save file with %d received items for %d players' % (sum([len(p) for p in received_items.values()]), len(received_items)))
logging.info('Loaded save file with %d received items for %d players' % (sum([len(p) for p in received_items.values()]), len(received_items)))
except FileNotFoundError:
print('No save data found, starting a new game')
logging.error('No save data found, starting a new game')
except Exception as e:
print(e)
logging.info(e)
ctx.server = websockets.serve(functools.partial(server,ctx=ctx), ctx.host, ctx.port, ping_timeout=None, ping_interval=None)
await ctx.server

6
Mystery.py
View File

@@ -39,6 +39,7 @@ def main():
parser.add_argument('--seed', help='Define seed number to generate.', type=int)
parser.add_argument('--multi', default=1, type=lambda value: min(max(int(value), 1), 255))
parser.add_argument('--names', default='')
parser.add_argument('--teams', default=1, type=lambda value: max(int(value), 1))
parser.add_argument('--create_spoiler', action='store_true')
parser.add_argument('--rom')
parser.add_argument('--enemizercli')
@@ -146,13 +147,14 @@ def roll_settings(weights):
door_shuffle = get_choice('door_shuffle')
ret.door_shuffle = door_shuffle if door_shuffle != 'none' else 'vanilla'
goal = get_choice('goals')
ret.goal = {'ganon': 'ganon',
'fast_ganon': 'crystals',
'dungeons': 'dungeons',
'pedestal': 'pedestal',
'triforce-hunt': 'triforcehunt'
}[get_choice('goals')]
ret.openpyramid = ret.goal == 'fast_ganon'
}[goal]
ret.openpyramid = goal == 'fast_ganon'
ret.crystals_gt = get_choice('tower_open')

3
Plando.py
View File

@@ -24,6 +24,7 @@ def main(args):
# initialize the world
world = World(1, 'vanilla', 'noglitches', 'standard', 'normal', 'none', 'on', 'ganon', 'freshness', False, False, False, False, False, False, None, False)
world.player_names[1].append("Player 1")
logger = logging.getLogger('')
hasher = hashlib.md5()
@@ -69,7 +70,7 @@ def main(args):
logger.info('Patching ROM.')
rom = LocalRom(args.rom)
patch_rom(world, 1, rom, False)
patch_rom(world, rom, 1, 1, False)
apply_rom_settings(rom, args.heartbeep, args.heartcolor, args.quickswap, args.fastmenu, args.disablemusic, args.sprite, args.ow_palettes, args.uw_palettes)

101
Rom.py
View File

@@ -12,6 +12,7 @@ import subprocess
from BaseClasses import CollectionState, ShopType, Region, Location, Item, DoorType
from DoorShuffle import compass_data
from Dungeons import dungeon_music_addresses
from Regions import location_table
from Text import MultiByteTextMapper, CompressedTextMapper, text_addresses, Credits, TextTable
from Text import Uncle_texts, Ganon1_texts, TavernMan_texts, Sahasrahla2_texts, Triforce_texts, Blind_texts, BombShop2_texts, junk_texts
from Text import KingsReturn_texts, Sanctuary_texts, Kakariko_texts, Blacksmiths_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
@@ -26,8 +27,9 @@ JAP10HASH = '03a63945398191337e896e5771f77173'
class JsonRom(object):
def __init__(self):
self.name = None
def __init__(self, name=None, hash=None):
self.name = name
self.hash = hash
self.orig_buffer = None
self.patches = {}
self.addresses = []
@@ -72,8 +74,9 @@ class JsonRom(object):
class LocalRom(object):
def __init__(self, file, patch=True):
self.name = None
def __init__(self, file, patch=True, name=None, hash=None):
self.name = name
self.hash = hash
self.orig_buffer = None
with open(file, 'rb') as stream:
self.buffer = read_rom(stream)
@@ -92,6 +95,14 @@ class LocalRom(object):
with open(file, 'wb') as outfile:
outfile.write(self.buffer)
@staticmethod
def fromJsonRom(rom, file, rom_size = 0x200000):
ret = LocalRom(file, True, rom.name, rom.hash)
ret.buffer.extend(bytearray([0x00] * (rom_size - len(ret.buffer))))
for address, values in rom.patches.items():
ret.write_bytes(int(address), values)
return ret
def patch_base_rom(self):
# verify correct checksum of baserom
basemd5 = hashlib.md5()
@@ -116,11 +127,6 @@ class LocalRom(object):
if RANDOMIZERBASEHASH != patchedmd5.hexdigest():
raise RuntimeError('Provided Base Rom unsuitable for patching. Please provide a JAP(1.0) "Zelda no Densetsu - Kamigami no Triforce (Japan).sfc" rom to use as a base.')
def merge_enemizer_patches(self, patches):
self.buffer.extend(bytearray([0x00] * (0x400000 - len(self.buffer))))
for address, values in patches.items():
self.write_bytes(int(address), values)
def write_crc(self):
crc = (sum(self.buffer[:0x7FDC] + self.buffer[0x7FE0:]) + 0x01FE) & 0xFFFF
inv = crc ^ 0xFFFF
@@ -471,7 +477,7 @@ class Sprite(object):
# split into palettes of 15 colors
return array_chunk(palette_as_colors, 15)
def patch_rom(world, player, rom, enemized):
def patch_rom(world, rom, player, team, enemized):
random.seed(world.rom_seeds[player])
# progressive bow silver arrow hint hack
@@ -492,23 +498,27 @@ def patch_rom(world, player, rom, enemized):
continue
if not location.crystal:
# Keys in their native dungeon should use the orignal item code for keys
if location.parent_region.dungeon:
dungeon = location.parent_region.dungeon
if location.item is not None and dungeon.is_dungeon_item(location.item):
if location.item.bigkey:
itemid = 0x32
if location.item.smallkey:
itemid = 0x24
if location.item.map:
itemid = 0x33
if location.item.compass:
itemid = 0x25
if location.item and location.item.player != player:
if location.player_address is not None:
rom.write_byte(location.player_address, location.item.player)
else:
itemid = 0x5A
if location.item is not None:
# Keys in their native dungeon should use the orignal item code for keys
if location.parent_region.dungeon:
if location.parent_region.dungeon.is_dungeon_item(location.item):
if location.item.bigkey:
itemid = 0x32
if location.item.smallkey:
itemid = 0x24
if location.item.map:
itemid = 0x33
if location.item.compass:
itemid = 0x25
if world.remote_items[player]:
itemid = list(location_table.keys()).index(location.name) + 1
assert itemid < 0x100
rom.write_byte(location.player_address, 0xFF)
elif location.item.player != player:
if location.player_address is not None:
rom.write_byte(location.player_address, location.item.player)
else:
itemid = 0x5A
rom.write_byte(location.address, itemid)
else:
# crystals
@@ -1255,16 +1265,22 @@ def patch_rom(world, player, rom, enemized):
rom.write_byte(0xFED31, 0x2A) # preopen bombable exit
rom.write_byte(0xFEE41, 0x2A) # preopen bombable exit
write_strings(rom, world, player)
write_strings(rom, world, player, team)
rom.write_byte(0x18636C, 1 if world.remote_items[player] else 0)
# set rom name
# 21 bytes
from Main import __version__
# todo: change to DR when Enemizer is okay with DR
rom.name = bytearray('ER_{0}_{1:09}\0'.format(__version__[0:7], world.seed), 'utf8')
assert len(rom.name) <= 21
rom.name = bytearray(f'ER{__version__.split("-")[0].replace(".","")[0:3]}_{team+1}_{player}_{world.seed:09}\0', 'utf8')[:21]
rom.name.extend([0] * (21 - len(rom.name)))
rom.write_bytes(0x7FC0, rom.name)
# set player names
for p in range(1, min(world.players, 64) + 1):
rom.write_bytes(0x186380 + ((p - 1) * 32), hud_format_text(world.player_names[p][team]))
# Write title screen Code
hashint = int(rom.get_hash(), 16)
code = [
@@ -1275,6 +1291,7 @@ def patch_rom(world, player, rom, enemized):
hashint & 0x1F,
]
rom.write_bytes(0x180215, code)
rom.hash = code
return rom
@@ -1338,7 +1355,7 @@ def hud_format_text(text):
return output[:32]
def apply_rom_settings(rom, beep, color, quickswap, fastmenu, disable_music, sprite, ow_palettes, uw_palettes, names = None):
def apply_rom_settings(rom, beep, color, quickswap, fastmenu, disable_music, sprite, ow_palettes, uw_palettes):
if sprite and not isinstance(sprite, Sprite):
sprite = Sprite(sprite) if os.path.isfile(sprite) else get_sprite_from_name(sprite)
@@ -1407,11 +1424,6 @@ def apply_rom_settings(rom, beep, color, quickswap, fastmenu, disable_music, spr
elif uw_palettes == 'blackout':
blackout_uw_palettes(rom)
# set player names
for player, name in names.items():
if 0 < player <= 64:
rom.write_bytes(0x186380 + ((player - 1) * 32), hud_format_text(name))
if isinstance(rom, LocalRom):
rom.write_crc()
@@ -1548,12 +1560,15 @@ def blackout_uw_palettes(rom):
rom.write_bytes(i+44, [0] * 76)
rom.write_bytes(i+136, [0] * 44)
def get_hash_string(hash):
return ", ".join([hash_alphabet[code & 0x1F] for code in hash])
def write_string_to_rom(rom, target, string):
address, maxbytes = text_addresses[target]
rom.write_bytes(address, MultiByteTextMapper.convert(string, maxbytes))
def write_strings(rom, world, player):
def write_strings(rom, world, player, team):
tt = TextTable()
tt.removeUnwantedText()
@@ -1571,11 +1586,11 @@ def write_strings(rom, world, player):
hint = dest.hint_text if dest.hint_text else "something"
if dest.player != player:
if ped_hint:
hint += " for p%d!" % dest.player
hint += f" for {world.player_names[dest.player][team]}!"
elif type(dest) in [Region, Location]:
hint += " in p%d's world" % dest.player
hint += f" in {world.player_names[dest.player][team]}'s world"
else:
hint += " for p%d" % dest.player
hint += f" for {world.player_names[dest.player][team]}"
return hint
# For hints, first we write hints about entrances, some from the inconvenient list others from all reasonable entrances.
@@ -2366,3 +2381,9 @@ BigKeys = ['Big Key (Eastern Palace)',
'Big Key (Turtle Rock)',
'Big Key (Ganons Tower)'
]
hash_alphabet = [
"Bow", "Boomerang", "Hookshot", "Bomb", "Mushroom", "Powder", "Rod", "Pendant", "Bombos", "Ether", "Quake",
"Lamp", "Hammer", "Shovel", "Ocarina", "Bug Net", "Book", "Bottle", "Potion", "Cane", "Cape", "Mirror", "Boots",
"Gloves", "Flippers", "Pearl", "Shield", "Tunic", "Heart", "Map", "Compass", "Key"
]

15
Utils.py
View File

@@ -4,9 +4,6 @@ import re
import subprocess
import sys
def parse_names_string(names):
return {player: name for player, name in enumerate([n for n in re.split(r'[, ]', names) if n], 1)}
def int16_as_bytes(value):
value = value & 0xFFFF
return [value & 0xFF, (value >> 8) & 0xFF]
@@ -21,6 +18,18 @@ def pc_to_snes(value):
def snes_to_pc(value):
return ((value & 0x7F0000)>>1)|(value & 0x7FFF)
def parse_player_names(names, players, teams):
names = [n for n in re.split(r'[, ]', names) if n]
ret = []
while names or len(ret) < teams:
team = [n[:16] for n in names[:players]]
while len(team) != players:
team.append(f"Player {len(team) + 1}")
ret.append(team)
names = names[players:]
return ret
def is_bundled():
return getattr(sys, 'frozen', False)

3
requirements.txt Normal file
View File

@@ -0,0 +1,3 @@
aioconsole==0.1.15
colorama==0.4.3
websockets==8.1