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

sync atx api

Browse Source
This commit is contained in:
Devaev Maxim
2020-09-09 16:21:49 +03:00
parent 2d44539484
commit 015baee6d7
4 changed files with 41 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
kvmd/apps/kvmd/api/atx.py
View File

@@ -25,6 +25,8 @@ from aiohttp.web import Response
from ....plugins.atx import BaseAtx
from ....validators.basic import valid_bool
from ....validators.kvm import valid_atx_power_action
from ....validators.kvm import valid_atx_button
@@ -46,20 +48,22 @@ class AtxApi:
@exposed_http("POST", "/atx/power")
async def __power_handler(self, request: Request) -> Response:
action = valid_atx_power_action(request.query.get("action"))
wait = valid_bool(request.query.get("wait", "0"))
processing = await ({
"on": self.__atx.power_on,
"off": self.__atx.power_off,
"off_hard": self.__atx.power_off_hard,
"reset_hard": self.__atx.power_reset_hard,
}[action])()
}[action])(wait)
return make_json_response({"processing": processing})
@exposed_http("POST", "/atx/click")
async def __click_handler(self, request: Request) -> Response:
button = valid_atx_button(request.query.get("button"))
wait = valid_bool(request.query.get("wait", "0"))
await ({
"power": self.__atx.click_power,
"power_long": self.__atx.click_power_long,
"reset": self.__atx.click_reset,
}[button])()
}[button])(wait)
return make_json_response()

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

@@ -59,27 +59,27 @@ class BaseAtx(BasePlugin):
# =====
async def power_on(self) -> bool:
async def power_on(self, wait: bool) -> bool:
raise NotImplementedError
async def power_off(self) -> bool:
async def power_off(self, wait: bool) -> bool:
raise NotImplementedError
async def power_off_hard(self) -> bool:
async def power_off_hard(self, wait: bool) -> bool:
raise NotImplementedError
async def power_reset_hard(self) -> bool:
async def power_reset_hard(self, wait: bool) -> bool:
raise NotImplementedError
# =====
async def click_power(self) -> None:
async def click_power(self, wait: bool) -> None:
raise NotImplementedError
async def click_power_long(self) -> None:
async def click_power_long(self, wait: bool) -> None:
raise NotImplementedError
async def click_reset(self) -> None:
async def click_reset(self, wait: bool) -> None:
raise NotImplementedError

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

@@ -54,25 +54,12 @@ class Plugin(BaseAtx):
# =====
async def power_on(self) -> bool:
async def __stub_power(self, wait: bool) -> bool:
raise AtxDisabledError()
async def power_off(self) -> bool:
power_on = power_off = power_off_hard = power_reset_hard = __stub_power
async def __stub_click(self, wait: bool) -> None:
raise AtxDisabledError()
async def power_off_hard(self) -> bool:
raise AtxDisabledError()
async def power_reset_hard(self) -> bool:
raise AtxDisabledError()
# =====
async def click_power(self) -> None:
raise AtxDisabledError()
async def click_power_long(self) -> None:
raise AtxDisabledError()
async def click_reset(self) -> None:
raise AtxDisabledError()
click_power = click_power_long = click_reset = __stub_click

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

@@ -130,40 +130,40 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
# =====
async def power_on(self) -> bool:
async def power_on(self, wait: bool) -> bool:
if not (await self.__get_power()):
await self.click_power()
await self.click_power(wait)
return True
return False
async def power_off(self) -> bool:
async def power_off(self, wait: bool) -> bool:
if (await self.__get_power()):
await self.click_power()
await self.click_power(wait)
return True
return False
async def power_off_hard(self) -> bool:
async def power_off_hard(self, wait: bool) -> bool:
if (await self.__get_power()):
await self.click_power_long()
await self.click_power_long(wait)
return True
return False
async def power_reset_hard(self) -> bool:
async def power_reset_hard(self, wait: bool) -> bool:
if (await self.__get_power()):
await self.click_reset()
await self.click_reset(wait)
return True
return False
# =====
async def click_power(self) -> None:
await self.__click("power", self.__power_switch_pin, self.__click_delay)
async def click_power(self, wait: bool) -> None:
await self.__click("power", self.__power_switch_pin, self.__click_delay, wait)
async def click_power_long(self) -> None:
await self.__click("power_long", self.__power_switch_pin, self.__long_click_delay)
async def click_power_long(self, wait: bool) -> None:
await self.__click("power_long", self.__power_switch_pin, self.__long_click_delay, wait)
async def click_reset(self) -> None:
await self.__click("reset", self.__reset_switch_pin, self.__click_delay)
async def click_reset(self, wait: bool) -> None:
await self.__click("reset", self.__reset_switch_pin, self.__click_delay, wait)
# =====
@@ -171,11 +171,15 @@ class Plugin(BaseAtx): # pylint: disable=too-many-instance-attributes
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(
"Can't perform ATX click or operation was not completed",
self.__region, self.__inner_click, name, pin, delay,
)
async def __click(self, name: str, pin: int, delay: float, wait: bool) -> None:
if wait:
async with self.__region:
await self.__inner_click(name, pin, delay)
else:
await aiotools.run_region_task(
"Can't perform ATX click or operation was not completed",
self.__region, self.__inner_click, name, pin, delay,
)
@aiotools.atomic
async def __inner_click(self, name: str, pin: int, delay: float) -> None: