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

refactoring

Browse Source
This commit is contained in:
Maxim Devaev 2022-08-08 00:57:17 +03:00
parent 2482b466a3
commit de6fd94349
7 changed files with 22 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -47,7 +47,7 @@ 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"))
wait = valid_bool(request.query.get("wait", False))
await ({
"on": self.__atx.power_on,
"off": self.__atx.power_off,
@ -59,7 +59,7 @@ class AtxApi:
@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"))
wait = valid_bool(request.query.get("wait", False))
await ({
"power": self.__atx.click_power,
"power_long": self.__atx.click_power_long,

2
kvmd/apps/kvmd/api/hid.py
View File

@ -132,7 +132,7 @@ class HidApi:
@exposed_http("POST", "/hid/print")
async def __print_handler(self, request: Request) -> Response:
text = await request.text()
limit = int(valid_int_f0(request.query.get("limit", "1024")))
limit = int(valid_int_f0(request.query.get("limit", 1024)))
if limit > 0:
text = text[:limit]
symmap = self.__ensure_symmap(request.query.get("keymap", self.__default_keymap_name))

7
kvmd/apps/kvmd/api/log.py
View File

@ -41,16 +41,13 @@ class LogApi:
@exposed_http("GET", "/log")
async def __log_handler(self, request: Request) -> StreamResponse:
seek = valid_log_seek(request.query.get("seek", "0"))
follow = valid_bool(request.query.get("follow", "false"))
seek = valid_log_seek(request.query.get("seek", 0))
follow = valid_bool(request.query.get("follow", False))
response = await start_streaming(request, "text/plain")
async for record in self.__log_reader.poll_log(seek, follow):
await response.write(("[%s %s] --- %s" % (
record["dt"].strftime("%Y-%m-%d %H:%M:%S"),
record["service"],
record["msg"],
)).encode("utf-8") + b"\r\n")
return response

2
kvmd/apps/kvmd/api/msd.py
View File

@ -138,7 +138,7 @@ class MsdApi:
@exposed_http("POST", "/msd/write_remote")
async def __write_remote_handler(self, request: Request) -> Union[Response, StreamResponse]: # pylint: disable=too-many-locals
url = valid_url(request.query.get("url"))
insecure = valid_bool(request.query.get("insecure", "0"))
insecure = valid_bool(request.query.get("insecure", False))
timeout = valid_float_f01(request.query.get("timeout", 10.0))
remove_incomplete = self.__get_remove_incomplete(request)

24
kvmd/apps/kvmd/api/streamer.py
View File

@ -57,12 +57,12 @@ class StreamerApi:
@exposed_http("GET", "/streamer/snapshot")
async def __take_snapshot_handler(self, request: Request) -> Response:
snapshot = await self.__streamer.take_snapshot(
save=valid_bool(request.query.get("save", "false")),
load=valid_bool(request.query.get("load", "false")),
allow_offline=valid_bool(request.query.get("allow_offline", "false")),
save=valid_bool(request.query.get("save", False)),
load=valid_bool(request.query.get("load", False)),
allow_offline=valid_bool(request.query.get("allow_offline", False)),
)
if snapshot:
if valid_bool(request.query.get("ocr", "false")):
if valid_bool(request.query.get("ocr", False)):
langs = self.__ocr.get_available_langs()
return Response(
body=(await self.__ocr.recognize(
@ -72,19 +72,19 @@ class StreamerApi:
subval=(lambda lang: check_string_in_list(lang, "OCR lang", langs)),
name="OCR langs list",
),
left=int(valid_number(request.query.get("ocr_left", "-1"))),
top=int(valid_number(request.query.get("ocr_top", "-1"))),
right=int(valid_number(request.query.get("ocr_right", "-1"))),
bottom=int(valid_number(request.query.get("ocr_bottom", "-1"))),
left=int(valid_number(request.query.get("ocr_left", -1))),
top=int(valid_number(request.query.get("ocr_top", -1))),
right=int(valid_number(request.query.get("ocr_right", -1))),
bottom=int(valid_number(request.query.get("ocr_bottom", -1))),
)),
headers=dict(snapshot.headers),
content_type="text/plain",
)
elif valid_bool(request.query.get("preview", "false")):
elif valid_bool(request.query.get("preview", False)):
data = await snapshot.make_preview(
max_width=valid_int_f0(request.query.get("preview_max_width", "0")),
max_height=valid_int_f0(request.query.get("preview_max_height", "0")),
quality=valid_stream_quality(request.query.get("preview_quality", "80")),
max_width=valid_int_f0(request.query.get("preview_max_width", 0)),
max_height=valid_int_f0(request.query.get("preview_max_height", 0)),
quality=valid_stream_quality(request.query.get("preview_quality", 80)),
)
else:
data = snapshot.data

6
kvmd/apps/kvmd/api/ugpio.py
View File

@ -51,14 +51,14 @@ class UserGpioApi:
async def __switch_handler(self, request: Request) -> Response:
channel = valid_ugpio_channel(request.query.get("channel"))
state = valid_bool(request.query.get("state"))
wait = valid_bool(request.query.get("wait", "0"))
wait = valid_bool(request.query.get("wait", False))
await self.__user_gpio.switch(channel, state, wait)
return make_json_response()
@exposed_http("POST", "/gpio/pulse")
async def __pulse_handler(self, request: Request) -> Response:
channel = valid_ugpio_channel(request.query.get("channel"))
delay = valid_float_f0(request.query.get("delay", "0"))
wait = valid_bool(request.query.get("wait", "0"))
delay = valid_float_f0(request.query.get("delay", 0.0))
wait = valid_bool(request.query.get("wait", False))
await self.__user_gpio.pulse(channel, delay, wait)
return make_json_response()

2
kvmd/apps/kvmd/server.py
View File

@ -226,7 +226,7 @@ class KvmdServer(HttpServer): # pylint: disable=too-many-arguments,too-many-ins
@exposed_http("GET", "/ws")
async def __ws_handler(self, request: Request) -> WebSocketResponse:
stream = valid_bool(request.query.get("stream", "true"))
stream = valid_bool(request.query.get("stream", True))
async with self._ws_session(request, stream=stream) as ws:
stage1 = [
("gpio_model_state", self.__user_gpio.get_model()),