126 lines
4.6 KiB
Python
126 lines
4.6 KiB
Python
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
|
|
}
|
|
|
|
|