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

improved error handling, made everything asynchronous, and created a quick test

Browse Source
This commit is contained in:
UpstreamData
2021-09-23 09:02:51 -06:00
parent 609e3e26a9
commit 400f816e17
1 changed files with 78 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
main.py
View File

@@ -17,7 +17,7 @@ class APIError(Exception):
class API: class API:
def __init__(self, port, ip): def __init__(self, ip, port):
self.port = port self.port = port
self.ip = ip self.ip = ip
@@ -45,116 +45,127 @@ class API:
writer.close() writer.close()
await writer.wait_closed() await writer.wait_closed()
# check if the data returned is correct or an error
if not data["STATUS"][0]["STATUS"] == "S":
# this is an error
raise APIError(data["STATUS"][0]["Msg"])
# return the data # return the data
return data return data
class CGMiner(API): class CGMiner(API):
def __init__(self, ip, port): def __init__(self, ip, port=4028):
super().__init__(ip, port) super().__init__(ip, port)
class BMMiner(API): class BMMiner(API):
def __init__(self, ip, port): def __init__(self, ip, port=4028):
super().__init__(ip, port) super().__init__(ip, port)
def version(self): async def version(self):
return self.send_command("version") return await self.send_command("version")
def config(self): async def config(self):
return self.send_command("config") return await self.send_command("config")
def summary(self): async def summary(self):
return self.send_command("summary") return await self.send_command("summary")
def pools(self): async def pools(self):
return self.send_command("pools") return await self.send_command("pools")
def devs(self): async def devs(self):
return self.send_command("devs") return await self.send_command("devs")
def edevs(self): async def edevs(self):
return self.send_command("edevs") return await self.send_command("edevs")
def pgacount(self): async def pgacount(self):
return self.send_command("pgacount") return await self.send_command("pgacount")
def notify(self): async def notify(self):
return self.send_command("notify") return await self.send_command("notify")
def devdetails(self): async def devdetails(self):
return self.send_command("devdetails") return await self.send_command("devdetails")
def stats(self): async def stats(self):
return self.send_command("stats") return await self.send_command("stats")
def estats(self): async def estats(self):
return self.send_command("estats") return await self.send_command("estats")
def check(self): async def check(self):
return self.send_command("check") return await self.send_command("check")
def coin(self): async def coin(self):
return self.send_command("coin") return await self.send_command("coin")
def asccount(self): async def asccount(self):
return self.send_command("asccount") return await self.send_command("asccount")
def lcd(self): async def lcd(self):
return self.send_command("lcd") return await self.send_command("lcd")
class BOSMiner(API): class BOSMiner(API):
def __init__(self, ip, port): def __init__(self, ip, port=4028):
super().__init__(ip, port) super().__init__(ip, port)
def asccount(self): async def asccount(self):
return self.send_command("asccount") return await self.send_command("asccount")
def devdetails(self): async def devdetails(self):
return self.send_command("devdetails") return await self.send_command("devdetails")
def devs(self): async def devs(self):
return self.send_command("devs") return await self.send_command("devs")
def edevs(self): async def edevs(self):
return self.send_command("edevs") return await self.send_command("edevs")
def pools(self): async def pools(self):
return self.send_command("pools") return await self.send_command("pools")
def summary(self): async def summary(self):
return self.send_command("summary") return await self.send_command("summary")
def stats(self): async def stats(self):
return self.send_command("stats") return await self.send_command("stats")
def version(self): async def version(self):
return self.send_command("version") return await self.send_command("version")
def estats(self): async def estats(self):
return self.send_command("estats") return await self.send_command("estats")
def check(self): async def check(self):
return self.send_command("check") return await self.send_command("check")
def coin(self): async def coin(self):
return self.send_command("coin") return await self.send_command("coin")
def lcd(self): async def lcd(self):
return self.send_command("lcd") return await self.send_command("lcd")
def fans(self): async def fans(self):
return self.send_command("fans") return await self.send_command("fans")
def tempctrl(self): async def tempctrl(self):
return self.send_command("tempctrl") return await self.send_command("tempctrl")
def temps(self): async def temps(self):
return self.send_command("temps") return await self.send_command("temps")
def tunerstatus(self): async def tunerstatus(self):
return self.send_command("tunerstatus") return await self.send_command("tunerstatus")
raise APIError async def main():
bosminer = BOSMiner("172.16.1.199")
data = await bosminer.stats()
print(data)
asyncio.get_event_loop().run_until_complete(main())