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

atx: independent power/reset regions

Browse Source
This commit is contained in:
Maxim Devaev 2025-07-14 18:15:39 +03:00
parent 49638ed896
commit d57c3c66cd
3 changed files with 21 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -54,7 +54,8 @@ class BaseAtx(BasePlugin):
async def poll_state(self) -> AsyncGenerator[dict, None]: async def poll_state(self) -> AsyncGenerator[dict, None]:
# ==== Granularity table ==== # ==== Granularity table ====
# - enabled -- Full # - enabled -- Full
# - busy -- Partial # - busy -- Partial, follows with acts
# - acts -- Partial, follows with busy
# - leds -- Partial # - leds -- Partial
# =========================== # ===========================

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

@ -43,6 +43,10 @@ class Plugin(BaseAtx):
return { return {
"enabled": False, "enabled": False,
"busy": False, "busy": False,
"acts": {
"power": False,
"reset": False,
},
"leds": { "leds": {
"power": False, "power": False,
"hdd": False, "hdd": False,

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

@ -75,7 +75,8 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
self.__long_click_delay = long_click_delay self.__long_click_delay = long_click_delay
self.__notifier = aiotools.AioNotifier() self.__notifier = aiotools.AioNotifier()
self.__region = aiotools.AioExclusiveRegion(AtxIsBusyError, self.__notifier) self.__power_region = aiotools.AioExclusiveRegion(AtxIsBusyError, self.__notifier)
self.__reset_region = aiotools.AioExclusiveRegion(AtxIsBusyError, self.__notifier)
self.__line_req: (gpiod.LineRequest | None) = None self.__line_req: (gpiod.LineRequest | None) = None
@ -122,9 +123,15 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
) )
async def get_state(self) -> dict: async def get_state(self) -> dict:
power_busy = self.__power_region.is_busy()
reset_busy = self.__reset_region.is_busy()
return { return {
"enabled": True, "enabled": True,
"busy": self.__region.is_busy(), "busy": (power_busy or reset_busy),
"acts": {
"power": power_busy,
"reset": reset_busy,
},
"leds": { "leds": {
"power": self.__reader.get(self.__power_led_pin), "power": self.__reader.get(self.__power_led_pin),
"hdd": self.__reader.get(self.__hdd_led_pin), "hdd": self.__reader.get(self.__hdd_led_pin),
@ -175,13 +182,13 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
# ===== # =====
async def click_power(self, wait: bool) -> None: async def click_power(self, wait: bool) -> None:
await self.__click("power", self.__power_switch_pin, self.__click_delay, wait) await self.__click("power", self.__power_region, self.__power_switch_pin, self.__click_delay, wait)
async def click_power_long(self, wait: bool) -> None: async def click_power_long(self, wait: bool) -> None:
await self.__click("power_long", self.__power_switch_pin, self.__long_click_delay, wait) await self.__click("power_long", self.__power_region, self.__power_switch_pin, self.__long_click_delay, wait)
async def click_reset(self, wait: bool) -> None: async def click_reset(self, wait: bool) -> None:
await self.__click("reset", self.__reset_switch_pin, self.__click_delay, wait) await self.__click("reset", self.__reset_region, self.__reset_switch_pin, self.__click_delay, wait)
# ===== # =====
@ -189,14 +196,14 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
return (await self.get_state())["leds"]["power"] return (await self.get_state())["leds"]["power"]
@aiotools.atomic_fg @aiotools.atomic_fg
async def __click(self, name: str, pin: int, delay: float, wait: bool) -> None: async def __click(self, name: str, region: aiotools.AioExclusiveRegion, pin: int, delay: float, wait: bool) -> None:
if wait: if wait:
with self.__region: with region:
await self.__inner_click(name, pin, delay) await self.__inner_click(name, pin, delay)
else: else:
await aiotools.run_region_task( await aiotools.run_region_task(
f"Can't perform ATX {name} click or operation was not completed", f"Can't perform ATX {name} click or operation was not completed",
self.__region, self.__inner_click, name, pin, delay, region, self.__inner_click, name, pin, delay,
) )
@aiotools.atomic_fg @aiotools.atomic_fg