CargoParser.py

from __future__ import annotations

import json
import os
import time
from datetime import datetime, timedelta
import queue
from sys import platform
import threading
from time import sleep
from EDlogger import logger
from WindowsKnownPaths import *


class CargoParser:
    """ Parses the Cargo.json file generated by the game. """
    def __init__(self, file_path=None):
        if platform != "win32":
            self.file_path = file_path if file_path else "./linux_ed/Cargo.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/Cargo.json")
        self.last_mod_time = None

        # Read json file data
        self.current_data = self.get_cargo_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 get_cargo_data(self):
        """Loads data from the JSON file and returns the data.
        { "timestamp":"2025-04-20T23:23:25Z", "event":"Cargo", "Vessel":"Ship", "Count":0, "Inventory":[

         ] }
        """
        # Check if file changed
        if self.get_file_modified_time() == self.last_mod_time:
            #logger.debug(f'Cargo.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', encoding='utf-8') as file:
                    data = json.load(file)
                    break
            except Exception as e:
                logger.debug('An error occurred reading Cargo.json file. File may be open.')
                sleep(backoff)
                logger.debug('Attempting to re-read Cargo.json file after delay.')
                backoff *= 2

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

    def get_item(self, item_name: str) -> dict[any] | None:
        """ Get details of one item. Returns the item detail as below, or None if item does not exist.
            Will not trigger a read of the json file.
        {
            "Name":"cmmcomposite",
            "Name_Localised":"CMM Composite", {This is optional}
            "Count":1236,
            "Stolen":0
        }
        @param item_name: The commodity name.
        @return: Details on the commodity if in hold, else None.
        """
        for good in self.current_data['Inventory']:
            if good['Name'].upper() == item_name.upper():
                # print(json.dumps(good, indent=4))
                return good
            if 'Name_Localised' in good:
                if good['Name_Localised'].upper() == item_name.upper():
                    # print(json.dumps(good, indent=4))
                    return good
        return None

    def get_items(self) -> list[any]:
        """ Get all items in the cargo in a list.
        Will not trigger a read of the json file. """
        return self.current_data['Inventory']

    def wait_for_file_change(self, start_timestamp, timeout: float = 5) -> bool:
        """ Waits for the file to change.
        Returns True if the file changes or False on a time-out.
        @param start_timestamp: The initial timestamp from 'timestamp' value.
        @param timeout: Timeout in seconds.
        """
        start_time = time.time()
        while (time.time() - start_time) < timeout:
            # Check file and read now data
            self.get_cargo_data()
            # Check if internal timestamp changed
            if self.current_data['timestamp'] != start_timestamp:
                return True

            sleep(0.5)

        return False

# Usage Example
if __name__ == "__main__":
    cargo_parser = CargoParser()
    while True:
        cleaned_data = cargo_parser.get_cargo_data()
        item = cargo_parser.get_item('Tritium')
        time.sleep(1)