MarketParser.py

from __future__ import annotations

import json
import os
import time
from operator import itemgetter
from sys import platform
from time import sleep
from EDlogger import logger


class MarketParser:
    """ Parses the Market.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/Market.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/Market.json")
        self.last_mod_time = None

        # Read json file data
        self.current_data = self.get_market_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_market_data(self):
        """Loads data from the JSON file and returns the data, or None if the file does not exist.
        {
        "timestamp": "2024-09-21T14:53:38Z",
        "event": "Market",
        "MarketID": 129019775,
        "StationName": "Rescue Ship Cornwallis",
        "StationType": "MegaShip",
        "StarSystem": "V886 Centauri",
        "Items": [
            {
                "id": 128049152,
                "Name": "$platinum_name;",
                "Name_Localised": "Platinum",
                "Category": "$MARKET_category_metals;",
                "Category_Localised": "Metals",
                "BuyPrice": 3485,
                "SellPrice": 3450,
                "MeanPrice": 58272,
                "StockBracket": 0,
                "DemandBracket": 0,
                "Stock": 0,
                "Demand": 0,
                "Consumer": false,
                "Producer": false,
                "Rare": false
            }, { etc. } ]
        """
        # Check file if exists
        if not os.path.exists(self.file_path):
            return None

        # Check if file changed
        if self.get_file_modified_time() == self.last_mod_time:
            # logger.debug(f'Market.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 Market.json file. File may be open.')
                sleep(backoff)
                logger.debug('Attempting to re-read Market.json file after delay.')
                backoff *= 2

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

    def get_sellable_items(self, cargo_parser) -> list | None:
        """ Get a list of items that can be sold to the station.
        Will trigger a read of the json file.
        {
            "id": 128049154,
            "Name": "$gold_name;",
            "Name_Localised": "Gold",
            "Category": "$MARKET_category_metals;",
            "Category_Localised": "Metals",
            "BuyPrice": 49118,
            "SellPrice": 48558,
            "MeanPrice": 47609,
            "StockBracket": 2,
            "DemandBracket": 0,
            "Stock": 89,
            "Demand": 1,
            "Consumer": true,
            "Producer": false,
            "Rare": false
        }
        @param cargo_parser: Current cargo to check if rare or demand=1 items exist in hold.
        @return: A list of commodities that can be sold.
        """
        data = self.get_market_data()
        if data is None:
            return None

        sellable_items = [x for x in data['Items'] if x['Consumer'] or
                          ((x['Demand'] > 1 or x['Demand'] or x['Rare']) and cargo_parser.get_item(x['Name_Localised']) is not None)]
        # DemandBracket: 0=Not listed, 1=Low Demand, 2=Medium Demand, 3=High Demand
        # sellable_items = [x for x in data['Items'] if x['DemandBracket'] > 0]
        # sellable_items = [x for x in data['Items'] if self.can_sell_item(x['Name_Localised'])]
        # print(json.dumps(newlist, indent=4))

        # Sort by name, then category
        sorted_list1 = sorted(sellable_items, key=lambda x: x['Name_Localised'].lower())
        sorted_list2 = sorted(sorted_list1, key=lambda x: x['Category_Localised'].lower())

        logger.debug("Listing sellable commodities in market order")
        for x in sorted_list2:
            logger.debug(f"\t{x}")
        logger.debug("Finished listing sellable commodities in market order")

        return sorted_list2

    def get_buyable_items(self) -> list | None:
        """ Get a list of items that can be bought from the station.
        Will trigger a read of the json file.
        {
            "id": 128049154,
            "Name": "$gold_name;",
            "Name_Localised": "Gold",
            "Category": "$MARKET_category_metals;",
            "Category_Localised": "Metals",
            "BuyPrice": 49118,
            "SellPrice": 48558,
            "MeanPrice": 47609,
            "StockBracket": 2,
            "DemandBracket": 0,
            "Stock": 89,
            "Demand": 1,
            "Consumer": false,
            "Producer": true,
            "Rare": false
        }
        """
        data = self.get_market_data()
        if data is None:
            return None

        # buyable_items = [x for x in data['Items'] if x['Producer'] and x['Stock'] > 0]
        # StockBracket: 0=Not listed, 1=Low Stock, 2=Medium Stock, 3=High Stock
        # buyable_items = [x for x in data['Items'] if x['StockBracket'] > 0]
        buyable_items = [x for x in data['Items'] if self.can_buy_item(x['Name_Localised'])]
        # print(json.dumps(newlist, indent=4))

        # Sort by name, then category
        sorted_list1 = sorted(buyable_items, key=lambda x: x['Name_Localised'].lower())
        sorted_list2 = sorted(sorted_list1, key=lambda x: x['Category_Localised'].lower())

        logger.debug("Listing buyable commodities in market order")
        for x in sorted_list2:
            logger.debug(f"\t{x}")
        logger.debug("Finished listing buyable commodities in market order")

        return sorted_list2

    def get_market_name(self) -> str:
        """ Gets the current market (station) name.
        Will not trigger a read of the json file.
        """
        if self.current_data is None:
            return ""

        return self.current_data['StationName']

    def get_item(self, item_name) -> 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.
        {
            "id": 128049154,
            "Name": "$gold_name;",
            "Name_Localised": "Gold",
            "Category": "$MARKET_category_metals;",
            "Category_Localised": "Metals",
            "BuyPrice": 49118,
            "SellPrice": 48558,
            "MeanPrice": 47609,
            "StockBracket": 2,
            "DemandBracket": 0,
            "Stock": 89,
            "Demand": 1,
            "Consumer": false,
            "Producer": true,
            "Rare": false
        }
        """
        if self.current_data is None:
            return None

        for good in self.current_data['Items']:
            if good['Name_Localised'].upper() == item_name.upper():
                # print(json.dumps(good, indent=4))
                return good

        return None

    def can_buy_item(self, item_name: str) -> bool:
        """ Can the item be bought from the market (is it sold and is there stock).
        Will not trigger a read of the json file.
        """
        good = self.get_item(item_name)
        if good is None:
            return False

        return ((good['Stock'] > 0 and good['Producer'] and not good['Rare'])
                or (good['Stock'] > 0 and good['Rare']))  # Need producer in for non-Rares?

    def can_sell_item(self, item_name: str) -> bool:
        """ Can the item be sold to the market (is it bought, regardless of demand).
        Will not trigger a read of the json file.
        """
        good = self.get_item(item_name)
        if good is None:
            return False

        return good['Consumer'] or good['Demand'] > 0 or good['Rare']


# Usage Example
if __name__ == "__main__":
    parser = MarketParser()
    while True:
        cleaned_data = parser.get_market_data()

        #item = parser.get_item('water')
        #sell = parser.can_sell_item('water')
        #buy = parser.can_buy_item('water')

        items = parser.get_buyable_items()
        #items = parser.get_sellable_items()
        sorted_list = sorted(items, key=itemgetter('Name_Localised'))
        sorted_list = sorted(sorted_list, key=itemgetter('Category_Localised'))
        for item in sorted_list:
            print(f"Item: {item['Name_Localised']} ({item['Category_Localised']}) DemandBracket:{item['DemandBracket']} StockBracket:{item['StockBracket']}")

        #print(f"Curr Time: {time.time()}")
        #print(f"Sell water: {sell}")
        # print(json.dumps(cleaned_data, indent=4))

        time.sleep(1)
        break