GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2026-01-29 00:51:53 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

common component interface

Browse Source
This commit is contained in:
Devaev Maxim
2020-06-06 06:29:29 +03:00
parent 4f3ebf0fd1
commit 1d7d4100a5
7 changed files with 60 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
kvmd/plugins/atx/__init__.py
View File

@@ -47,7 +47,7 @@ class AtxIsBusyError(IsBusyError, AtxError):
# =====
class BaseAtx(BasePlugin):
def get_state(self) -> Dict:
async def get_state(self) -> Dict:
raise NotImplementedError
async def poll_state(self) -> AsyncGenerator[Dict, None]:

4
kvmd/plugins/atx/disabled.py
View File

@@ -37,7 +37,7 @@ class AtxDisabledError(AtxOperationError):
# =====
class Plugin(BaseAtx):
def get_state(self) -> Dict:
async def get_state(self) -> Dict:
return {
"enabled": False,
"busy": False,
@@ -49,7 +49,7 @@ class Plugin(BaseAtx):
async def poll_state(self) -> AsyncGenerator[Dict, None]:
while True:
yield self.get_state()
yield (await self.get_state())
await aiotools.wait_infinite()
# =====

15
kvmd/plugins/atx/gpio.py
View File

@@ -92,7 +92,7 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
"state_poll": Option(0.1, type=valid_float_f01),
}
def get_state(self) -> Dict:
async def get_state(self) -> Dict:
return {
"enabled": True,
"busy": self.__region.is_busy(),
@@ -105,7 +105,7 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
async def poll_state(self) -> AsyncGenerator[Dict, None]:
prev_state: Dict = {}
while True:
state = self.get_state()
state = await self.get_state()
if state != prev_state:
yield state
prev_state = state
@@ -124,25 +124,25 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
# =====
async def power_on(self) -> bool:
if not self.get_state()["leds"]["power"]:
if not (await self.__get_power()):
await self.click_power()
return True
return False
async def power_off(self) -> bool:
if self.get_state()["leds"]["power"]:
if (await self.__get_power()):
await self.click_power()
return True
return False
async def power_off_hard(self) -> bool:
if self.get_state()["leds"]["power"]:
if (await self.__get_power()):
await self.click_power_long()
return True
return False
async def power_reset_hard(self) -> bool:
if self.get_state()["leds"]["power"]:
if (await self.__get_power()):
await self.click_reset()
return True
return False
@@ -160,6 +160,9 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
# =====
async def __get_power(self) -> bool:
return (await self.get_state())["leds"]["power"]
@aiotools.atomic
async def __click(self, name: str, pin: int, delay: float) -> None:
await aiotools.run_region_task(