add new runner to get inverter data from apsystems cloud

Inverter.py

@@ -0,0 +1,51 @@
+import asyncio
+import logging
+
+import requests
+from requests import Response
+
+from InverterData import InverterData
+
+
+class Inverter:
+    def __init__(self, queue: asyncio.Queue, inverter_id, user_token):
+        self._queue = queue
+        self.inverter_id = inverter_id
+        self.user_token = user_token
+        self.gather_data = False
+
+    async def start(self):
+        self.gather_data = True
+        await self.gather()
+
+    async def gather(self):
+        while self.gather_data:
+            realtime_data: dict = get_realtime_data(self.inverter_id, self.user_token)
+            data = InverterData(
+                self.inverter_id,
+                realtime_data
+            )
+            await self._queue.put(data)
+            await asyncio.sleep(60)
+
+
+def get_realtime_data(inverter_id: str, user_token: str) -> dict:
+    url = f"https://app.api.apsystemsema.com:9223/aps-api-web/api/v2/data/device/ezInverter/realTime/{inverter_id}"
+    headers = {"authorization": f"Bearer {user_token}"}
+    result = requests.get(url, headers=headers)
+    return handle_api_return_codes(result)
+
+
+def handle_api_return_codes(result: Response) -> [dict, None]:
+    result_api_code = result.json()["code"]
+    if result.status_code == 200:
+        if result_api_code == 0:
+            return result.json()
+        if result_api_code == 3003:
+            print("not Implemented yet, you need to refresh your token.")
+            return None
+        if result_api_code == 3001:
+            logging.error("Not authorized")
+            exit(1)
+    else:
+        logging.error(f"Something went wrong in the request. Status code: {result.status_code}")

InverterData.py

@@ -0,0 +1,7 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class InverterData:
+    name: str
+    data: dict

Usertoken.py

@@ -0,0 +1,27 @@
+import os
+import requests
+
+
+def generate_user_token(username: str, password: str) -> str:
+    url = "https://app.api.apsystemsema.com:9223/api/token/generateToken/user/login"
+
+    data = {
+        "password":  password,
+        "app_id": "4029817264d4821d0164d4821dd80015",
+        "app_secret": "EZAd2023",
+        "username": username
+    }
+    post = requests.post(url, data=data)
+    return post.json()["data"]['access_token']
+
+
+def cache_user_token_on_disk(user_token) -> None:
+    with open("user_token", "w") as f:
+        f.write(user_token)
+
+
+def get_cached_user_token() -> str:
+    if os.path.exists("user_token"):
+        with open("user_token", "r") as f:
+            return f.read()
+    return ""

config.toml-default

@@ -0,0 +1,2 @@
+username = ""
+password = ""

run.py

@@ -0,0 +1,34 @@
+import asyncio
+import os
+import toml
+
+from Inverter import Inverter
+from Usertoken import generate_user_token, cache_user_token_on_disk, get_cached_user_token
+
+
+async def main():
+    os.path.dirname(__file__)
+    with open("config.toml", "r") as f:
+        config = toml.load(f)
+
+    token = get_cached_user_token()
+
+    if not token:
+        token = generate_user_token(config["username"], config["password"])
+        cache_user_token_on_disk(token)
+
+    inverter_stefan = "E07000000405"
+    inverter_christian = "E07000000083"
+
+    queue = asyncio.Queue()
+    inverters = [Inverter(queue, inverter_christian, token)]
+    for inverter in inverters:
+        asyncio.create_task(inverter.start())
+
+    while True:
+        item = await queue.get()
+        print(item)
+
+
+if __name__ == '__main__':
+    asyncio.run(main())