StatusParser.py

import json
import os
import time
from datetime import datetime, timedelta
import queue
from sys import platform
import threading
from time import sleep

from EDAP_data import *
from EDlogger import logger
from Voice import Voice
from WindowsKnownPaths import *


class StatusParser:
    """ Parses the Status.json file generated by the game.
    Thanks go to Rude. See source at 'https://github.com/RatherRude/Elite-Dangerous-AI-Integration'."""
    def __init__(self, file_path=None):
        if platform != "win32":
            self.file_path = file_path if file_path else "./linux_ed/Status.json"
        else:
            from WindowsKnownPaths import get_path, FOLDERID, UserHandle

            self.file_path = file_path if file_path else (get_path(FOLDERID.SavedGames, UserHandle.current)
                                                          + "/Frontier Developments/Elite Dangerous/Status.json")
        self.last_mod_time = None

        # Read json file data
        self.last_data = None
        self.current_data = None
        self.current_data = self.get_cleaned_data()

    #     self.watch_thread = threading.Thread(target=self._watch_file_thread, daemon=True)
    #     self.watch_thread.start()
    #     self.status_queue = queue.Queue()
    #
    # def _watch_file_thread(self):
    #     backoff = 1
    #     while True:
    #         try:
    #             self._watch_file()
    #         except Exception as e:
    #             logger.debug('An error occurred when reading status file')
    #             sleep(backoff)
    #             logger.debug('Attempting to restart status file reader after failure')
    #             backoff *= 2
    #
    # def _watch_file(self):
    #     """Detects changes in the Status.json file."""
    #     while True:
    #         status = self.get_cleaned_data()
    #         if status != self.current_data:
    #             self.status_queue.put(status)
    #             self.current_data = status
    #         sleep(1)

    def get_file_modified_time(self) -> float:
        return os.path.getmtime(self.file_path)

    def translate_flags(self, flags_value):
        """Translates flags integer to a dictionary of only True flags."""
        all_flags = {
            "Docked": bool(flags_value & 1),
            "Landed": bool(flags_value & 2),
            "Landing Gear Down": bool(flags_value & 4),
            "Shields Up": bool(flags_value & 8),
            "Supercruise": bool(flags_value & 16),
            "FlightAssist Off": bool(flags_value & 32),
            "Hardpoints Deployed": bool(flags_value & 64),
            "In Wing": bool(flags_value & 128),
            "Lights On": bool(flags_value & 256),
            "Cargo Scoop Deployed": bool(flags_value & 512),
            "Silent Running": bool(flags_value & 1024),
            "Scooping Fuel": bool(flags_value & 2048),
            "Srv Handbrake": bool(flags_value & 4096),
            "Srv using Turret view": bool(flags_value & 8192),
            "Srv Turret retracted (close to ship)": bool(flags_value & 16384),
            "Srv DriveAssist": bool(flags_value & 32768),
            "Fsd MassLocked": bool(flags_value & 65536),
            "Fsd Charging": bool(flags_value & 131072),
            "Fsd Cooldown": bool(flags_value & 262144),
            "Low Fuel (< 25%)": bool(flags_value & 524288),
            "Over Heating (> 100%)": bool(flags_value & 1048576),
            "Has Lat Long": bool(flags_value & 2097152),
            "IsInDanger": bool(flags_value & 4194304),
            "Being Interdicted": bool(flags_value & 8388608),
            "In MainShip": bool(flags_value & 16777216),
            "In Fighter": bool(flags_value & 33554432),
            "In SRV": bool(flags_value & 67108864),
            "Hud in Analysis mode": bool(flags_value & 134217728),
            "Night Vision": bool(flags_value & 268435456),
            "Altitude from Average radius": bool(flags_value & 536870912),
            "Fsd Jump": bool(flags_value & 1073741824),
            "Srv HighBeam": bool(flags_value & 2147483648),
        }

        # Return only flags that are True
        true_flags = {key: value for key, value in all_flags.items() if value}
        return true_flags

    def translate_flags2(self, flags2_value):
        """Translates Flags2 integer to a dictionary of only True flags."""
        all_flags2 = {
            "OnFoot": bool(flags2_value & 1),
            "InTaxi": bool(flags2_value & 2),
            "InMulticrew": bool(flags2_value & 4),
            "OnFootInStation": bool(flags2_value & 8),
            "OnFootOnPlanet": bool(flags2_value & 16),
            "AimDownSight": bool(flags2_value & 32),
            "LowOxygen": bool(flags2_value & 64),
            "LowHealth": bool(flags2_value & 128),
            "Cold": bool(flags2_value & 256),
            "Hot": bool(flags2_value & 512),
            "VeryCold": bool(flags2_value & 1024),
            "VeryHot": bool(flags2_value & 2048),
            "Glide Mode": bool(flags2_value & 4096),
            "OnFootInHangar": bool(flags2_value & 8192),
            "OnFootSocialSpace": bool(flags2_value & 16384),
            "OnFootExterior": bool(flags2_value & 32768),
            "BreathableAtmosphere": bool(flags2_value & 65536),
            "Telepresence Multicrew": bool(flags2_value & 131072),
            "Physical Multicrew": bool(flags2_value & 262144),
            "Fsd hyperdrive charging": bool(flags2_value & 524288),
            "Flags2Future20": bool(flags2_value & 1048576),
            "Flags2Future21": bool(flags2_value & 2097152),
            "Flags2Future22": bool(flags2_value & 4194304),
            "Flags2Future23": bool(flags2_value & 8388608),
            "Flags2Future24": bool(flags2_value & 16777216),
            "Flags2Future25": bool(flags2_value & 33554432),
            "Flags2Future26": bool(flags2_value & 67108864),
            "Flags2Future27": bool(flags2_value & 134217728),
            "Flags2Future28": bool(flags2_value & 268435456),
            "Flags2Future29": bool(flags2_value & 536870912),
            "Flags2Future30": bool(flags2_value & 1073741824),
            "Flags2Future31": bool(flags2_value & 2147483648),
        }

        # Return only flags that are True
        true_flags2 = {key: value for key, value in all_flags2.items() if value}
        return true_flags2

    def transform_pips(self, pips_list):
        """Transforms the pips list to a dictionary and halves each value."""
        return {
            'system': pips_list[0] / 2,
            'engine': pips_list[1] / 2,
            'weapons': pips_list[2] / 2
        }

    def adjust_year(self, timestamp):
        """Increases the year in the timestamp by 1286 years."""
        # Parse the timestamp string into a datetime object
        dt = datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%SZ")

        # Increase the year by 1286
        dt = dt.replace(year=dt.year + 1286)

        # Format the datetime object back into a string
        return dt.strftime("%Y-%m-%dT%H:%M:%SZ")

    def get_cleaned_data(self):
        """Loads data from the JSON file and returns cleaned data with only the necessary fields.
        {
        "timestamp":"2024-09-28T16:01:47Z",
        "event":"Status",
        "Flags":150994968,
        "Flags2":0,
        "Pips":[4,4,4],
        "FireGroup":0,
        "GuiFocus":0,
        "Fuel":
            {
                "FuelMain":36.160004,
                 "FuelReservoir":0.534688
            },
        "Cargo":728.000000,
        "LegalState":"Clean",
        "Balance":3119756215,
        "Destination":
            {
                "System":7267487524297,
                "Body":12,
                "Name":"P.T.N. PERSEVERANCE TFK-N3G"
            }
        }
        """
        # Check if file changed
        if self.get_file_modified_time() == self.last_mod_time:
            #logger.debug(f'Status.json mod timestamp {self.last_mod_time} unchanged.')
            #print(f'Status.json mod timestamp {self.last_mod_time} unchanged.')
            return self.current_data

        # Read file
        backoff = 1
        while True:
            try:
                with open(self.file_path, 'r') as file:
                    data = json.load(file)
                    break
            except Exception as e:
                logger.debug('An error occurred when reading Status.json file')
                sleep(backoff)
                logger.debug('Attempting to restart status file reader after failure')
                backoff *= 2

        # Combine flags from Flags and Flags2 into a single dictionary
        # combined_flags = {**self.translate_flags(data['Flags'])}

        # if 'Flags2' in data:
        #    combined_flags = {**combined_flags, **self.translate_flags2(data['Flags2'])}

        # Initialize cleaned_data with common fields
        cleaned_data = {
            #'status': combined_flags,
            'time': (datetime.now() + timedelta(days=469711)).isoformat(),
            'Flags': data['Flags']
        }

        # Add optional status flags
        if 'Flags2' in data:
            cleaned_data['Flags2'] = data['Flags2']
        if 'Pips' in data:
            cleaned_data['pips'] = self.transform_pips(data['Pips'])
        if 'GuiFocus' in data:
            cleaned_data['GuiFocus'] = data['GuiFocus']
        if 'Cargo' in data:
            cleaned_data['cargo'] = data['Cargo']
        if 'LegalState' in data:
            cleaned_data['legalState'] = data['LegalState']
        if 'Balance' in data:
            cleaned_data['balance'] = data['Balance']
        if 'Destination' in data:
            cleaned_data['Destination'] = data['Destination']
        else:
            cleaned_data['Destination'] = None

        # Store data
        self.last_data = self.current_data
        self.current_data = cleaned_data
        self.last_mod_time = self.get_file_modified_time()
        #logger.debug(f'Status.json mod timestamp {self.last_mod_time} updated.')
        # print(f'Status.json mod timestamp {self.last_mod_time} updated.')
        # print(json.dumps(data, indent=4))

        # Enable the following to print all the changes to the status flags.
        # self.log_flag_diffs()

        return cleaned_data

    def log_flag_diffs(self):
        if self.last_data is None:
            return
        if self.current_data is None:
            return

        old_flags = self.last_data['Flags']
        new_flags = self.current_data['Flags']

        flags_on = new_flags & ~ old_flags
        flag_array = self.translate_flags(flags_on)
        for item in flag_array:
            print(f"Status Flags: '{item}' is ON")

        flags_off = ~ new_flags & old_flags
        flag_array = self.translate_flags(flags_off)
        for item in flag_array:
            print(f"Status Flags: '{item}' is OFF")

        if self.last_data['Flags2'] is None:
            return
        if self.current_data['Flags2'] is None:
            return

        old_flags2 = self.last_data['Flags2']
        new_flags2 = self.current_data['Flags2']

        flags_on2 = new_flags2 & ~ old_flags2
        flag_array2 = self.translate_flags2(flags_on2)
        for item in flag_array2:
            print(f"Status Flags2: '{item}' is ON")

        flags_off2 = ~ new_flags2 & old_flags2
        flag_array2 = self.translate_flags2(flags_off2)
        for item in flag_array2:
            print(f"Status Flags2: '{item}' is OFF")

    def get_gui_focus(self) -> int:
        """ Gets the value of the GUI Focus flag.
        The Flag constants are defined in 'EDAP_data.py'.
        """
        self.get_cleaned_data()
        return self.current_data.get('GuiFocus', 0)

    def wait_for_flag_on(self, flag: int, timeout: float = 15) -> bool:
        """ Waits for the of the selected flag to turn true.
        Returns True if the flag turns true or False on a time-out.
        The Flag constants are defined in 'EDAP_data.py'.
        @param timeout: Timeout in seconds.
        @param flag: The flag to check for.
        """
        start_time = time.time()
        while (time.time() - start_time) < timeout:
            self.get_cleaned_data()
            if bool(self.current_data['Flags'] & flag):
                return True

            sleep(0.5)

        return False

    def wait_for_flag_off(self, flag: int, timeout: float = 15) -> bool:
        """ Waits for the of the selected flag to turn false.
        Returns True if the flag turns false or False on a time-out.
        The Flag constants are defined in 'EDAP_data.py'.
        @param timeout: Timeout in seconds.
        @param flag: The flag to check for.
        """
        start_time = time.time()
        while (time.time() - start_time) < timeout:
            self.get_cleaned_data()
            if not bool(self.current_data['Flags'] & flag):
                return True

            sleep(0.5)

        return False

    def wait_for_flag2_on(self, flag: int, timeout: float = 15) -> bool:
        """ Waits for the of the selected flag to turn true.
        Returns True if the flag turns true or False on a time-out.
        The Flag constants are defined in 'EDAP_data.py'.
        @param timeout: Timeout in seconds.
        @param flag: The flag to check for.
        """
        if 'Flags2' not in data:
            return False

        start_time = time.time()
        while (time.time() - start_time) < timeout:
            self.get_cleaned_data()
            if bool(self.current_data['Flags2'] & flag):
                return True

            sleep(0.5)

        return False

    def wait_for_flag2_off(self, flag: int, timeout: float = 15) -> bool:
        """ Waits for the of the selected flag to turn false.
        Returns True if the flag turns false or False on a time-out.
        The Flag constants are defined in 'EDAP_data.py'.
        @param timeout: Timeout in seconds.
        @param flag: The flag to check for.
        """
        if 'Flags2' not in data:
            return False

        start_time = time.time()
        while (time.time() - start_time) < timeout:
            self.get_cleaned_data()
            if not bool(self.current_data['Flags2'] & flag):
                return True

            sleep(0.5)

        return False

    def get_flag(self, flag: int) -> bool:
        """ Gets the value of the selected flag.
        The Flag constants are defined in 'EDAP_data.py'.
        @param flag: The flag to check for.
        """
        self.get_cleaned_data()
        return bool(self.current_data['Flags'] & flag)

    def get_flag2(self, flag: int) -> bool:
        """ Gets the value of the selected flag2.
        The Flag2 constants are defined in 'EDAP_data.py'.
        @param flag: The flag to check for.
        """
        self.get_cleaned_data()

        if 'Flags2' in self.current_data:
            return bool(self.current_data['Flags2'] & flag)
        else:
            return False

# Usage Example
if __name__ == "__main__":
    parser = StatusParser()

    while True:
        start_time = time.time()
        data = parser.get_cleaned_data()
        sleep(1)
        #docked = parser.get_flag(FlagsDocked, False)
        #landed = parser.get_flag(FlagsLanded, False)

        inmainship = parser.get_flag(FlagsInMainShip)
        print(f"Time: {(time.time() - start_time)}")
        #sc = parser.get_flag(FlagsSupercruise, False)

        #print(json.dumps(data, indent=4))
        #print(f"Supercruise: {sc}")
        time.sleep(1)
        #print("\n" * 10)