feature: add BTMinerV3 control support

Author: b-rowan

pyasic/miners/backends/btminer.py

@@ -829,6 +829,85 @@ class BTMinerV3(StockFirmware):
     supports_autotuning = True
     supports_power_modes = True
 
+    async def fault_light_off(self) -> bool:
+        try:
+            data = await self.rpc.set_system_led()
+        except APIError:
+            return False
+        if data:
+            if "code" in data.keys():
+                if data["code"] == 0:
+                    self.light = False
+                    return True
+        return False
+
+    async def fault_light_on(self) -> bool:
+        try:
+            data = await self.rpc.set_system_led(
+                leds=[
+                    {
+                        {"color": "red", "period": 60, "duration": 20, "start": 0},
+                    }
+                ],
+            )
+        except APIError:
+            return False
+        if data:
+            if "code" in data.keys():
+                if data["code"] == 0:
+                    self.light = True
+                    return True
+        return False
+
+    async def reboot(self) -> bool:
+        try:
+            data = await self.rpc.set_system_reboot()
+        except APIError:
+            return False
+        if data.get("msg"):
+            if data["msg"] == "ok":
+                return True
+        return False
+
+    async def restart_backend(self) -> bool:
+        try:
+            data = await self.rpc.set_miner_service("restart")
+        except APIError:
+            return False
+        if data.get("msg"):
+            if data["msg"] == "ok":
+                return True
+        return False
+
+    async def stop_mining(self) -> bool:
+        try:
+            data = await self.rpc.set_miner_service("stop")
+        except APIError:
+            return False
+        if data.get("msg"):
+            if data["msg"] == "ok":
+                return True
+        return False
+
+    async def resume_mining(self) -> bool:
+        try:
+            data = await self.rpc.set_miner_service("start")
+        except APIError:
+            return False
+        if data.get("msg"):
+            if data["msg"] == "ok":
+                return True
+        return False
+
+    async def set_power_limit(self, wattage: int) -> bool:
+        try:
+            await self.rpc.set_miner_power_limit(wattage)
+        except Exception as e:
+            logging.warning(f"{self} set_power_limit: {e}")
+            return False
+        else:
+            return True
+
     async def _get_mac(self, rpc_get_device_info: dict = None) -> str | None:
         if rpc_get_device_info is None:
             try:

pyasic/rpc/btminer.py

@@ -1256,8 +1256,7 @@ async def get_log_download(self) -> dict | None:
     async def get_fan_setting(self) -> dict | None:
         return await self.send_command("get.fan.setting")
 
-    async def set_system_reboot(self, *args, **kwargs) -> dict | None:
-        raise NotImplementedError
+    async def set_system_reboot(self) -> dict | None:
         return await self.send_command("set.system.reboot")
 
     async def set_system_factory_reset(self, *args, **kwargs) -> dict | None:
@@ -1272,8 +1271,11 @@ async def set_system_net_config(self, *args, **kwargs) -> dict | None:
         raise NotImplementedError
         return await self.send_command("set.system.net_config")
 
-    async def set_system_led(self, *args, **kwargs) -> dict | None:
-        return await self.send_command("set.system.led", "auto")
+    async def set_system_led(self, leds: list | None = None) -> dict | None:
+        if leds is None:
+            return await self.send_command("set.system.led", parameters="auto")
+        else:
+            return await self.send_command("set.system.led", parameters=leds)
 
     async def set_system_time_randomized(self, *args, **kwargs) -> dict | None:
         raise NotImplementedError
@@ -1303,9 +1305,8 @@ async def set_system_ntp_server(self, *args, **kwargs) -> dict | None:
         raise NotImplementedError
         return await self.send_command("set.system.ntp_server")
 
-    async def set_miner_service(self, *args, **kwargs) -> dict | None:
-        raise NotImplementedError
-        return await self.send_command("set.miner.service")
+    async def set_miner_service(self, value: str) -> dict | None:
+        return await self.send_command("set.miner.service", parameters=value)
 
     async def set_miner_power_mode(self, *args, **kwargs) -> dict | None:
         raise NotImplementedError
@@ -1340,9 +1341,8 @@ async def set_miner_report(self, frequency: int = 1) -> dict | None:
             "set.miner.report", parameters={"gap": frequency}
         )
 
-    async def set_miner_power_limit(self, *args, **kwargs) -> dict | None:
-        raise NotImplementedError
-        return await self.send_command("set.miner.power_limit")
+    async def set_miner_power_limit(self, power: int) -> dict | None:
+        return await self.send_command("set.miner.power_limit", parameters=power)
 
     async def set_miner_upfreq_speed(self, *args, **kwargs) -> dict | None:
         raise NotImplementedError