361 lines
14 KiB
Python
361 lines
14 KiB
Python
# OpenMFOR
|
|
# credits manually reinstated due to the comments being lost from the object code decompilation
|
|
|
|
# Original release is Copyright (C) 2022 Kazuto88
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# Source Generated with Decompyle++
|
|
# File: Fusion_Graph.pyc (Python 3.8)
|
|
|
|
import struct
|
|
import sys
|
|
import time
|
|
|
|
class Game:
|
|
|
|
def __init__(self, vanillaGame, randoSettings=None):
|
|
self.graph = dict()
|
|
self.areaConnections = dict()
|
|
self.areaConnectionOffsets = dict()
|
|
self.doorConnections = dict()
|
|
self.rooms = dict()
|
|
self.requirements = dict()
|
|
self.visited = list()
|
|
self.queue = list()
|
|
self.majorItemLocations = list()
|
|
self.minorItemLocations = list()
|
|
self.itemLocations = set()
|
|
self.patcher = dict()
|
|
self.oldtime = 0.0
|
|
self.bfstime = 0.0
|
|
self.dfstime = 0.0
|
|
|
|
# print('DEBUG: Opening ROM to pick stuff')
|
|
|
|
try:
|
|
with open(vanillaGame, 'rb') as sourceRom:
|
|
sourceRom.seek(3967888, 0)
|
|
sourceArea = int.from_bytes(sourceRom.read(1), 'little')
|
|
sourceDoor = int.from_bytes(sourceRom.read(1), 'little')
|
|
targetOffset = sourceRom.tell()
|
|
targetArea = int.from_bytes(sourceRom.read(1), 'little')
|
|
while sourceArea != 255:
|
|
self.areaConnections.update({
|
|
'S{}-{:02X}'.format(sourceArea, sourceDoor): targetArea })
|
|
self.areaConnectionOffsets.update({
|
|
'S{}-{:02X}'.format(sourceArea, sourceDoor): targetOffset })
|
|
sourceArea = int.from_bytes(sourceRom.read(1), 'little')
|
|
sourceDoor = int.from_bytes(sourceRom.read(1), 'little')
|
|
targetOffset = sourceRom.tell()
|
|
targetArea = int.from_bytes(sourceRom.read(1), 'little')
|
|
for currentArea in range(7):
|
|
# print('DEBUG: Parsing sector {}'.format(currentArea))
|
|
sourceRom.seek(7977108 + currentArea * 4, 0)
|
|
data = sourceRom.read(4)
|
|
unpacked = struct.unpack('<L', data)
|
|
doorData = unpacked[0] - 134217728
|
|
sourceRom.seek(doorData, 0)
|
|
doorNumber = 0
|
|
while True:
|
|
connectionType = int.from_bytes(sourceRom.read(1), 'little')
|
|
roomNumber = int.from_bytes(sourceRom.read(1), 'little')
|
|
sourceRom.seek(4, 1)
|
|
connectedDoor = int.from_bytes(sourceRom.read(1), 'little')
|
|
sourceRom.seek(5, 1)
|
|
if connectionType == 0 and connectedDoor == 0:
|
|
break
|
|
sourceNode = 'S{}-{:02X}'.format(currentArea, doorNumber)
|
|
connectedArea = self.areaConnections.get(sourceNode, currentArea)
|
|
targetNode = 'S{}-{:02X}'.format(connectedArea, connectedDoor)
|
|
self.doorConnections.update({
|
|
sourceNode: targetNode })
|
|
roomNumber = format(roomNumber, '02X')
|
|
roomString = 'Room-S{}-{}'.format(currentArea, roomNumber)
|
|
doorString = 'S{}-{:02X}'.format(currentArea, doorNumber)
|
|
# print('DEBUG: Adding door S{}-{:02X} to S{}-{}'.format(currentArea, doorNumber, currentArea, roomNumber))
|
|
if roomString in self.rooms:
|
|
self.rooms[roomString].append(doorString)
|
|
else:
|
|
self.rooms[roomString] = [
|
|
doorString]
|
|
doorNumber += 1
|
|
# print('DEBUG: Parsing seemingly done?')
|
|
except FileNotFoundError:
|
|
print('Error:', vanillaGame, 'could not be opened.')
|
|
sys.exit(1)
|
|
|
|
|
|
def print_stats(self):
|
|
print("Path search time: old search = {}s, bfs time = {}s, dfs time = {}s".format(self.oldtime, self.bfstime, self.dfstime))
|
|
|
|
def set_setting(self, setting, value):
|
|
self.settings[setting] = value
|
|
|
|
|
|
def get_setting(self, setting):
|
|
return self.settings[setting]
|
|
|
|
|
|
def AddNodeToRoom(self, room, node):
|
|
nodeList = self.rooms.get(room)
|
|
nodeList.append(node)
|
|
nodeList.sort()
|
|
self.rooms.update({
|
|
room: nodeList })
|
|
|
|
|
|
def RemoveNodeFromRoom(self, room, node):
|
|
nodeList = self.rooms.get(room)
|
|
if node in nodeList:
|
|
nodeList.remove(node)
|
|
nodeList.sort()
|
|
self.rooms.update({
|
|
room: nodeList })
|
|
|
|
|
|
def ClearGraph(self):
|
|
self.graph.clear()
|
|
|
|
|
|
def ConnectAllNodes(self):
|
|
self.ClearGraph()
|
|
self.ConnectNodesInRooms()
|
|
self.ConnectNodesBetweenRooms()
|
|
|
|
|
|
def ConnectNodesInRooms(self):
|
|
for room in self.rooms:
|
|
for start in self.rooms[room]:
|
|
for end in self.rooms[room]:
|
|
if start != end:
|
|
self.add_edges(start, end)
|
|
|
|
|
|
def ConnectNodesBetweenRooms(self):
|
|
for connection in self.doorConnections.items():
|
|
if len(connection) == 2:
|
|
self.add_directed_edge(connection[0], connection[1])
|
|
|
|
def UpdateDoorConnection(self, source, destination):
|
|
self.doorConnections.update({
|
|
source: destination })
|
|
connectedArea = self.areaConnections.get(source)
|
|
if connectedArea != None:
|
|
self.areaConnections.update({
|
|
source: int(destination[1:2]) })
|
|
return self.areaConnectionOffsets.get(source)
|
|
|
|
|
|
def AddConnectedNodesToRoom(self, targetRoom, *nodes):
|
|
for currentNode in nodes:
|
|
if targetRoom in self.rooms:
|
|
self.rooms[targetRoom].append(currentNode)
|
|
for targetNode in self.rooms[targetRoom]:
|
|
if currentNode != targetNode:
|
|
self.add_edges(currentNode, targetNode)
|
|
else:
|
|
self.rooms[targetRoom] = [
|
|
currentNode]
|
|
|
|
def add_directed_edge(self, start, end):
|
|
if start != end:
|
|
if start in self.graph:
|
|
if end not in self.graph[start]:
|
|
self.graph[start].append(end)
|
|
else:
|
|
self.graph[start] = [
|
|
end]
|
|
|
|
def add_edges(self, start, *nodes):
|
|
for end in nodes:
|
|
self.add_directed_edge(start, end)
|
|
self.add_directed_edge(end, start)
|
|
|
|
|
|
def add_to_majors(self, item):
|
|
self.itemLocations.add(item)
|
|
if item not in self.majorItemLocations:
|
|
self.majorItemLocations.append(item)
|
|
|
|
|
|
def add_list_to_majors(self, locations):
|
|
for item in locations:
|
|
self.itemLocations.add(item)
|
|
if item not in self.majorItemLocations:
|
|
self.majorItemLocations.append(item)
|
|
|
|
def add_to_minors(self, item):
|
|
self.itemLocations.add(item)
|
|
if item not in self.minorItemLocations:
|
|
self.minorItemLocations.append(item)
|
|
|
|
def add_list_to_minors(self, locations):
|
|
for item in locations:
|
|
self.itemLocations.add(item)
|
|
if item not in self.minorItemLocations:
|
|
self.minorItemLocations.append(item)
|
|
|
|
def get_requirements(self, start, end):
|
|
checkRequirement = (start, end)
|
|
return self.requirements.get(checkRequirement)
|
|
|
|
def get_path(self, start, end, LimitArea=False, path=None, depth=100):
|
|
#return [start,end] if self.has_path(start,end,LimitArea, depth) else None
|
|
|
|
t = time.perf_counter()
|
|
path = self._get_path(start, end, LimitArea, path, depth)
|
|
self.oldtime += time.perf_counter() - t
|
|
|
|
return path
|
|
|
|
def _get_path(self, start, end, LimitArea, path, depth):
|
|
if path == None:
|
|
self.visited.clear()
|
|
self.queue.clear()
|
|
path = list()
|
|
path = path + [start]
|
|
if start not in self.graph:
|
|
return None
|
|
if start == end:
|
|
return path
|
|
if len(path) >= depth:
|
|
return None
|
|
for point in self.graph[start]:
|
|
if point in self.itemLocations:
|
|
if point not in end:
|
|
continue
|
|
if point in path:
|
|
continue
|
|
if LimitArea and point != end:
|
|
for area in range(0, 7):
|
|
if 'S{}'.format(area) in start:
|
|
if 'S{}'.format(area) not in point:
|
|
return None
|
|
edge = (start, point)
|
|
self.queue.append(edge)
|
|
while self.queue:
|
|
edge = self.queue.pop()
|
|
if edge not in self.visited:
|
|
self.visited.append(edge)
|
|
node = edge[1]
|
|
pathReqs = self.get_requirements(start, node)
|
|
if pathReqs == None:
|
|
newpath = self._get_path(node, end, LimitArea, path, depth)
|
|
if newpath:
|
|
path = path + [node]
|
|
return newpath
|
|
elif pathReqs == True:
|
|
newpath = self._get_path(node, end, LimitArea, path, depth)
|
|
if newpath:
|
|
path = path + [node]
|
|
return newpath
|
|
|
|
def has_path(self, start, end, LimitArea=False, depth=100):
|
|
if LimitArea:
|
|
area = self.itemArea.get(start)
|
|
if area == None:
|
|
for n in range(0, 7):
|
|
if 'S{}'.format(n) in start:
|
|
area = n
|
|
if area == None:
|
|
for n in range(0, 7):
|
|
if 'S{}'.format(n) in end:
|
|
area = n
|
|
else:
|
|
area = None
|
|
t = time.perf_counter()
|
|
result = self.has_path_bfs(start, end, area=area, max_depth=depth)
|
|
self.bfstime += time.perf_counter() - t
|
|
t = time.perf_counter()
|
|
r2 =self.has_path_dfs(start, end, area=area)
|
|
assert result == r2, (start, end, result, r2, area)
|
|
self.dfstime += time.perf_counter() - t
|
|
return result
|
|
|
|
# NOTE: both the start and end node need to be excluded from
|
|
# area checks because they can be things like bosses or the Water Pump
|
|
# that aren't associated with an area. all the intermediate nodes
|
|
# should be doors, and doors all have their sector number in their names.
|
|
|
|
def has_path_bfs(self, start, end, area=None, max_depth=100):
|
|
if start not in self.graph:
|
|
return False
|
|
if start == end:
|
|
return True
|
|
if area != None:
|
|
areaStr = 'S{}'.format(area)
|
|
seen = {start}
|
|
frontier = [start]
|
|
next_frontier = []
|
|
depth = 0
|
|
while frontier:
|
|
depth += 1
|
|
#if depth > max_depth: break
|
|
for node in frontier:
|
|
for neighbor in self.graph[node]:
|
|
if neighbor in seen:
|
|
continue
|
|
if neighbor in self.itemLocations:
|
|
if neighbor == end:
|
|
pathReqs = self.get_requirements(node, neighbor)
|
|
if pathReqs == None or pathReqs == True:
|
|
return True
|
|
#seen.add(neighbor)
|
|
continue
|
|
if area is not None:
|
|
if areaStr not in neighbor and neighbor != end:
|
|
continue
|
|
pathReqs = self.get_requirements(node, neighbor)
|
|
if pathReqs != None and pathReqs != True:
|
|
continue
|
|
if neighbor == end:
|
|
return True
|
|
|
|
next_frontier.append(neighbor)
|
|
|
|
# since this is a breadth-first search on an unweighted graph,
|
|
# we know that no other paths can be shorter than this one,
|
|
# so we can immediately mark the node as seen to prevent
|
|
# adding it to the frontier again
|
|
seen.add(neighbor)
|
|
|
|
frontier.clear()
|
|
frontier, next_frontier = next_frontier, frontier
|
|
|
|
return False
|
|
|
|
def has_path_dfs(self, start, end, area=None):
|
|
if start not in self.graph:
|
|
return False
|
|
if start == end:
|
|
return True
|
|
visited = set()
|
|
stack = [start]
|
|
if area != None:
|
|
areaStr = 'S{}'.format(area)
|
|
while stack:
|
|
node = stack.pop()
|
|
if node in visited:
|
|
continue
|
|
visited.add(node)
|
|
if node == end:
|
|
return True
|
|
if node is not start:
|
|
if node in self.itemLocations:
|
|
continue
|
|
if area is not None:
|
|
if areaStr not in node:
|
|
continue
|
|
|
|
for neighbor in reversed(self.graph[node]):
|
|
if neighbor in visited:
|
|
continue
|
|
pathReqs = self.get_requirements(node, neighbor)
|
|
if pathReqs != None and pathReqs != True:
|
|
continue
|
|
stack.append(neighbor)
|
|
return False
|