CAT_TOM/pyasic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature: start refactoring BOSer and BOSMiner into separate classes.

Browse Source
This commit is contained in:
b-rowan
2024-01-10 22:12:27 -07:00
committed by UpstreamData
parent 672e753afb
commit 928e0dd028
5 changed files with 675 additions and 338 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
pyasic/web/braiins_os/__init__.py
View File

@@ -84,29 +84,31 @@ class BOSerWebAPI(BOSMinerWebAPI):
**parameters: Union[str, int, bool],
) -> dict:
command_type = self.select_command_type(command)
if command_type == "gql":
if command_type is "gql":
return await self.gql.send_command(command)
elif command_type == "grpc":
elif command_type is "grpc":
try:
return await (getattr(self.grpc, command.replace("grpc_", "")))()
except AttributeError:
raise APIError(f"No gRPC command found for command: {command}")
elif command_type == "luci":
elif command_type is "luci":
return await self.luci.send_command(command)
@staticmethod
def select_command_type(command: Union[str, dict]) -> str:
if isinstance(command, dict):
return "gql"
else:
elif command.startswith("grpc_"):
return "grpc"
else:
return "luci"
async def multicommand(
self, *commands: Union[dict, str], allow_warning: bool = True
) -> dict:
cmd_types = {"grpc": [], "gql": []}
cmd_types = {"grpc": [], "gql": [], "luci": []}
for cmd in commands:
cmd_types[self.select_command_type(cmd)].append(cmd)
cmd_types[self.select_command_type(cmd)] = cmd
async def no_op():
return {}
@@ -116,13 +118,21 @@ class BOSerWebAPI(BOSMinerWebAPI):
self.grpc.multicommand(*cmd_types["grpc"])
)
else:
grpc_data_t = asyncio.create_task(no_op())
grpc_data_t = no_op()
if len(cmd_types["gql"]) > 0:
gql_data_t = asyncio.create_task(self.gql.multicommand(*cmd_types["gql"]))
else:
gql_data_t = asyncio.create_task(no_op())
gql_data_t = no_op()
if len(cmd_types["luci"]) > 0:
luci_data_t = asyncio.create_task(
self.luci.multicommand(*cmd_types["luci"])
)
else:
luci_data_t = no_op()
await asyncio.gather(grpc_data_t, gql_data_t)
await asyncio.gather(grpc_data_t, gql_data_t, luci_data_t)
data = dict(**grpc_data_t.result(), **gql_data_t.result())
data = dict(
**luci_data_t.result(), **gql_data_t.result(), **luci_data_t.result()
)
return data

125
pyasic/web/braiins_os/grpc.py
View File

