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

added more type hinting for returns

Browse Source
This commit is contained in:
UpstreamData
2021-09-23 12:09:17 -06:00
parent 5ba768d0bc
commit e3bc84455d
4 changed files with 111 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
API/__init__.py
View File

@@ -21,11 +21,11 @@ class BaseMinerAPI:
self.port = port self.port = port
self.ip = ip self.ip = ip
async def multicommand(self, *commands: str): async def multicommand(self, *commands: str) -> dict:
command = "+".join(commands) command = "+".join(commands)
return await self.send_command(command) return await self.send_command(command)
async def send_command(self, command, parameters: dict = None): async def send_command(self, command, parameters: dict = None) -> dict:
# get reader and writer streams # get reader and writer streams
reader, writer = await asyncio.open_connection(self.ip, self.port) reader, writer = await asyncio.open_connection(self.ip, self.port)

86
API/bmminer.py
View File

@@ -5,146 +5,146 @@ class BMMinerAPI(BaseMinerAPI):
def __init__(self, ip, port=4028): def __init__(self, ip, port=4028):
super().__init__(ip, port) super().__init__(ip, port)
async def version(self): async def version(self) -> dict:
return await self.send_command("version") return await self.send_command("version")
async def config(self): async def config(self) -> dict:
return await self.send_command("config") return await self.send_command("config")
async def summary(self): async def summary(self) -> dict:
return await self.send_command("summary") return await self.send_command("summary")
async def pools(self): async def pools(self) -> dict:
return await self.send_command("pools") return await self.send_command("pools")
async def devs(self): async def devs(self) -> dict:
return await self.send_command("devs") return await self.send_command("devs")
async def edevs(self, old: bool = False): async def edevs(self, old: bool = False) -> dict:
if old: if old:
return await self.send_command("edevs", parameters="old") return await self.send_command("edevs", parameters="old")
else: else:
return await self.send_command("edevs") return await self.send_command("edevs")
async def pga(self, n: int): async def pga(self, n: int) -> dict:
return await self.send_command("pga", parameters=n) return await self.send_command("pga", parameters=n)
async def pgacount(self): async def pgacount(self) -> dict:
return await self.send_command("pgacount") return await self.send_command("pgacount")
async def switchpool(self, n: int): async def switchpool(self, n: int) -> dict:
return await self.send_command("switchpool", parameters=n) return await self.send_command("switchpool", parameters=n)
async def enablepool(self, n: int): async def enablepool(self, n: int) -> dict:
return await self.send_command("enablepool", parameters=n) return await self.send_command("enablepool", parameters=n)
async def addpool(self, url: str, username: str, password: str): async def addpool(self, url: str, username: str, password: str) -> dict:
return await self.send_command("addpool", parameters=f"{url}, {username}, {password}") return await self.send_command("addpool", parameters=f"{url}, {username}, {password}")
async def poolpriority(self, *n: int): 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(n)}")
async def poolquota(self, n: int, q: int): async def poolquota(self, n: int, q: int) -> dict:
return await self.send_command("poolquota", parameters=f"{n}, {q}") return await self.send_command("poolquota", parameters=f"{n}, {q}")
async def disablepool(self, n: int): async def disablepool(self, n: int) -> dict:
return await self.send_command("disablepool", parameters=n) return await self.send_command("disablepool", parameters=n)
async def removepool(self, n: int): async def removepool(self, n: int) -> dict:
return await self.send_command("removepool", parameters=n) return await self.send_command("removepool", parameters=n)
async def save(self, filename: str = None): async def save(self, filename: str = None) -> dict:
if filename: if filename:
return await self.send_command("save", parameters=filename) return await self.send_command("save", parameters=filename)
else: else:
return await self.send_command("save") return await self.send_command("save")
async def quit(self): async def quit(self) -> dict:
return await self.send_command("quit") return await self.send_command("quit")
async def notify(self): async def notify(self) -> dict:
return await self.send_command("notify") return await self.send_command("notify")
async def privileged(self): async def privileged(self) -> dict:
return await self.send_command("privileged") return await self.send_command("privileged")
async def pgaenable(self, n: int): async def pgaenable(self, n: int) -> dict:
return await self.send_command("pgaenable", parameters=n) return await self.send_command("pgaenable", parameters=n)
async def pgadisable(self, n: int): async def pgadisable(self, n: int) -> dict:
return await self.send_command("pgadisable", parameters=n) return await self.send_command("pgadisable", parameters=n)
async def pgaidentify(self, n: int): async def pgaidentify(self, n: int) -> dict:
return await self.send_command("pgaidentify", parameters=n) return await self.send_command("pgaidentify", parameters=n)
async def devdetails(self): async def devdetails(self) -> dict:
return await self.send_command("devdetails") return await self.send_command("devdetails")
async def restart(self): async def restart(self) -> dict:
return await self.send_command("restart") return await self.send_command("restart")
async def stats(self): async def stats(self) -> dict:
return await self.send_command("stats") return await self.send_command("stats")
async def estats(self, old: bool = False): async def estats(self, old: bool = False) -> dict:
if old: if old:
return await self.send_command("estats", parameters="old") return await self.send_command("estats", parameters="old")
else: else:
return await self.send_command("estats") return await self.send_command("estats")
async def check(self, command): async def check(self, command) -> dict:
return await self.send_command("check", parameters=command) return await self.send_command("check", parameters=command)
async def failover_only(self, failover: bool): async def failover_only(self, failover: bool) -> dict:
return self.send_command("failover-only", parameters=failover) return self.send_command("failover-only", parameters=failover)
async def coin(self): async def coin(self) -> dict:
return await self.send_command("coin") return await self.send_command("coin")
async def debug(self, setting: str): async def debug(self, setting: str) -> dict:
return await self.send_command("debug", parameters=setting) return await self.send_command("debug", parameters=setting)
async def setconfig(self, name: str, n: int): async def setconfig(self, name: str, n: int) -> dict:
return await self.send_command("setconfig", parameters=f"{name}, {n}") return await self.send_command("setconfig", parameters=f"{name}, {n}")
async def usbstats(self): async def usbstats(self) -> dict:
return await self.send_command("usbstats") return await self.send_command("usbstats")
async def pgaset(self, n: int, opt: str, val: int = None): async def pgaset(self, n: int, opt: str, val: int = None) -> dict:
if val: if val:
return await self.send_command("pgaset", parameters=f"{n}, {opt}, {val}") return await self.send_command("pgaset", parameters=f"{n}, {opt}, {val}")
else: else:
return await self.send_command("pgaset", parameters=f"{n}, {opt}") return await self.send_command("pgaset", parameters=f"{n}, {opt}")
async def zero(self, which: str, value: bool): async def zero(self, which: str, value: bool) -> dict:
return await self.send_command("zero", parameters=f"{which}, {value}") return await self.send_command("zero", parameters=f"{which}, {value}")
async def hotplug(self, n: int): async def hotplug(self, n: int) -> dict:
return await self.send_command("hotplug", parameters=n) return await self.send_command("hotplug", parameters=n)
async def asc(self, n: int): async def asc(self, n: int) -> dict:
return await self.send_command("asc", parameters=n) return await self.send_command("asc", parameters=n)
async def ascenable(self, n: int): async def ascenable(self, n: int) -> dict:
return await self.send_command("ascenable", parameters=n) return await self.send_command("ascenable", parameters=n)
async def ascdisable(self, n: int): async def ascdisable(self, n: int) -> dict:
return await self.send_command("ascdisable", parameters=n) return await self.send_command("ascdisable", parameters=n)
async def ascidentify(self, n: int): async def ascidentify(self, n: int) -> dict:
return await self.send_command("ascidentify", parameters=n) return await self.send_command("ascidentify", parameters=n)
async def asccount(self): async def asccount(self) -> dict:
return await self.send_command("asccount") return await self.send_command("asccount")
async def ascset(self, n: int, opt: str, val: int = None): async def ascset(self, n: int, opt: str, val: int = None) -> dict:
if val: if val:
return await self.send_command("ascset", parameters=f"{n}, {opt}, {val}") return await self.send_command("ascset", parameters=f"{n}, {opt}, {val}")
else: else:
return await self.send_command("ascset", parameters=f"{n}, {opt}") return await self.send_command("ascset", parameters=f"{n}, {opt}")
async def lcd(self): async def lcd(self) -> dict:
return await self.send_command("lcd") return await self.send_command("lcd")
async def lockstats(self): async def lockstats(self) -> dict:
return await self.send_command("lockstats") return await self.send_command("lockstats")

