feature: add MAC address to mskminer

pyasic/miners/backends/mskminer.py

@@ -3,7 +3,14 @@ from typing import Optional
 from pyasic import APIError
 from pyasic.device.algorithm import AlgoHashRate
 from pyasic.miners.backends import BMMiner
-from pyasic.miners.data import DataFunction, DataLocations, DataOptions, RPCAPICommand
+from pyasic.miners.data import (
+    DataFunction,
+    DataLocations,
+    DataOptions,
+    RPCAPICommand,
+    WebAPICommand,
+)
+from pyasic.web.mskminer import MSKMinerWebAPI
 
 MSKMINER_DATA_LOC = DataLocations(
     **{
@@ -15,6 +22,10 @@ MSKMINER_DATA_LOC = DataLocations(
             "_get_fw_ver",
             [RPCAPICommand("rpc_version", "version")],
         ),
+        str(DataOptions.MAC): DataFunction(
+            "_get_mac",
+            [WebAPICommand("web_info_v1", "info_v1")],
+        ),
         str(DataOptions.HASHRATE): DataFunction(
             "_get_hashrate",
             [RPCAPICommand("rpc_stats", "stats")],
@@ -52,6 +63,9 @@ class MSKMiner(BMMiner):
 
     data_locations = MSKMINER_DATA_LOC
 
+    web: MSKMinerWebAPI
+    _web_cls = MSKMinerWebAPI
+
     async def _get_hashrate(self, rpc_stats: dict = None) -> Optional[AlgoHashRate]:
         # get hr from API
         if rpc_stats is None:
@@ -81,3 +95,16 @@ class MSKMiner(BMMiner):
                 return rpc_stats["STATS"][0]["total_power"]
             except (LookupError, ValueError, TypeError):
                 pass
+
+    async def _get_mac(self, web_info_v1: dict = None) -> Optional[str]:
+        if web_info_v1 is None:
+            try:
+                web_info_v1 = await self.web.info_v1()
+            except APIError:
+                pass
+
+        if web_info_v1 is not None:
+            try:
+                return web_info_v1["network_info"]["result"]["macaddr"].upper()
+            except (LookupError, ValueError, TypeError):
+                pass

pyasic/settings/__init__.py

@@ -41,6 +41,7 @@ _settings = {  # defaults
     "default_hive_web_password": "root",
     "default_iceriver_web_password": "12345678",
     "default_elphapex_web_password": "root",
+    "default_mskminer_web_password": "root",
     "default_antminer_ssh_password": "miner",
     "default_bosminer_ssh_password": "root",
 }

pyasic/web/mskminer.py

@@ -0,0 +1,72 @@
+# ------------------------------------------------------------------------------
+#  Copyright 2024 Upstream Data Inc                                            -
+#                                                                              -
+#  Licensed under the Apache License, Version 2.0 (the "License");             -
+#  you may not use this file except in compliance with the License.            -
+#  You may obtain a copy of the License at                                     -
+#                                                                              -
+#      http://www.apache.org/licenses/LICENSE-2.0                              -
+#                                                                              -
+#  Unless required by applicable law or agreed to in writing, software         -
+#  distributed under the License is distributed on an "AS IS" BASIS,           -
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.    -
+#  See the License for the specific language governing permissions and         -
+#  limitations under the License.                                              -
+# ------------------------------------------------------------------------------
+from __future__ import annotations
+
+import asyncio
+import warnings
+from typing import Any
+
+import httpx
+
+from pyasic import settings
+from pyasic.errors import APIError
+from pyasic.web.base import BaseWebAPI
+
+
+class MSKMinerWebAPI(BaseWebAPI):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.username = "admin"
+        self.pwd = settings.get("default_mskminer_web_password", "root")
+
+    async def multicommand(
+        self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
+    ) -> dict:
+        tasks = {c: asyncio.create_task(getattr(self, c)()) for c in commands}
+        await asyncio.gather(*[t for t in tasks.values()])
+        return {t: tasks[t].result() for t in tasks}
+
+    async def send_command(
+        self,
+        command: str | bytes,
+        ignore_errors: bool = False,
+        allow_warning: bool = True,
+        privileged: bool = False,
+        **parameters: Any,
+    ) -> dict:
+        async with httpx.AsyncClient(transport=settings.transport()) as client:
+            try:
+                # auth
+                await client.post(
+                    f"http://{self.ip}:{self.port}/admin/login",
+                    data={"username": self.username, "password": self.pwd},
+                )
+            except httpx.HTTPError:
+                warnings.warn(f"Could not authenticate with miner web: {self}")
+            try:
+                resp = await client.post(
+                    f"http://{self.ip}:{self.port}/api/{command}", params=parameters
+                )
+                if not resp.status_code == 200:
+                    if not ignore_errors:
+                        raise APIError(f"Command failed: {command}")
+                    warnings.warn(f"Command failed: {command}")
+                return resp.json()
+            except httpx.HTTPError:
+                raise APIError(f"Command failed: {command}")
+
+    async def info_v1(self):
+        return await self.send_command("info_v1")