import sys import json import requests class EnergyProducer: def __init__(self, config): self.ID = config.get("ID") self.IP = config.get("IP") self.Adr = config.get("Adr") def get_current_value(self): raise NotImplementedError("Subclasses must implement use_energy method") @staticmethod def create_producerlist_from_json(json_data): subclasses = [cls for cls in sys.modules[__name__].__dict__.values() if isinstance(cls, type) and issubclass(cls, EnergyProducer) and cls != EnergyProducer] energy_producers = [] data = json.loads(json_data) for user_data in data["producers"]: user_type = user_data["type"] user_config = user_data["config"] for subclass in subclasses: if user_type in subclass.__name__: energy_producers.append(subclass(user_config)) return energy_producers @staticmethod def get_all_current_values(producers): data = {"Producer_sum": {"success": True, "day_energy": 0, "pac": 0, "total_energy": 0, "year_energy": 0}} for producer in producers: if isinstance(producer, EnergyProducer): current_value = producer.get_current_value() producer_data = { "success": current_value["success"], "day_energy": current_value.get("day_energy", 0), "pac": current_value.get("pac", 0), "total_energy": current_value.get("total_energy", 0), "year_energy": current_value.get("year_energy", 0) } data[producer.ID] = producer_data if current_value["success"]: data["Producer_sum"]["success"] &= True data["Producer_sum"]["day_energy"] += producer_data["day_energy"] data["Producer_sum"]["pac"] += producer_data["pac"] data["Producer_sum"]["total_energy"] += producer_data["total_energy"] data["Producer_sum"]["year_energy"] += producer_data["year_energy"] else: data["Producer_sum"]["success"] &= False return data class Producer1(EnergyProducer): def __init__(self, config): self.ID = config.get("ID") self.IP = config.get("IP") self.Adr = config.get("Adr") def get_current_value(self): raise NotImplementedError("Subclasses must implement use_energy method") class Producer2(EnergyProducer): def __init__(self, config): self.ID = config.get("ID") self.IP = config.get("IP") self.Adr = config.get("Adr") def get_current_value(self): raise NotImplementedError("Subclasses must implement use_energy method") class Fronius(EnergyProducer): def __init__(self, config): self.ID = config.get("ID") self.IP = config.get("IP") self.Adr = config.get("Adr") def get_current_value(self): url = f"http://{self.IP}/solar_api/v1/GetInverterRealtimeData.cgi?Scope=System" try: response = requests.get(url) success = response.status_code == 200 data = response.json() if success else None if data: day_energy = data.get("Body", {}).get("Data", {}).get("DAY_ENERGY", {}).get("Values", {}).get(str(self.Adr)) pac = data.get("Body", {}).get("Data", {}).get("PAC", {}).get("Values", {}).get(str(self.Adr)) total_energy = data.get("Body", {}).get("Data", {}).get("TOTAL_ENERGY", {}).get("Values", {}).get(str(self.Adr)) year_energy = data.get("Body", {}).get("Data", {}).get("YEAR_ENERGY", {}).get("Values", {}).get(str(self.Adr)) return { "success": success, "day_energy": day_energy, "pac": pac, "total_energy": total_energy, "year_energy": year_energy } else: return { "success": success, "day_energy": None, "pac": None, "total_energy": None, "year_energy": None } except: return { "success": False, "day_energy": None, "pac": None, "total_energy": None, "year_energy": None }