OriginDex-DataManager/ui/workers/solve_pddl_route.py

import json
import re
import sys
from PyQt6.QtCore import QObject, pyqtSignal, QRunnable
import subprocess
import os
from db import db

from routes.pokemon_game_desc import PokemonGameDesc
from utility.convert_to_pddl import format_name, generate_pddl_problem

class SolvePDDLsWorkerSignals(QObject):
    finished = pyqtSignal(list)

class SolvePDDLsWorker(QRunnable):
    def __init__(self, data: PokemonGameDesc):
        super().__init__()
        self.signals = SolvePDDLsWorkerSignals()
        self.data = data

    def run(self):
        try:
            gathered_data = self.solve_PDDL(self.data)
            self.signals.finished.emit(gathered_data)
        except Exception as e:
            print(f"Error gathering Pokémon home storage status: {e}")

    def solve_PDDL(self, data: PokemonGameDesc, game_name: str = ""):
        if game_name == "":
            game_name = "Crystal"

        route, _, _ = data.astar_search()

        locations = set(route)

        print("Unique locations visisted: ", locations)

        output_file = "./temp/plan.json"
        plan = {}
        with open(output_file, 'r', encoding='utf-8') as f:
            plan = json.load(f)

        target = None
        for game in plan:
            if game["game_name"] == game_name:
                target = game
                break

        if not target:
            return []
        
        # gather up all the pokemon needed for say crystal and find all the encounter routes/locations
        needed_locations = set()
        
        game_info = db.get_game_id_by_name(game_name)
        
        for key in target["pokemon"]:
            catch_stats = target["pokemon"][key]
            rep = catch_stats["representative"]
            rand = db.get_encounters(rep, "random")
            static = db.get_encounters(rep, "static")
            encounters = []
            encounters.extend(rand)
            encounters.extend(static)

            for encounter in encounters:
                if encounter["game_id"] != game_info["id"]:
                    continue
                
                encounter_data = json.loads(encounter["data"])
                needed_locations.add(encounter_data["location"].replace("*", ""))

        # compare those to all the ones we visit in the generated route
        missing_locations = needed_locations - locations
        print("missing_locations:", missing_locations)

        node_list = list(data.graph.nodes())

        # add missing ones to the has_visited list and re-gen the problem, then run it again
        nodes_to_visit = []
        for loc in missing_locations:
            if loc in node_list:
                nodes_to_visit.append(loc)

        print("Additional Nodes to Visit:",  nodes_to_visit)

        data.must_visit = data.must_visit | set(nodes_to_visit[0:2])

        route, _, _ = data.astar_search()

        ordered_dict = dict.fromkeys(route)
        unique_list = list(ordered_dict)

        return unique_list

    def solve_PDDL_old(self, data: PokemonGameDesc, game_name: str = ""):
        if game_name == "":
            game_name = "Crystal"

        plan_file_name = f'./{game_name.lower()}_sas_plan'

        if os.path.exists(os.path.abspath(plan_file_name)) == False:
            self.generate_sas_plan(data)
        
        locations = set()
        pattern = re.compile(r"\((?:move|fly)\s+(\w+)\s+(\w+)\)")
        with open(plan_file_name, 'r') as f:
            for line in f:
                # Find all matches for move and fly actions
                matches = pattern.findall(line)
                for loc1, loc2 in matches:
                    # Add both locations to the set
                    locations.add(loc1)
                    locations.add(loc2)

        print("Unique locations visisted: ", locations)

        output_file = "./temp/plan.json"
        plan = {}
        with open(output_file, 'r', encoding='utf-8') as f:
            plan = json.load(f)

        target = None
        for game in plan:
            if game["game_name"] == game_name:
                target = game
                break

        if not target:
            return []
        
        # gather up all the pokemon needed for say crystal and find all the encounter routes/locations
        needed_locations = set()
        
        game_info = db.get_game_id_by_name(game_name)
        
        for key in target["pokemon"]:
            catch_stats = target["pokemon"][key]
            rep = catch_stats["representative"]
            rand = db.get_encounters(rep, "random")
            static = db.get_encounters(rep, "static")
            encounters = []
            encounters.extend(rand)
            encounters.extend(static)

            for encounter in encounters:
                if encounter["game_id"] != game_info["id"]:
                    continue
                
                encounter_data = json.loads(encounter["data"])
                needed_locations.add(format_name(encounter_data["location"].replace("*", "")).lower())

        # compare those to all the ones we visit in the generated route
        missing_locations = needed_locations - locations
        print("missing_locations:", missing_locations)

        node_list = list(data.graph.nodes())
        for i in range(len(node_list)):
            node_list[i] = format_name(node_list[i])

        # add missing ones to the has_visited list and re-gen the problem, then run it again
        nodes_to_visit = []
        for loc in missing_locations:
            if loc in node_list:
                nodes_to_visit.append(loc)

        print("Additional Nodes to Visit:",  nodes_to_visit)

        data.has_visited.extend(nodes_to_visit)

        generate_pddl_problem(data)

        self.generate_sas_plan(data)

        return []
    
    def generate_sas_plan(self, data: PokemonGameDesc):
        print("Starting fast-downward...")
        try:
            command_translate = [
                'python',
                os.path.abspath('./downward/builds/release/bin/translate/translate.py'),
                os.path.abspath('./temp/pokemon_domain.pddl'),
                os.path.abspath(f'./temp/{data.file_name}'),
                '--sas-file',
                'output.sas'
            ]

            command_search = [
                os.path.abspath('./downward/builds/release/bin/downward.exe'),
                '--search',
                'astar(hmax())',
                '--internal-plan-file',
                'sas_plan'
            ]

            if os.path.exists(os.path.abspath('./output.sas')):
                os.remove(os.path.abspath('./output.sas'))

            if os.path.exists(os.path.abspath('./sas_plan')):
                os.remove(os.path.abspath('./sas_plan'))
            
            print("Command to be executed:", command_translate)

            ret = subprocess.run(command_translate, check=True)

            with open(f'./output.sas', 'r') as f:
                print("Command to be executed:", command_translate)
                ret = subprocess.run(command_search, check=True, creationflags=subprocess.CREATE_NEW_PROCESS_GROUP, stdin=f)
        except subprocess.CalledProcessError as e:
            print(f"An error occurred: {e}")
        print("fast-downward compelted")