@@ -13,12 +13,9 @@
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
import asyncio
import logging
from datetime import timedelta
from betterproto import Message
from grpclib import GRPCError, Status
from grpclib.client import Channel
from pyasic.errors import APIError
@@ -46,7 +43,6 @@ class BOSerGRPCAPI:
self.ip = ip
self.username = "root"
self.pwd = pwd
self.port = 50051
self._auth = None
self._auth_time = datetime.now()
@@ -68,20 +64,7 @@ class BOSerGRPCAPI:
]
async def multicommand(self, *commands: str) -> dict:
result = {"multicommand": True}
tasks = {}
for command in commands:
try:
tasks[command] = asyncio.create_task(getattr(self, command)())
except AttributeError:
result["command"] = {}
await asyncio.gather(*list(tasks.values()))
for cmd in tasks:
result[cmd] = tasks[cmd].result()
return result
pass
async def send_command(
self,
@@ -93,23 +76,13 @@ class BOSerGRPCAPI:
metadata = []
if auth:
metadata.append(("authorization", await self.auth()))
try:
async with Channel(self.ip, self.port) as c:
endpoint = getattr(BOSMinerGRPCStub(c), command)
if endpoint is None:
if not ignore_errors:
raise APIError(f"Command not found - {endpoint}")
return {}
try:
return (await endpoint(message, metadata=metadata)).to_pydict()
except GRPCError as e:
if e.status == Status.UNAUTHENTICATED:
await self._get_auth()
metadata = [("authorization", await self.auth())]
return (await endpoint(message, metadata=metadata)).to_pydict()
raise e
except GRPCError as e:
raise APIError(f"gRPC command failed - {endpoint}") from e
async with Channel(self.ip, 50051) as c:
endpoint = getattr(BOSMinerGRPCStub(c), command)
if endpoint is None:
if not ignore_errors:
raise APIError(f"Command not found - {endpoint}")
return {}
return (await endpoint(message, metadata=metadata)).to_pydict()
async def auth(self):
if self._auth is not None and self._auth_time - datetime.now() < timedelta(
@@ -120,7 +93,7 @@ class BOSerGRPCAPI:
return self._auth
async def _get_auth(self):
async with Channel(self.ip, self.port) as c:
async with Channel(self.ip, 50051) as c:
req = LoginRequest(username=self.username, password=self.pwd)
async with c.request(
"/braiins.bos.v1.AuthenticationService/Login",
@@ -165,9 +138,7 @@ class BOSerGRPCAPI:
)
async def get_locate_device_status(self):
return await self.send_command(
"get_locate_device_status", GetLocateDeviceStatusRequest()
)
return await self.send_command("get_locate_device_status")
async def set_password(self, password: str = None):
return await self.send_command(
@@ -190,12 +161,10 @@ class BOSerGRPCAPI:
)
async def get_tuner_state(self):
return await self.send_command("get_tuner_state", GetTunerStateRequest())
return await self.send_command("get_tuner_state")
async def list_target_profiles(self):
return await self.send_command(
"list_target_profiles", ListTargetProfilesRequest()
)
return await self.send_command("list_target_profiles")
async def set_default_power_target(
self, save_action: SaveAction = SaveAction.SAVE_ACTION_SAVE_AND_APPLY
@@ -296,71 +265,15 @@ class BOSerGRPCAPI:
async def set_dps(
self,
enable: bool,
power_step: int,
min_power_target: int,
enable_shutdown: bool = None,
shutdown_duration: int = None,
):
return await self.send_command(
"set_dps",
message=SetDpsRequest(
enable=enable,
enable_shutdown=enable_shutdown,
shutdown_duration=shutdown_duration,
target=DpsTarget(
power_target=DpsPowerTarget(
power_step=Power(power_step),
min_power_target=Power(min_power_target),
)
),
),
)
raise NotImplementedError
return await self.send_command("braiins.bos.v1.PerformanceService/SetDPS")
async def set_performance_mode(
self,
wattage_target: int = None,
hashrate_target: int = None,
save_action: SaveAction = SaveAction.SAVE_ACTION_SAVE_AND_APPLY,
):
if wattage_target is not None and hashrate_target is not None:
logging.error(
"Cannot use both wattage_target and hashrate_target, using wattage_target."
)
elif wattage_target is None and hashrate_target is None:
raise APIError(
"No target supplied, please supply either wattage_target or hashrate_target."
)
if wattage_target is not None:
return await self.send_command(
"set_performance_mode",
message=SetPerformanceModeRequest(
save_action=save_action,
mode=PerformanceMode(
tuner_mode=TunerPerformanceMode(
power_target=PowerTargetMode(
power_target=Power(watt=wattage_target)
)
)
),
),
)
if hashrate_target is not None:
return await self.send_command(
"set_performance_mode",
message=SetPerformanceModeRequest(
save_action=save_action,
mode=PerformanceMode(
tuner_mode=TunerPerformanceMode(
hashrate_target=HashrateTargetMode(
hashrate_target=TeraHashrate(
terahash_per_second=hashrate_target
)
)
)
),
),
)
async def set_performance_mode(self):
raise NotImplementedError
return await self.send_command(
"braiins.bos.v1.PerformanceService/SetPerformanceMode"
)
async def get_active_performance_mode(self):
return await self.send_command(