OpenMFOR/Fusion_Graph.py

# 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