GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2025-12-13 01:30:31 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

async gpio plugins

Browse Source
This commit is contained in:
Devaev Maxim 2021-04-09 05:57:04 +03:00
parent 6cc161427a
commit 6f60118320
6 changed files with 23 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
kvmd/apps/kvmd/ugpio.py
View File

@ -91,10 +91,10 @@ class _GpioInput:
},
}
def get_state(self) -> Dict:
async def get_state(self) -> Dict:
(online, state) = (True, False)
try:
state = (self.__driver.read(self.__pin) ^ self.__inverted)
state = (await self.__driver.read(self.__pin) ^ self.__inverted)
except GpioDriverOfflineError:
online = False
return {
@ -153,12 +153,12 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes
},
}
def get_state(self) -> Dict:
async def get_state(self) -> Dict:
busy = self.__region.is_busy()
(online, state) = (True, False)
if not busy:
try:
state = self.__read()
state = await self.__read()
except GpioDriverOfflineError:
online = False
return {
@ -201,27 +201,27 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes
@aiotools.atomic
async def __inner_switch(self, state: bool) -> None:
self.__write(state)
await self.__write(state)
get_logger(0).info("Ensured switch %s to state=%d", self, state)
await asyncio.sleep(self.__busy_delay)
@aiotools.atomic
async def __inner_pulse(self, delay: float) -> None:
try:
self.__write(True)
await self.__write(True)
await asyncio.sleep(delay)
finally:
self.__write(False)
await self.__write(False)
await asyncio.sleep(self.__busy_delay)
get_logger(0).info("Pulsed %s with delay=%.2f", self, delay)
# =====
def __read(self) -> bool:
return (self.__driver.read(self.__pin) ^ self.__inverted)
async def __read(self) -> bool:
return (await self.__driver.read(self.__pin) ^ self.__inverted)
def __write(self, state: bool) -> None:
self.__driver.write(self.__pin, (state ^ self.__inverted))
async def __write(self, state: bool) -> None:
await self.__driver.write(self.__pin, (state ^ self.__inverted))
def __str__(self) -> str:
return f"Output({self.__channel}, driver={self.__driver}, pin={self.__pin})"
@ -267,8 +267,8 @@ class UserGpio:
async def get_state(self) -> Dict:
return {
"inputs": {channel: gin.get_state() for (channel, gin) in self.__inputs.items()},
"outputs": {channel: gout.get_state() for (channel, gout) in self.__outputs.items()},
"inputs": {channel: await gin.get_state() for (channel, gin) in self.__inputs.items()},
"outputs": {channel: await gout.get_state() for (channel, gout) in self.__outputs.items()},
}
async def poll_state(self) -> AsyncGenerator[Dict, None]:

4
kvmd/plugins/ugpio/__init__.py
View File

@ -89,10 +89,10 @@ class BaseUserGpioDriver(BasePlugin):
def cleanup(self) -> None:
raise NotImplementedError
def read(self, pin: int) -> bool:
async def read(self, pin: int) -> bool:
raise NotImplementedError
def write(self, pin: int, state: bool) -> None:
async def write(self, pin: int, state: bool) -> None:
raise NotImplementedError

4
kvmd/plugins/ugpio/ezcoo.py
View File

@ -108,12 +108,12 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
if self.__proc.exitcode is not None:
self.__proc.join()
def read(self, pin: int) -> bool:
async def read(self, pin: int) -> bool:
if not self.__is_online():
raise GpioDriverOfflineError(self)
return (self.__channel == pin)
def write(self, pin: int, state: bool) -> None:
async def write(self, pin: int, state: bool) -> None:
if not self.__is_online():
raise GpioDriverOfflineError(self)
if state and (0 <= pin <= 3):

4
kvmd/plugins/ugpio/gpio.py
View File

@ -95,13 +95,13 @@ class Plugin(BaseUserGpioDriver):
except Exception:
pass
def read(self, pin: int) -> bool:
async def read(self, pin: int) -> bool:
assert self.__reader
if pin in self.__input_pins:
return self.__reader.get(pin)
return bool(self.__output_lines[pin].get_value())
def write(self, pin: int, state: bool) -> None:
async def write(self, pin: int, state: bool) -> None:
self.__output_lines[pin].set_value(int(state))
def __str__(self) -> str:

4
kvmd/plugins/ugpio/hidrelay.py
View File

@ -113,13 +113,13 @@ class Plugin(BaseUserGpioDriver):
self.__close_device()
self.__stop = True
def read(self, pin: int) -> bool:
async def read(self, pin: int) -> bool:
try:
return self.__inner_read(pin)
except Exception:
raise GpioDriverOfflineError(self)
def write(self, pin: int, state: bool) -> None:
async def write(self, pin: int, state: bool) -> None:
try:
return self.__inner_write(pin, state)
except Exception:

4
kvmd/plugins/ugpio/otgbind.py
View File

@ -100,11 +100,11 @@ class Plugin(BaseUserGpioDriver):
def cleanup(self) -> None:
pass
def read(self, pin: int) -> bool:
async def read(self, pin: int) -> bool:
_ = pin
return os.path.islink(self.__get_driver_path(self.__udc))
def write(self, pin: int, state: bool) -> None:
async def write(self, pin: int, state: bool) -> None:
_ = pin
with open(self.__get_driver_path("bind" if state else "unbind"), "w") as ctl_file:
ctl_file.write(f"{self.__udc}\n")