48
API/bosminer.py
View File

@@ -5,77 +5,77 @@ class BOSMinerAPI(BaseMinerAPI):
def __init__(self, ip, port=4028): def __init__(self, ip, port=4028):
super().__init__(ip, port) super().__init__(ip, port)
async def asccount(self): async def asccount(self) -> dict:
return await self.send_command("asccount") return await self.send_command("asccount")
async def asc(self, n: int): async def asc(self, n: int) -> dict:
return await self.send_command("asc", parameters=n) return await self.send_command("asc", parameters=n)
async def devdetails(self): async def devdetails(self) -> dict:
return await self.send_command("devdetails") return await self.send_command("devdetails")
async def devs(self): async def devs(self) -> dict:
return await self.send_command("devs") return await self.send_command("devs")
async def edevs(self, old: bool = False): async def edevs(self, old: bool = False) -> dict:
if old: if old:
return await self.send_command("edevs", parameters="old") return await self.send_command("edevs", parameters="old")
else: else:
return await self.send_command("edevs") return await self.send_command("edevs")
async def pools(self): async def pools(self) -> dict:
return await self.send_command("pools") return await self.send_command("pools")
async def summary(self): async def summary(self) -> dict:
return await self.send_command("summary") return await self.send_command("summary")
async def stats(self): async def stats(self) -> dict:
return await self.send_command("stats") return await self.send_command("stats")
async def version(self): async def version(self) -> dict:
return await self.send_command("version") return await self.send_command("version")
async def estats(self): async def estats(self) -> dict:
return await self.send_command("estats") return await self.send_command("estats")
async def check(self): async def check(self) -> dict:
return await self.send_command("check") return await self.send_command("check")
async def coin(self): async def coin(self) -> dict:
return await self.send_command("coin") return await self.send_command("coin")
async def lcd(self): async def lcd(self) -> dict:
return await self.send_command("lcd") return await self.send_command("lcd")
async def switchpool(self, n: int): async def switchpool(self, n: int) -> dict:
return await self.send_command("switchpool", parameters=n) return await self.send_command("switchpool", parameters=n)
async def enablepool(self, n: int): async def enablepool(self, n: int) -> dict:
return await self.send_command("enablepool", parameters=n) return await self.send_command("enablepool", parameters=n)
async def disablepool(self, n: int): async def disablepool(self, n: int) -> dict:
return await self.send_command("disablepool", parameters=n) return await self.send_command("disablepool", parameters=n)
async def addpool(self, url: str, username: str, password: str): async def addpool(self, url: str, username: str, password: str) -> dict:
return await self.send_command("addpool", parameters=f"{url}, {username}, {password}") return await self.send_command("addpool", parameters=f"{url}, {username}, {password}")
async def removepool(self, n: int): async def removepool(self, n: int) -> dict:
return await self.send_command("removepool", parameters=n) return await self.send_command("removepool", parameters=n)
async def fans(self): async def fans(self) -> dict:
return await self.send_command("fans") return await self.send_command("fans")
async def tempctrl(self): async def tempctrl(self) -> dict:
return await self.send_command("tempctrl") return await self.send_command("tempctrl")
async def temps(self): async def temps(self) -> dict:
return await self.send_command("temps") return await self.send_command("temps")
async def tunerstatus(self): async def tunerstatus(self) -> dict:
return await self.send_command("tunerstatus") return await self.send_command("tunerstatus")
async def pause(self): async def pause(self) -> dict:
return await self.send_command("pause") return await self.send_command("pause")
async def resume(self): async def resume(self) -> dict:
return await self.send_command("resume") return await self.send_command("resume")

