feature: remove some mara support in order to get rid of Antminer function calls.

pyasic/miners/backends/marathon.py

@@ -2,6 +2,7 @@ from typing import Optional
 
 from pyasic.errors import APIError
 from pyasic.miners.backends import AntminerModern
+from pyasic.miners.base import BaseMiner
 from pyasic.miners.data import (
     DataFunction,
     DataLocations,
@@ -13,50 +14,6 @@ from pyasic.web.marathon import MaraWebAPI
 
 MARA_DATA_LOC = DataLocations(
     **{
-        str(DataOptions.MAC): DataFunction(
-            "_get_mac",
-            [WebAPICommand("web_get_system_info", "get_system_info")],
-        ),
-        str(DataOptions.API_VERSION): DataFunction(
-            "_get_api_ver",
-            [RPCAPICommand("rpc_version", "version")],
-        ),
-        str(DataOptions.FW_VERSION): DataFunction(
-            "_get_fw_ver",
-            [RPCAPICommand("rpc_version", "version")],
-        ),
-        str(DataOptions.HOSTNAME): DataFunction(
-            "_get_hostname",
-            [WebAPICommand("web_get_system_info", "get_system_info")],
-        ),
-        str(DataOptions.HASHRATE): DataFunction(
-            "_get_hashrate",
-            [RPCAPICommand("rpc_summary", "summary")],
-        ),
-        str(DataOptions.EXPECTED_HASHRATE): DataFunction(
-            "_get_expected_hashrate",
-            [RPCAPICommand("rpc_stats", "stats")],
-        ),
-        str(DataOptions.FANS): DataFunction(
-            "_get_fans",
-            [RPCAPICommand("rpc_stats", "stats")],
-        ),
-        str(DataOptions.ERRORS): DataFunction(
-            "_get_errors",
-            [WebAPICommand("web_summary", "summary")],
-        ),
-        str(DataOptions.FAULT_LIGHT): DataFunction(
-            "_get_fault_light",
-            [WebAPICommand("web_get_blink_status", "get_blink_status")],
-        ),
-        str(DataOptions.IS_MINING): DataFunction(
-            "_is_mining",
-            [WebAPICommand("web_get_conf", "get_miner_conf")],
-        ),
-        str(DataOptions.UPTIME): DataFunction(
-            "_get_uptime",
-            [RPCAPICommand("rpc_stats", "stats")],
-        ),
         str(DataOptions.WATTAGE): DataFunction(
             "_get_wattage",
             [WebAPICommand("web_brief", "brief")],
@@ -65,7 +22,7 @@ MARA_DATA_LOC = DataLocations(
 )
 
 
-class MaraMiner(AntminerModern):
+class MaraMiner(BaseMiner):
     _web_cls = MaraWebAPI
     web: MaraWebAPI
 

pyasic/web/marathon.py

@@ -1,30 +1,56 @@
 from __future__ import annotations
 
+import asyncio
 import json
 from typing import Any
 
 import httpx
 
 from pyasic import settings
-from pyasic.web.antminer import AntminerModernWebAPI
+from pyasic.web.base import BaseWebAPI
 
 
-class MaraWebAPI(AntminerModernWebAPI):
+class MaraWebAPI(BaseWebAPI):
     def __init__(self, ip: str) -> None:
-        self.am_commands = [
-            "get_miner_conf",
-            "set_miner_conf",
-            "blink",
-            "reboot",
-            "get_system_info",
-            "get_network_info",
-            "summary",
-            "get_blink_status",
-            "set_network_conf",
-        ]
         super().__init__(ip)
 
-    async def _send_mara_command(
+    async def multicommand(
+        self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
+    ) -> dict:
+        async with httpx.AsyncClient(transport=settings.transport()) as client:
+            tasks = [
+                asyncio.create_task(self._handle_multicommand(client, command))
+                for command in commands
+            ]
+            all_data = await asyncio.gather(*tasks)
+
+        data = {}
+        for item in all_data:
+            data.update(item)
+
+        data["multicommand"] = True
+        return data
+
+    async def _handle_multicommand(
+        self, client: httpx.AsyncClient, command: str
+    ) -> dict:
+        auth = httpx.DigestAuth(self.username, self.pwd)
+
+        try:
+            url = f"http://{self.ip}:{self.port}/kaonsu/v1/{command}"
+            ret = await client.get(url, auth=auth)
+        except httpx.HTTPError:
+            pass
+        else:
+            if ret.status_code == 200:
+                try:
+                    json_data = ret.json()
+                    return {command: json_data}
+                except json.decoder.JSONDecodeError:
+                    pass
+        return {command: {}}
+
+    async def send_command(
         self,
         command: str | bytes,
         ignore_errors: bool = False,
@@ -56,76 +82,66 @@ class MaraWebAPI(AntminerModernWebAPI):
                 except json.decoder.JSONDecodeError:
                     pass
 
-    async def _send_am_command(
-        self,
-        command: str | bytes,
-        ignore_errors: bool = False,
-        allow_warning: bool = True,
-        privileged: bool = False,
-        **parameters: Any,
-    ):
-        url = f"http://{self.ip}:{self.port}/cgi-bin/{command}.cgi"
-        auth = httpx.DigestAuth(self.username, self.pwd)
-        try:
-            async with httpx.AsyncClient(
-                transport=settings.transport(),
-            ) as client:
-                if parameters:
-                    data = await client.post(
-                        url,
-                        auth=auth,
-                        timeout=settings.get("api_function_timeout", 3),
-                        json=parameters,
-                    )
-                else:
-                    data = await client.get(url, auth=auth)
-        except httpx.HTTPError:
-            pass
-        else:
-            if data.status_code == 200:
-                try:
-                    return data.json()
-                except json.decoder.JSONDecodeError:
-                    pass
-
-    async def send_command(
-        self,
-        command: str | bytes,
-        ignore_errors: bool = False,
-        allow_warning: bool = True,
-        privileged: bool = False,
-        **parameters: Any,
-    ) -> dict:
-        if command in self.am_commands:
-            return await self._send_am_command(
-                command,
-                ignore_errors=ignore_errors,
-                allow_warning=allow_warning,
-                privileged=privileged,
-                **parameters,
-            )
-        return await self._send_mara_command(
-            command,
-            ignore_errors=ignore_errors,
-            allow_warning=allow_warning,
-            privileged=privileged,
-            **parameters,
-        )
-
     async def brief(self):
         return await self.send_command("brief")
 
+    async def ping(self):
+        return await self.send_command("ping")
+
+    async def get_locate_miner(self):
+        return await self.send_command("locate_miner")
+
+    async def set_locate_miner(self, blinking: bool):
+        return await self.send_command("locate_miner", blinking=blinking)
+
+    async def reboot(self):
+        return await self.send_command("maintenance", type="reboot")
+
+    async def reset(self):
+        return await self.send_command("maintenance", type="reset")
+
+    async def reload(self):
+        return await self.send_command("maintenance", type="reload")
+
+    async def set_password(self, new_pwd: str):
+        return await self.send_command(
+            "maintenance",
+            type="passwd",
+            params={"curPwd": self.pwd, "confirmPwd": self.pwd, "newPwd": new_pwd},
+        )
+
+    async def get_network_config(self):
+        return await self.send_command("network_config")
+
+    async def set_network_config(self, **params):
+        return await self.send_command("network_config", **params)
+
+    async def get_miner_config(self):
+        return await self.send_command("miner_config")
+
+    async def set_network_config(self, **params):
+        return await self.send_command("miner_config", **params)
+
+    async def fans(self):
+        return await self.send_command("fans")
+
+    async def log(self):
+        return await self.send_command("log")
+
     async def overview(self):
         return await self.send_command("overview")
 
     async def connections(self):
         return await self.send_command("connections")
 
+    async def controlboard_info(self):
+        return await self.send_command("controlboard_info")
+
     async def event_chart(self):
         return await self.send_command("event_chart")
 
     async def hashboards(self):
         return await self.send_command("hashboards")
 
-    async def mara_pools(self):
+    async def pools(self):
-        return await self._send_mara_command("pools")
+        return await self.send_command("pools")