import random
import logging
from BaseClasses import CollectionState
class FillError(RuntimeError):
pass
def distribute_items_cutoff(world, cutoffrate=0.33):
# get list of locations to fill in
fill_locations = world.get_unfilled_locations()
random.shuffle(fill_locations)
# get items to distribute
random.shuffle(world.itempool)
itempool = world.itempool
total_advancement_items = len([item for item in itempool if item.advancement])
placed_advancement_items = 0
progress_done = False
advancement_placed = False
# sweep once to pick up preplaced items
world.state.sweep_for_events()
while itempool and fill_locations:
candidate_item_to_place = None
item_to_place = None
for item in itempool:
if advancement_placed or (progress_done and (item.advancement or item.priority)):
item_to_place = item
break
if item.advancement:
candidate_item_to_place = item
if world.unlocks_new_location(item):
item_to_place = item
placed_advancement_items += 1
break
if item_to_place is None:
# check if we can reach all locations and that is why we find no new locations to place
if not progress_done and len(world.get_reachable_locations()) == len(world.get_locations()):
progress_done = True
continue
# check if we have now placed all advancement items
if progress_done:
advancement_placed = True
continue
# we might be in a situation where all new locations require multiple items to reach. If that is the case, just place any advancement item we've found and continue trying
if candidate_item_to_place is not None:
item_to_place = candidate_item_to_place
placed_advancement_items += 1
else:
# we placed all available progress items. Maybe the game can be beaten anyway?
if world.can_beat_game():
logging.getLogger('').warning('Not all locations reachable. Game beatable anyway.')
progress_done = True
continue
raise FillError('No more progress items left to place.')
spot_to_fill = None
for location in fill_locations if placed_advancement_items / total_advancement_items < cutoffrate else reversed(fill_locations):
if location.can_fill(world.state, item_to_place):
spot_to_fill = location
break
if spot_to_fill is None:
# we filled all reachable spots. Maybe the game can be beaten anyway?
if world.can_beat_game():
logging.getLogger('').warning('Not all items placed. Game beatable anyway.')
break
raise FillError('No more spots to place %s' % item_to_place)
world.push_item(spot_to_fill, item_to_place, True)
itempool.remove(item_to_place)
fill_locations.remove(spot_to_fill)
logging.getLogger('').debug('Unplaced items: %s - Unfilled Locations: %s', [item.name for item in itempool], [location.name for location in fill_locations])
def distribute_items_staleness(world):
# get list of locations to fill in
fill_locations = world.get_unfilled_locations()
random.shuffle(fill_locations)
# get items to distribute
random.shuffle(world.itempool)
itempool = world.itempool
progress_done = False
advancement_placed = False
# sweep once to pick up preplaced items
world.state.sweep_for_events()
while itempool and fill_locations:
candidate_item_to_place = None
item_to_place = None
for item in itempool:
if advancement_placed or (progress_done and (item.advancement or item.priority)):
item_to_place = item
break
if item.advancement:
candidate_item_to_place = item
if world.unlocks_new_location(item):
item_to_place = item
break
if item_to_place is None:
# check if we can reach all locations and that is why we find no new locations to place
if not progress_done and len(world.get_reachable_locations()) == len(world.get_locations()):
progress_done = True
continue
# check if we have now placed all advancement items
if progress_done:
advancement_placed = True
continue
# we might be in a situation where all new locations require multiple items to reach. If that is the case, just place any advancement item we've found and continue trying
if candidate_item_to_place is not None:
item_to_place = candidate_item_to_place
else:
# we placed all available progress items. Maybe the game can be beaten anyway?
if world.can_beat_game():
logging.getLogger('').warning('Not all locations reachable. Game beatable anyway.')
progress_done = True
continue
raise FillError('No more progress items left to place.')
spot_to_fill = None
for location in fill_locations:
# increase likelyhood of skipping a location if it has been found stale
if not progress_done and random.randint(0, location.staleness_count) > 2:
continue
if location.can_fill(world.state, item_to_place):
spot_to_fill = location
break
else:
location.staleness_count += 1
# might have skipped too many locations due to potential staleness. Do not check for staleness now to find a candidate
if spot_to_fill is None:
for location in fill_locations:
if location.can_fill(world.state, item_to_place):
spot_to_fill = location
break
if spot_to_fill is None:
# we filled all reachable spots. Maybe the game can be beaten anyway?
if world.can_beat_game():
logging.getLogger('').warning('Not all items placed. Game beatable anyway.')
break
raise FillError('No more spots to place %s' % item_to_place)
world.push_item(spot_to_fill, item_to_place, True)
itempool.remove(item_to_place)
fill_locations.remove(spot_to_fill)
logging.getLogger('').debug('Unplaced items: %s - Unfilled Locations: %s', [item.name for item in itempool], [location.name for location in fill_locations])
def fill_restrictive(world, base_state: CollectionState, locations, itempool, single_player_placement=False):
def sweep_from_pool():
new_state = base_state.copy()
for item in itempool:
new_state.collect(item, True)
new_state.sweep_for_events()
return new_state
unplaced_items = []
no_access_checks = {}
reachable_items = {}
for item in itempool:
if world.accessibility[item.player] == 'none':
no_access_checks.setdefault(item.player, []).append(item)
else:
reachable_items.setdefault(item.player, []).append(item)
for player_items in [no_access_checks, reachable_items]:
while any(player_items.values()) and locations:
items_to_place = [[itempool.remove(items[-1]), items.pop()][-1] for items in player_items.values() if items]
maximum_exploration_state = sweep_from_pool()
has_beaten_game = world.has_beaten_game(maximum_exploration_state)
for item_to_place in items_to_place:
perform_access_check = True
if world.accessibility[item_to_place.player] == 'none':
perform_access_check = not world.has_beaten_game(maximum_exploration_state, item_to_place.player) if single_player_placement else not has_beaten_game
for location in locations:
if (not single_player_placement or location.player == item_to_place.player)\
and location.can_fill(maximum_exploration_state, item_to_place, perform_access_check):
spot_to_fill = location
break
else:
# we filled all reachable spots. Maybe the game can be beaten anyway?
unplaced_items.insert(0, item_to_place)
if world.accessibility[item_to_place.player] != 'none' and world.can_beat_game():
logging.getLogger('').warning(
'Not all items placed. Game beatable anyway. (Could not place %s)' % item_to_place)
continue
raise FillError('No more spots to place %s' % item_to_place)
world.push_item(spot_to_fill, item_to_place, False)
locations.remove(spot_to_fill)
spot_to_fill.event = True
itempool.extend(unplaced_items)
def distribute_items_restrictive(world, gftower_trash=False, fill_locations=None):
# If not passed in, then get a shuffled list of locations to fill in
if not fill_locations:
fill_locations = world.get_unfilled_locations()
random.shuffle(fill_locations)
# get items to distribute
random.shuffle(world.itempool)
progitempool = []
localprioitempool = {player: [] for player in range(1, world.players + 1)}
localrestitempool = {player: [] for player in range(1, world.players + 1)}
prioitempool = []
restitempool = []
for item in world.itempool:
if item.advancement:
progitempool.append(item)
elif item.name in world.local_items[item.player]:
if item.priority:
localprioitempool[item.player].append(item)
else:
localrestitempool[item.player].append(item)
elif item.priority:
prioitempool.append(item)
else:
restitempool.append(item)
# fill in gtower locations with trash first
for player in range(1, world.players + 1):
if not gftower_trash or not world.ganonstower_vanilla[player] or world.logic[player] == 'owglitches':
continue
gftower_trash_count = (
random.randint(15, 50) if 'triforcehunt' in world.goal[player]
else random.randint(0, 15))
gtower_locations = [location for location in fill_locations if
'Ganons Tower' in location.name and location.player == player]
random.shuffle(gtower_locations)
trashcnt = 0
localrest = localrestitempool[player]
if localrest:
gt_item_pool = restitempool + localrest
random.shuffle(gt_item_pool)
else:
gt_item_pool = restitempool.copy()
while gtower_locations and gt_item_pool and trashcnt < gftower_trash_count:
spot_to_fill = gtower_locations.pop()
item_to_place = gt_item_pool.pop()
if item_to_place in localrest:
localrest.remove(item_to_place)
else:
restitempool.remove(item_to_place)
world.push_item(spot_to_fill, item_to_place, False)
fill_locations.remove(spot_to_fill)
trashcnt += 1
random.shuffle(fill_locations)
fill_locations.reverse()
# Make sure the escape small key is placed first in standard with key shuffle to prevent running out of spots
progitempool.sort(
key=lambda item: 1 if item.name == 'Small Key (Hyrule Castle)' and world.mode[item.player] == 'standard' and
world.keyshuffle[item.player] else 0)
fill_restrictive(world, world.state, fill_locations, progitempool)
if any(
localprioitempool.values() or localrestitempool.values()): # we need to make sure some fills are limited to certain worlds
for player, items in localprioitempool.items(): # already shuffled
local_locations = [location for location in fill_locations if location.player == player]
random.shuffle(local_locations)
for item_to_place in items:
spot_to_fill = local_locations.pop()
world.push_item(spot_to_fill, item_to_place, False)
fill_locations.remove(spot_to_fill)
for player, items in localrestitempool.items(): # already shuffled
local_locations = [location for location in fill_locations if location.player == player]
random.shuffle(local_locations)
for item_to_place in items:
spot_to_fill = local_locations.pop()
world.push_item(spot_to_fill, item_to_place, False)
fill_locations.remove(spot_to_fill)
random.shuffle(fill_locations)
fast_fill(world, prioitempool, fill_locations)
fast_fill(world, restitempool, fill_locations)
logging.getLogger('').debug('Unplaced items: %s - Unfilled Locations: %s',
[item.name for item in progitempool + prioitempool + restitempool],
[location.name for location in fill_locations])
def fast_fill(world, item_pool, fill_locations):
while item_pool and fill_locations:
world.push_item(fill_locations.pop(), item_pool.pop(), False)
def flood_items(world):
# get items to distribute
random.shuffle(world.itempool)
itempool = world.itempool
progress_done = False
# sweep once to pick up preplaced items
world.state.sweep_for_events()
# fill world from top of itempool while we can
while not progress_done:
location_list = world.get_unfilled_locations()
random.shuffle(location_list)
spot_to_fill = None
for location in location_list:
if location.can_fill(world.state, itempool[0]):
spot_to_fill = location
break
if spot_to_fill:
item = itempool.pop(0)
world.push_item(spot_to_fill, item, True)
continue
# ran out of spots, check if we need to step in and correct things
if len(world.get_reachable_locations()) == len(world.get_locations()):
progress_done = True
continue
# need to place a progress item instead of an already placed item, find candidate
item_to_place = None
candidate_item_to_place = None
for item in itempool:
if item.advancement:
candidate_item_to_place = item
if world.unlocks_new_location(item):
item_to_place = item
break
# we might be in a situation where all new locations require multiple items to reach. If that is the case, just place any advancement item we've found and continue trying
if item_to_place is None:
if candidate_item_to_place is not None:
item_to_place = candidate_item_to_place
else:
raise FillError('No more progress items left to place.')
# find item to replace with progress item
location_list = world.get_reachable_locations()
random.shuffle(location_list)
for location in location_list:
if location.item is not None and not location.item.advancement and not location.item.priority and not location.item.smallkey and not location.item.bigkey:
# safe to replace
replace_item = location.item
replace_item.location = None
itempool.append(replace_item)
world.push_item(location, item_to_place, True)
itempool.remove(item_to_place)
break
def balance_multiworld_progression(world):
balanceable_players = {player for player in range(1, world.players + 1) if world.progression_balancing[player]}
if not balanceable_players:
logging.info('Skipping multiworld progression balancing.')
else:
logging.info(f'Balancing multiworld progression for Players {balanceable_players}.')
state = CollectionState(world)
checked_locations = []
unchecked_locations = world.get_locations().copy()
random.shuffle(unchecked_locations)
reachable_locations_count = {player: 0 for player in range(1, world.players + 1)}
def get_sphere_locations(sphere_state, locations):
sphere_state.sweep_for_events(key_only=True, locations=locations)
return [loc for loc in locations if sphere_state.can_reach(loc)]
while True:
sphere_locations = get_sphere_locations(state, unchecked_locations)
for location in sphere_locations:
unchecked_locations.remove(location)
reachable_locations_count[location.player] += 1
if checked_locations:
threshold = max(reachable_locations_count.values()) - 20
balancing_players = [player for player, reachables in reachable_locations_count.items() if
reachables < threshold and player in balanceable_players]
if balancing_players:
balancing_state = state.copy()
balancing_unchecked_locations = unchecked_locations.copy()
balancing_reachables = reachable_locations_count.copy()
balancing_sphere = sphere_locations.copy()
candidate_items = []
while True:
for location in balancing_sphere:
if location.event and (
world.keyshuffle[location.item.player] or not location.item.smallkey) and (
world.bigkeyshuffle[location.item.player] or not location.item.bigkey):
balancing_state.collect(location.item, True, location)
if location.item.player in balancing_players and not location.locked:
candidate_items.append(location)
balancing_sphere = get_sphere_locations(balancing_state, balancing_unchecked_locations)
for location in balancing_sphere:
balancing_unchecked_locations.remove(location)
balancing_reachables[location.player] += 1
if world.has_beaten_game(balancing_state) or all(
[reachables >= threshold for reachables in balancing_reachables.values()]):
break
elif not balancing_sphere:
raise RuntimeError('Not all required items reachable. Something went terribly wrong here.')
unlocked_locations = [l for l in unchecked_locations if l not in balancing_unchecked_locations]
items_to_replace = []
for player in balancing_players:
locations_to_test = [l for l in unlocked_locations if l.player == player]
# only replace items that end up in another player's world
items_to_test = [l for l in candidate_items if l.item.player == player and l.player != player]
while items_to_test:
testing = items_to_test.pop()
reducing_state = state.copy()
for location in [*[l for l in items_to_replace if l.item.player == player], *items_to_test]:
reducing_state.collect(location.item, True, location)
reducing_state.sweep_for_events(locations=locations_to_test)
if world.has_beaten_game(balancing_state):
if not world.has_beaten_game(reducing_state):
items_to_replace.append(testing)
else:
reduced_sphere = get_sphere_locations(reducing_state, locations_to_test)
if reachable_locations_count[player] + len(reduced_sphere) < threshold:
items_to_replace.append(testing)
replaced_items = False
replacement_locations = [l for l in checked_locations if not l.event and not l.locked]
while replacement_locations and items_to_replace:
new_location = replacement_locations.pop()
old_location = items_to_replace.pop()
while not new_location.can_fill(state, old_location.item, False) or (
new_location.item and not old_location.can_fill(state, new_location.item, False)):
replacement_locations.insert(0, new_location)
new_location = replacement_locations.pop()
new_location.item, old_location.item = old_location.item, new_location.item
new_location.event, old_location.event = True, False
logging.debug(f"Progression balancing moved {new_location.item} to {new_location}, "
f"displacing {old_location.item} in {old_location}")
state.collect(new_location.item, True, new_location)
replaced_items = True
if replaced_items:
for location in get_sphere_locations(state, [l for l in unlocked_locations if
l.player in balancing_players]):
unchecked_locations.remove(location)
reachable_locations_count[location.player] += 1
sphere_locations.append(location)
for location in sphere_locations:
if location.event and (world.keyshuffle[location.item.player] or not location.item.smallkey) and (
world.bigkeyshuffle[location.item.player] or not location.item.bigkey):
state.collect(location.item, True, location)
checked_locations.extend(sphere_locations)
if world.has_beaten_game(state):
break
elif not sphere_locations:
raise RuntimeError('Not all required items reachable. Something went terribly wrong here.')