RIP: MultiMystery and Mystery, now there's just Generate

Other changes:
host.yaml Multi Mystery options were moved and changed
generate_output now has an output_directory argument
MultiWorld.get_game_players(<game>) now replaces <game>_player_ids
Python venv should now work properly
This commit is contained in:
Fabian Dill
2021-07-21 18:08:15 +02:00
parent 47f7ec16c0
commit 2fc4006dfa
21 changed files with 305 additions and 645 deletions

345
Main.py
View File

@@ -6,6 +6,8 @@ import time
import zlib
import concurrent.futures
import pickle
import tempfile
import zipfile
from typing import Dict, Tuple
from BaseClasses import MultiWorld, CollectionState, Region, Item
@@ -128,7 +130,7 @@ def main(args, seed=None):
AutoWorld.call_all(world, "generate_early")
# system for sharing ER layouts
for player in world.alttp_player_ids:
for player in world.get_game_players("A Link to the Past"):
world.er_seeds[player] = str(world.random.randint(0, 2 ** 64))
if "-" in world.shuffle[player]:
@@ -160,7 +162,7 @@ def main(args, seed=None):
world.player_names[player].append(name)
logger.info('')
for player in world.alttp_player_ids:
for player in world.get_game_players("A Link to the Past"):
world.difficulty_requirements[player] = difficulties[world.difficulty[player]]
for player in world.player_ids:
@@ -168,7 +170,7 @@ def main(args, seed=None):
world.push_precollected(world.create_item(item_name, player))
for player in world.player_ids:
if player in world.alttp_player_ids:
if player in world.get_game_players("A Link to the Past"):
# enforce pre-defined local items.
if world.goal[player] in ["localtriforcehunt", "localganontriforcehunt"]:
world.local_items[player].add('Triforce Piece')
@@ -195,7 +197,7 @@ def main(args, seed=None):
AutoWorld.call_all(world, "create_regions")
for player in world.alttp_player_ids:
for player in world.get_game_players("A Link to the Past"):
if world.open_pyramid[player] == 'goal':
world.open_pyramid[player] = world.goal[player] in {'crystals', 'ganontriforcehunt',
'localganontriforcehunt', 'ganonpedestal'}
@@ -220,7 +222,7 @@ def main(args, seed=None):
logger.info('Shuffling the World about.')
for player in world.alttp_player_ids:
for player in world.get_game_players("A Link to the Past"):
if world.logic[player] not in ["noglitches", "minorglitches"] and world.shuffle[player] in \
{"vanilla", "dungeonssimple", "dungeonsfull", "simple", "restricted", "full"}:
world.fix_fake_world[player] = False
@@ -241,7 +243,7 @@ def main(args, seed=None):
logger.info('Generating Item Pool.')
for player in world.alttp_player_ids:
for player in world.get_game_players("A Link to the Past"):
generate_itempool(world, player)
logger.info('Calculating Access Rules.')
@@ -251,7 +253,7 @@ def main(args, seed=None):
AutoWorld.call_all(world, "set_rules")
for player in world.alttp_player_ids:
for player in world.get_game_players("A Link to the Past"):
set_rules(world, player)
for player in world.player_ids:
@@ -301,7 +303,7 @@ def main(args, seed=None):
outfilebase = 'AP_' + world.seed_name
rom_names = []
def _gen_rom(team: int, player: int):
def _gen_rom(team: int, player: int, output_directory:str):
use_enemizer = (world.boss_shuffle[player] != 'none' or world.enemy_shuffle[player]
or world.enemy_health[player] != 'default' or world.enemy_damage[player] != 'default'
or world.shufflepots[player] or world.bush_shuffle[player]
@@ -390,180 +392,184 @@ def main(args, seed=None):
"-prog_" + outfilestuffs["progressive"] if outfilestuffs["progressive"] in ['off', 'random'] else "", # A
"-nohints" if not outfilestuffs["hints"] == "True" else "") # B
) if not args.outputname else ''
rompath = output_path(f'{outfilebase}{outfilepname}{outfilesuffix}.sfc')
rompath = os.path.join(output_directory, f'{outfilebase}{outfilepname}{outfilesuffix}.sfc')
rom.write_to_file(rompath, hide_enemizer=True)
if args.create_diff:
Patch.create_patch_file(rompath, player=player, player_name=world.player_names[player][team])
Patch.create_patch_file(rompath, player=player, player_name=world.player_names[player][team])
os.unlink(rompath)
return player, team, bytes(rom.name)
pool = concurrent.futures.ThreadPoolExecutor()
check_accessibility_task = pool.submit(world.fulfills_accessibility)
output = tempfile.TemporaryDirectory()
with output as temp_dir:
check_accessibility_task = pool.submit(world.fulfills_accessibility)
rom_futures = []
output_file_futures = []
for team in range(world.teams):
for player in world.get_game_players("A Link to the Past"):
rom_futures.append(pool.submit(_gen_rom, team, player, temp_dir))
for player in world.player_ids:
output_file_futures.append(pool.submit(AutoWorld.call_single, world, "generate_output", player, temp_dir))
rom_futures = []
output_file_futures = []
for team in range(world.teams):
for player in world.alttp_player_ids:
rom_futures.append(pool.submit(_gen_rom, team, player))
for player in world.player_ids:
output_file_futures.append(pool.submit(AutoWorld.call_single, world, "generate_output", player))
def get_entrance_to_region(region: Region):
for entrance in region.entrances:
if entrance.parent_region.type in (RegionType.DarkWorld, RegionType.LightWorld, RegionType.Generic):
return entrance
for entrance in region.entrances: # BFS might be better here, trying DFS for now.
return get_entrance_to_region(entrance.parent_region)
def get_entrance_to_region(region: Region):
for entrance in region.entrances:
if entrance.parent_region.type in (RegionType.DarkWorld, RegionType.LightWorld, RegionType.Generic):
return entrance
for entrance in region.entrances: # BFS might be better here, trying DFS for now.
return get_entrance_to_region(entrance.parent_region)
# collect ER hint info
er_hint_data = {player: {} for player in world.get_game_players("A Link to the Past") if
world.shuffle[player] != "vanilla" or world.retro[player]}
from worlds.alttp.Regions import RegionType
for region in world.regions:
if region.player in er_hint_data and region.locations:
main_entrance = get_entrance_to_region(region)
for location in region.locations:
if type(location.address) == int: # skips events and crystals
if lookup_vanilla_location_to_entrance[location.address] != main_entrance.name:
er_hint_data[region.player][location.address] = main_entrance.name
# collect ER hint info
er_hint_data = {player: {} for player in world.alttp_player_ids if
world.shuffle[player] != "vanilla" or world.retro[player]}
from worlds.alttp.Regions import RegionType
for region in world.regions:
if region.player in er_hint_data and region.locations:
main_entrance = get_entrance_to_region(region)
for location in region.locations:
if type(location.address) == int: # skips events and crystals
if lookup_vanilla_location_to_entrance[location.address] != main_entrance.name:
er_hint_data[region.player][location.address] = main_entrance.name
ordered_areas = ('Light World', 'Dark World', 'Hyrule Castle', 'Agahnims Tower', 'Eastern Palace', 'Desert Palace',
'Tower of Hera', 'Palace of Darkness', 'Swamp Palace', 'Skull Woods', 'Thieves Town', 'Ice Palace',
'Misery Mire', 'Turtle Rock', 'Ganons Tower', "Total")
ordered_areas = ('Light World', 'Dark World', 'Hyrule Castle', 'Agahnims Tower', 'Eastern Palace', 'Desert Palace',
'Tower of Hera', 'Palace of Darkness', 'Swamp Palace', 'Skull Woods', 'Thieves Town', 'Ice Palace',
'Misery Mire', 'Turtle Rock', 'Ganons Tower', "Total")
checks_in_area = {player: {area: list() for area in ordered_areas}
for player in range(1, world.players + 1)}
checks_in_area = {player: {area: list() for area in ordered_areas}
for player in range(1, world.players + 1)}
for player in range(1, world.players + 1):
checks_in_area[player]["Total"] = 0
for player in range(1, world.players + 1):
checks_in_area[player]["Total"] = 0
for location in [loc for loc in world.get_filled_locations() if type(loc.address) is int]:
main_entrance = get_entrance_to_region(location.parent_region)
if location.game != "A Link to the Past":
checks_in_area[location.player]["Light World"].append(location.address)
elif location.parent_region.dungeon:
dungeonname = {'Inverted Agahnims Tower': 'Agahnims Tower',
'Inverted Ganons Tower': 'Ganons Tower'} \
.get(location.parent_region.dungeon.name, location.parent_region.dungeon.name)
checks_in_area[location.player][dungeonname].append(location.address)
elif main_entrance.parent_region.type == RegionType.LightWorld:
checks_in_area[location.player]["Light World"].append(location.address)
elif main_entrance.parent_region.type == RegionType.DarkWorld:
checks_in_area[location.player]["Dark World"].append(location.address)
checks_in_area[location.player]["Total"] += 1
for location in [loc for loc in world.get_filled_locations() if type(loc.address) is int]:
main_entrance = get_entrance_to_region(location.parent_region)
if location.game != "A Link to the Past":
checks_in_area[location.player]["Light World"].append(location.address)
elif location.parent_region.dungeon:
dungeonname = {'Inverted Agahnims Tower': 'Agahnims Tower',
'Inverted Ganons Tower': 'Ganons Tower'} \
.get(location.parent_region.dungeon.name, location.parent_region.dungeon.name)
checks_in_area[location.player][dungeonname].append(location.address)
elif main_entrance.parent_region.type == RegionType.LightWorld:
checks_in_area[location.player]["Light World"].append(location.address)
elif main_entrance.parent_region.type == RegionType.DarkWorld:
checks_in_area[location.player]["Dark World"].append(location.address)
checks_in_area[location.player]["Total"] += 1
oldmancaves = []
takeanyregions = ["Old Man Sword Cave", "Take-Any #1", "Take-Any #2", "Take-Any #3", "Take-Any #4"]
for index, take_any in enumerate(takeanyregions):
for region in [world.get_region(take_any, player) for player in range(1, world.players + 1) if
world.retro[player]]:
item = world.create_item(region.shop.inventory[(0 if take_any == "Old Man Sword Cave" else 1)]['item'],
region.player)
player = region.player
location_id = SHOP_ID_START + total_shop_slots + index
oldmancaves = []
takeanyregions = ["Old Man Sword Cave", "Take-Any #1", "Take-Any #2", "Take-Any #3", "Take-Any #4"]
for index, take_any in enumerate(takeanyregions):
for region in [world.get_region(take_any, player) for player in range(1, world.players + 1) if
world.retro[player]]:
item = world.create_item(region.shop.inventory[(0 if take_any == "Old Man Sword Cave" else 1)]['item'],
region.player)
player = region.player
location_id = SHOP_ID_START + total_shop_slots + index
main_entrance = get_entrance_to_region(region)
if main_entrance.parent_region.type == RegionType.LightWorld:
checks_in_area[player]["Light World"].append(location_id)
else:
checks_in_area[player]["Dark World"].append(location_id)
checks_in_area[player]["Total"] += 1
main_entrance = get_entrance_to_region(region)
if main_entrance.parent_region.type == RegionType.LightWorld:
checks_in_area[player]["Light World"].append(location_id)
er_hint_data[player][location_id] = main_entrance.name
oldmancaves.append(((location_id, player), (item.code, player)))
FillDisabledShopSlots(world)
def write_multidata(roms, outputs):
import base64
import NetUtils
for future in roms:
rom_name = future.result()
rom_names.append(rom_name)
slot_data = {}
client_versions = {}
minimum_versions = {"server": (0, 1, 1), "clients": client_versions}
games = {}
for slot in world.player_ids:
client_versions[slot] = world.worlds[slot].get_required_client_version()
games[slot] = world.game[slot]
connect_names = {base64.b64encode(rom_name).decode(): (team, slot) for
slot, team, rom_name in rom_names}
precollected_items = {player: [] for player in range(1, world.players + 1)}
for item in world.precollected_items:
precollected_items[item.player].append(item.code)
precollected_hints = {player: set() for player in range(1, world.players + 1)}
# for now special case Factorio tech_tree_information
sending_visible_players = set()
for player in world.get_game_players("Factorio"):
if world.tech_tree_information[player].value == 2:
sending_visible_players.add(player)
for i, team in enumerate(parsed_names):
for player, name in enumerate(team, 1):
if player not in world.get_game_players("A Link to the Past"):
connect_names[name] = (i, player)
for slot in world.player_ids:
slot_data[slot] = world.worlds[slot].fill_slot_data()
locations_data: Dict[int, Dict[int, Tuple[int, int]]] = {player: {} for player in world.player_ids}
for location in world.get_filled_locations():
if type(location.address) == int:
locations_data[location.player][location.address] = location.item.code, location.item.player
if location.player in sending_visible_players and location.item.player != location.player:
hint = NetUtils.Hint(location.item.player, location.player, location.address,
location.item.code, False)
precollected_hints[location.player].add(hint)
precollected_hints[location.item.player].add(hint)
elif location.item.name in args.start_hints[location.item.player]:
hint = NetUtils.Hint(location.item.player, location.player, location.address,
location.item.code, False,
er_hint_data.get(location.player, {}).get(location.address, ""))
precollected_hints[location.player].add(hint)
precollected_hints[location.item.player].add(hint)
multidata = zlib.compress(pickle.dumps({
"slot_data": slot_data,
"games": games,
"names": parsed_names,
"connect_names": connect_names,
"remote_items": {player for player in world.player_ids if
world.worlds[player].remote_items},
"locations": locations_data,
"checks_in_area": checks_in_area,
"server_options": get_options()["server_options"],
"er_hint_data": er_hint_data,
"precollected_items": precollected_items,
"precollected_hints": precollected_hints,
"version": tuple(version_tuple),
"tags": ["AP"],
"minimum_versions": minimum_versions,
"seed_name": world.seed_name
}), 9)
with open(os.path.join(temp_dir, '%s.archipelago' % outfilebase), 'wb') as f:
f.write(bytes([1])) # version of format
f.write(multidata)
for future in outputs:
future.result() # collect errors if they occured
multidata_task = pool.submit(write_multidata, rom_futures, output_file_futures)
if not check_accessibility_task.result():
if not world.can_beat_game():
raise Exception("Game appears as unbeatable. Aborting.")
else:
checks_in_area[player]["Dark World"].append(location_id)
checks_in_area[player]["Total"] += 1
er_hint_data[player][location_id] = main_entrance.name
oldmancaves.append(((location_id, player), (item.code, player)))
FillDisabledShopSlots(world)
def write_multidata(roms, outputs):
import base64
import NetUtils
for future in roms:
rom_name = future.result()
rom_names.append(rom_name)
slot_data = {}
client_versions = {}
minimum_versions = {"server": (0, 1, 1), "clients": client_versions}
games = {}
for slot in world.player_ids:
client_versions[slot] = world.worlds[slot].get_required_client_version()
games[slot] = world.game[slot]
connect_names = {base64.b64encode(rom_name).decode(): (team, slot) for
slot, team, rom_name in rom_names}
precollected_items = {player: [] for player in range(1, world.players + 1)}
for item in world.precollected_items:
precollected_items[item.player].append(item.code)
precollected_hints = {player: set() for player in range(1, world.players + 1)}
# for now special case Factorio tech_tree_information
sending_visible_players = set()
for player in world.factorio_player_ids:
if world.tech_tree_information[player].value == 2:
sending_visible_players.add(player)
for i, team in enumerate(parsed_names):
for player, name in enumerate(team, 1):
if player not in world.alttp_player_ids:
connect_names[name] = (i, player)
if world.hk_player_ids:
for slot in world.hk_player_ids:
slot_data[slot] = AutoWorld.call_single(world, "fill_slot_data", slot)
for slot in world.minecraft_player_ids:
slot_data[slot] = AutoWorld.call_single(world, "fill_slot_data", slot)
locations_data: Dict[int, Dict[int, Tuple[int, int]]] = {player: {} for player in world.player_ids}
for location in world.get_filled_locations():
if type(location.address) == int:
locations_data[location.player][location.address] = (location.item.code, location.item.player)
if location.player in sending_visible_players and location.item.player != location.player:
hint = NetUtils.Hint(location.item.player, location.player, location.address,
location.item.code, False)
precollected_hints[location.player].add(hint)
precollected_hints[location.item.player].add(hint)
elif location.item.name in args.start_hints[location.item.player]:
hint = NetUtils.Hint(location.item.player, location.player, location.address,
location.item.code, False,
er_hint_data.get(location.player, {}).get(location.address, ""))
precollected_hints[location.player].add(hint)
precollected_hints[location.item.player].add(hint)
multidata = zlib.compress(pickle.dumps({
"slot_data": slot_data,
"games": games,
"names": parsed_names,
"connect_names": connect_names,
"remote_items": {player for player in world.player_ids if
world.worlds[player].remote_items},
"locations": locations_data,
"checks_in_area": checks_in_area,
"server_options": get_options()["server_options"],
"er_hint_data": er_hint_data,
"precollected_items": precollected_items,
"precollected_hints": precollected_hints,
"version": tuple(version_tuple),
"tags": ["AP"],
"minimum_versions": minimum_versions,
"seed_name": world.seed_name
}), 9)
with open(output_path('%s.archipelago' % outfilebase), 'wb') as f:
f.write(bytes([1])) # version of format
f.write(multidata)
for future in outputs:
future.result() # collect errors if they occured
multidata_task = pool.submit(write_multidata, rom_futures, output_file_futures)
if not check_accessibility_task.result():
if not world.can_beat_game():
raise Exception("Game appears as unbeatable. Aborting.")
else:
logger.warning("Location Accessibility requirements not fulfilled.")
if multidata_task:
multidata_task.result() # retrieve exception if one exists
pool.shutdown() # wait for all queued tasks to complete
if not args.skip_playthrough:
logger.info('Calculating playthrough.')
create_playthrough(world)
if args.create_spoiler: # needs spoiler.hashes to be filled, that depend on rom_futures being done
world.spoiler.to_file(output_path('%s_Spoiler.txt' % outfilebase))
logger.warning("Location Accessibility requirements not fulfilled.")
if multidata_task:
multidata_task.result() # retrieve exception if one exists
pool.shutdown() # wait for all queued tasks to complete
if not args.skip_playthrough:
logger.info('Calculating playthrough.')
create_playthrough(world)
if args.create_spoiler: # needs spoiler.hashes to be filled, that depend on rom_futures being done
world.spoiler.to_file(os.path.join(temp_dir, '%s_Spoiler.txt' % outfilebase))
logger.info('Creating final archive.')
with zipfile.ZipFile(output_path(f"AP_{world.seed_name}.zip"), mode="w", compression=zipfile.ZIP_LZMA,
compresslevel=9) as zf:
for file in os.scandir(temp_dir):
zf.write(os.path.join(temp_dir, file), arcname=file.name)
logger.info('Done. Enjoy. Total Time: %s', time.perf_counter() - start)
return world
@@ -581,7 +587,8 @@ def create_playthrough(world):
while sphere_candidates:
state.sweep_for_events(key_only=True)
# build up spheres of collection radius. Everything in each sphere is independent from each other in dependencies and only depends on lower spheres
# build up spheres of collection radius.
# Everything in each sphere is independent from each other in dependencies and only depends on lower spheres
sphere = {location for location in sphere_candidates if state.can_reach(location)}
@@ -682,7 +689,7 @@ def create_playthrough(world):
world.spoiler.paths.update(
{str(location): get_path(state, location.parent_region) for sphere in collection_spheres for location in
sphere if location.player == player})
if player in world.alttp_player_ids:
if player in world.get_game_players("A Link to the Past"):
for path in dict(world.spoiler.paths).values():
if any(exit == 'Pyramid Fairy' for (_, exit) in path):
if world.mode[player] != 'inverted':