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

improved format and readability

Browse Source
This commit is contained in:
UpstreamData
2021-10-20 13:47:58 -06:00
parent 62ea4578a2
commit ea2ec6463e
14 changed files with 77 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
API/__init__.py
View File

@@ -18,18 +18,25 @@ class APIError(Exception):
class BaseMinerAPI:
def __init__(self, ip: str, port: int):
def __init__(self, ip: str, port: int) -> None:
self.port = port
self.ip = ipaddress.ip_address(ip)
def get_commands(self):
return [func for func in dir(self) if callable(getattr(self, func)) and not func.startswith("__") and func not in [func for func in dir(BaseMinerAPI) if callable(getattr(BaseMinerAPI, func))]]
def get_commands(self) -> list:
return [func for func in
dir(self) if callable(getattr(self, func)) and
not func.startswith("__") and
func not in
[func for func in
dir(BaseMinerAPI) if callable(getattr(BaseMinerAPI, func))
]
]
async def multicommand(self, *commands: str) -> dict:
command = "+".join(commands)
return await self.send_command(command)
async def send_command(self, command: str, parameters: str = None) -> dict:
async def send_command(self, command: str, parameters: str or int or bool = None) -> dict:
try:
# get reader and writer streams
reader, writer = await asyncio.open_connection(str(self.ip), self.port)

6
API/bmminer.py
View File

@@ -2,7 +2,7 @@ from API import BaseMinerAPI
class BMMinerAPI(BaseMinerAPI):
def __init__(self, ip, port=4028):
def __init__(self, ip: str, port: int = 4028) -> None:
super().__init__(ip, port)
async def version(self) -> dict:
@@ -42,7 +42,7 @@ class BMMinerAPI(BaseMinerAPI):
return await self.send_command("addpool", parameters=f"{url}, {username}, {password}")
async def poolpriority(self, *n: int) -> dict:
return await self.send_command("poolpriority", parameters=f"{','.join(n)}")
return await self.send_command("poolpriority", parameters=f"{','.join([str(item) for item in n])}")
async def poolquota(self, n: int, q: int) -> dict:
return await self.send_command("poolquota", parameters=f"{n}, {q}")
@@ -96,7 +96,7 @@ class BMMinerAPI(BaseMinerAPI):
return await self.send_command("check", parameters=command)
async def failover_only(self, failover: bool) -> dict:
return self.send_command("failover-only", parameters=failover)
return await self.send_command("failover-only", parameters=failover)
async def coin(self) -> dict:
return await self.send_command("coin")

10
API/bosminer.py
View File

@@ -49,27 +49,27 @@ class BOSMinerAPI(BaseMinerAPI):
async def switchpool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
return NotImplementedError
raise NotImplementedError
# return await self.send_command("switchpool", parameters=n)
async def enablepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
return NotImplementedError
raise NotImplementedError
# return await self.send_command("enablepool", parameters=n)
async def disablepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
return NotImplementedError
raise NotImplementedError
# return await self.send_command("disablepool", parameters=n)
async def addpool(self, url: str, username: str, password: str) -> dict:
# BOS has not implemented this yet, they will in the future
return NotImplementedError
raise NotImplementedError
# return await self.send_command("addpool", parameters=f"{url}, {username}, {password}")
async def removepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
return NotImplementedError
raise NotImplementedError
# return await self.send_command("removepool", parameters=n)
async def fans(self) -> dict:

4
API/cgminer.py
View File

@@ -39,7 +39,7 @@ class CGMinerAPI(BaseMinerAPI):
return await self.send_command("addpool", parameters=f"{url}, {username}, {password}")
async def poolpriority(self, *n: int) -> dict:
return await self.send_command("poolpriority", parameters=f"{','.join(n)}")
return await self.send_command("poolpriority", parameters=f"{','.join([str(item) for item in n])}")
async def poolquota(self, n: int, q: int) -> dict:
return await self.send_command("poolquota", parameters=f"{n}, {q}")
@@ -93,7 +93,7 @@ class CGMinerAPI(BaseMinerAPI):
return await self.send_command("check", parameters=command)
async def failover_only(self, failover: bool) -> dict:
return self.send_command("failover-only", parameters=failover)
return await self.send_command("failover-only", parameters=failover)
async def coin(self) -> dict:
return await self.send_command("coin")

10
API/unknown.py
View File

@@ -49,25 +49,25 @@ class UnknownAPI(BaseMinerAPI):
async def switchpool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
return NotImplementedError
raise NotImplementedError
# return await self.send_command("switchpool", parameters=n)
async def enablepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
return NotImplementedError
raise NotImplementedError
# return await self.send_command("enablepool", parameters=n)
async def disablepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
return NotImplementedError
raise NotImplementedError
# return await self.send_command("disablepool", parameters=n)
async def addpool(self, url: str, username: str, password: str) -> dict:
# BOS has not implemented this yet, they will in the future
return NotImplementedError
raise NotImplementedError
# return await self.send_command("addpool", parameters=f"{url}, {username}, {password}")
async def removepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
return NotImplementedError
raise NotImplementedError
# return await self.send_command("removepool", parameters=n)