84
API/cgminer.py
View File

@@ -5,143 +5,143 @@ class CGMinerAPI(BaseMinerAPI):
def __init__(self, ip, port=4028): def __init__(self, ip, port=4028):
super().__init__(ip, port) super().__init__(ip, port)
async def version(self): async def version(self) -> dict:
return await self.send_command("version") return await self.send_command("version")
async def summary(self): async def summary(self) -> dict:
return await self.send_command("summary") return await self.send_command("summary")
async def pools(self): async def pools(self) -> dict:
return await self.send_command("pools") return await self.send_command("pools")
async def devs(self): async def devs(self) -> dict:
return await self.send_command("devs") return await self.send_command("devs")
async def edevs(self, old: bool = False): async def edevs(self, old: bool = False) -> dict:
if old: if old:
return await self.send_command("edevs", parameters="old") return await self.send_command("edevs", parameters="old")
else: else:
return await self.send_command("edevs") return await self.send_command("edevs")
async def pga(self, n: int): async def pga(self, n: int) -> dict:
return await self.send_command("pga", parameters=n) return await self.send_command("pga", parameters=n)
async def pgacount(self): async def pgacount(self) -> dict:
return await self.send_command("pgacount") return await self.send_command("pgacount")
async def switchpool(self, n: int): async def switchpool(self, n: int) -> dict:
return await self.send_command("switchpool", parameters=n) return await self.send_command("switchpool", parameters=n)
async def enablepool(self, n: int): async def enablepool(self, n: int) -> dict:
return await self.send_command("enablepool", parameters=n) return await self.send_command("enablepool", parameters=n)
async def addpool(self, url: str, username: str, password: str): async def addpool(self, url: str, username: str, password: str) -> dict:
return await self.send_command("addpool", parameters=f"{url}, {username}, {password}") return await self.send_command("addpool", parameters=f"{url}, {username}, {password}")
async def poolpriority(self, *n: int): 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(n)}")
async def poolquota(self, n: int, q: int): async def poolquota(self, n: int, q: int) -> dict:
return await self.send_command("poolquota", parameters=f"{n}, {q}") return await self.send_command("poolquota", parameters=f"{n}, {q}")
async def disablepool(self, n: int): async def disablepool(self, n: int) -> dict:
return await self.send_command("disablepool", parameters=n) return await self.send_command("disablepool", parameters=n)
async def removepool(self, n: int): async def removepool(self, n: int) -> dict:
return await self.send_command("removepool", parameters=n) return await self.send_command("removepool", parameters=n)
async def save(self, filename: str = None): async def save(self, filename: str = None) -> dict:
if filename: if filename:
return await self.send_command("save", parameters=filename) return await self.send_command("save", parameters=filename)
else: else:
return await self.send_command("save") return await self.send_command("save")
async def quit(self): async def quit(self) -> dict:
return await self.send_command("quit") return await self.send_command("quit")
async def notify(self): async def notify(self) -> dict:
return await self.send_command("notify") return await self.send_command("notify")
async def privileged(self): async def privileged(self) -> dict:
return await self.send_command("privileged") return await self.send_command("privileged")
async def pgaenable(self, n: int): async def pgaenable(self, n: int) -> dict:
return await self.send_command("pgaenable", parameters=n) return await self.send_command("pgaenable", parameters=n)
async def pgadisable(self, n: int): async def pgadisable(self, n: int) -> dict:
return await self.send_command("pgadisable", parameters=n) return await self.send_command("pgadisable", parameters=n)
async def pgaidentify(self, n: int): async def pgaidentify(self, n: int) -> dict:
return await self.send_command("pgaidentify", parameters=n) return await self.send_command("pgaidentify", parameters=n)
async def devdetails(self): async def devdetails(self) -> dict:
return await self.send_command("devdetails") return await self.send_command("devdetails")
async def restart(self): async def restart(self) -> dict:
return await self.send_command("restart") return await self.send_command("restart")
async def stats(self): async def stats(self) -> dict:
return await self.send_command("stats") return await self.send_command("stats")
async def estats(self, old: bool = False): async def estats(self, old: bool = False) -> dict:
if old: if old:
return await self.send_command("estats", parameters="old") return await self.send_command("estats", parameters="old")
else: else:
return await self.send_command("estats") return await self.send_command("estats")
async def check(self, command): async def check(self, command) -> dict:
return await self.send_command("check", parameters=command) return await self.send_command("check", parameters=command)
async def failover_only(self, failover: bool): async def failover_only(self, failover: bool) -> dict:
return self.send_command("failover-only", parameters=failover) return self.send_command("failover-only", parameters=failover)
async def coin(self): async def coin(self) -> dict:
return await self.send_command("coin") return await self.send_command("coin")
async def debug(self, setting: str): async def debug(self, setting: str) -> dict:
return await self.send_command("debug", parameters=setting) return await self.send_command("debug", parameters=setting)
async def setconfig(self, name: str, n: int): async def setconfig(self, name: str, n: int) -> dict:
return await self.send_command("setconfig", parameters=f"{name}, {n}") return await self.send_command("setconfig", parameters=f"{name}, {n}")
async def usbstats(self): async def usbstats(self) -> dict:
return await self.send_command("usbstats") return await self.send_command("usbstats")
async def pgaset(self, n: int, opt: str, val: int = None): async def pgaset(self, n: int, opt: str, val: int = None) -> dict:
if val: if val:
return await self.send_command("pgaset", parameters=f"{n}, {opt}, {val}") return await self.send_command("pgaset", parameters=f"{n}, {opt}, {val}")
else: else:
return await self.send_command("pgaset", parameters=f"{n}, {opt}") return await self.send_command("pgaset", parameters=f"{n}, {opt}")
async def zero(self, which: str, value: bool): async def zero(self, which: str, value: bool) -> dict:
return await self.send_command("zero", parameters=f"{which}, {value}") return await self.send_command("zero", parameters=f"{which}, {value}")
async def hotplug(self, n: int): async def hotplug(self, n: int) -> dict:
return await self.send_command("hotplug", parameters=n) return await self.send_command("hotplug", parameters=n)
async def asc(self, n: int): async def asc(self, n: int) -> dict:
return await self.send_command("asc", parameters=n) return await self.send_command("asc", parameters=n)
async def ascenable(self, n: int): async def ascenable(self, n: int) -> dict:
return await self.send_command("ascenable", parameters=n) return await self.send_command("ascenable", parameters=n)
async def ascdisable(self, n: int): async def ascdisable(self, n: int) -> dict:
return await self.send_command("ascdisable", parameters=n) return await self.send_command("ascdisable", parameters=n)
async def ascidentify(self, n: int): async def ascidentify(self, n: int) -> dict:
return await self.send_command("ascidentify", parameters=n) return await self.send_command("ascidentify", parameters=n)
async def asccount(self): async def asccount(self) -> dict:
return await self.send_command("asccount") return await self.send_command("asccount")
async def ascset(self, n: int, opt: str, val: int = None): async def ascset(self, n: int, opt: str, val: int = None) -> dict:
if val: if val:
return await self.send_command("ascset", parameters=f"{n}, {opt}, {val}") return await self.send_command("ascset", parameters=f"{n}, {opt}, {val}")
else: else:
return await self.send_command("ascset", parameters=f"{n}, {opt}") return await self.send_command("ascset", parameters=f"{n}, {opt}")
async def lcd(self): async def lcd(self) -> dict:
return await self.send_command("lcd") return await self.send_command("lcd")
async def lockstats(self): async def lockstats(self) -> dict:
return await self.send_command("lockstats") return await self.send_command("lockstats")