diff options
Diffstat (limited to 'kvmd/apps/__init__.py')
| -rw-r--r-- | kvmd/apps/__init__.py | 197 |
1 files changed, 91 insertions, 106 deletions
diff --git a/kvmd/apps/__init__.py b/kvmd/apps/__init__.py index d2007f7d..062e9bfe 100644 --- a/kvmd/apps/__init__.py +++ b/kvmd/apps/__init__.py @@ -29,14 +29,13 @@ import logging.config from typing import Tuple from typing import List from typing import Dict -from typing import Sequence from typing import Optional -from typing import Union import pygments import pygments.lexers.data import pygments.formatters +from ..yamlconf import ConfigError from ..yamlconf import make_config from ..yamlconf import Section from ..yamlconf import Option @@ -44,31 +43,59 @@ from ..yamlconf import build_raw_from_options from ..yamlconf.dumper import make_config_dump from ..yamlconf.loader import load_yaml_file +from ..validators.basic import valid_bool +from ..validators.basic import valid_number +from ..validators.basic import valid_int_f1 +from ..validators.basic import valid_float_f01 + +from ..validators.fs import valid_abs_path +from ..validators.fs import valid_abs_path_exists +from ..validators.fs import valid_unix_mode + +from ..validators.net import valid_ip_or_host +from ..validators.net import valid_port + +from ..validators.auth import valid_auth_type + +from ..validators.kvm import valid_stream_quality +from ..validators.kvm import valid_stream_fps + +from ..validators.hw import valid_tty_speed +from ..validators.hw import valid_gpio_pin +from ..validators.hw import valid_gpio_pin_optional + # ===== def init( - prog: str=sys.argv[0], + prog: Optional[str]=None, description: Optional[str]=None, add_help: bool=True, + argv: Optional[List[str]]=None, ) -> Tuple[argparse.ArgumentParser, List[str], Section]: - args_parser = argparse.ArgumentParser(prog=prog, description=description, add_help=add_help) + argv = (argv or sys.argv) + assert len(argv) > 0 + + args_parser = argparse.ArgumentParser(prog=(prog or argv[0]), description=description, add_help=add_help) args_parser.add_argument("-c", "--config", dest="config_path", default="/etc/kvmd/main.yaml", metavar="<file>", - help="Set config file path") + type=valid_abs_path_exists, help="Set config file path") args_parser.add_argument("-o", "--set-options", dest="set_options", default=[], nargs="+", help="Override config options list (like sec/sub/opt=value)") args_parser.add_argument("-m", "--dump-config", dest="dump_config", action="store_true", help="View current configuration (include all overrides)") - (options, remaining) = args_parser.parse_known_args(sys.argv) + (options, remaining) = args_parser.parse_known_args(argv) + raw_config: Dict = {} - options.config_path = os.path.expanduser(options.config_path) - if os.path.exists(options.config_path): + if options.config_path: + options.config_path = os.path.expanduser(options.config_path) raw_config = load_yaml_file(options.config_path) - else: - raw_config = {} - _merge_dicts(raw_config, build_raw_from_options(options.set_options)) + scheme = _get_config_scheme() - config = make_config(raw_config, scheme) + try: + _merge_dicts(raw_config, build_raw_from_options(options.set_options)) + config = make_config(raw_config, scheme) + except ConfigError as err: + raise SystemExit("Config error: " + str(err)) if options.dump_config: dump = make_config_dump(config) @@ -96,135 +123,93 @@ def _merge_dicts(dest: Dict, src: Dict) -> None: dest[key] = src[key] -def _as_pin(pin: int) -> int: - if not isinstance(pin, int) or pin <= 0: - raise ValueError("Invalid pin number") - return pin - - -def _as_optional_pin(pin: int) -> int: - if not isinstance(pin, int) or pin < -1: - raise ValueError("Invalid optional pin number") - return pin - - -def _as_path(path: str) -> str: - if not isinstance(path, str): - raise ValueError("Invalid path") - path = str(path).strip() - if not path: - raise ValueError("Invalid path") - return path - - -def _as_optional_path(path: str) -> str: - if not isinstance(path, str): - raise ValueError("Invalid path") - return str(path).strip() - - -def _as_string_list(values: Union[str, Sequence]) -> List[str]: - if isinstance(values, str): - values = [values] - return list(map(str, values)) - - -def _as_auth_type(auth_type: str) -> str: - if not isinstance(auth_type, str): - raise ValueError("Invalid auth type") - auth_type = str(auth_type).strip() - if auth_type not in ["basic"]: - raise ValueError("Invalid auth type") - return auth_type - - def _get_config_scheme() -> Dict: return { "kvmd": { "server": { - "host": Option("localhost"), - "port": Option(0), - "unix": Option("", type=_as_optional_path, rename="unix_path"), - "unix_rm": Option(False), - "unix_mode": Option(0), - "heartbeat": Option(3.0), + "host": Option("localhost", type=valid_ip_or_host), + "port": Option(0, type=valid_port), + "unix": Option("", type=valid_abs_path, only_if="!port", unpack_as="unix_path"), + "unix_rm": Option(False, type=valid_bool), + "unix_mode": Option(0, type=valid_unix_mode), + "heartbeat": Option(3.0, type=valid_float_f01), "access_log_format": Option("[%P / %{X-Real-IP}i] '%r' => %s; size=%b ---" " referer='%{Referer}i'; user_agent='%{User-Agent}i'"), }, "auth": { - "type": Option("basic", type=_as_auth_type, rename="auth_type"), + "type": Option("basic", type=valid_auth_type, unpack_as="auth_type"), "basic": { - "htpasswd": Option("/etc/kvmd/htpasswd", type=_as_path, rename="htpasswd_path"), + "htpasswd": Option("/etc/kvmd/htpasswd", type=valid_abs_path_exists, unpack_as="htpasswd_path"), }, }, "info": { - "meta": Option("/etc/kvmd/meta.yaml", type=_as_path, rename="meta_path"), - "extras": Option("/usr/share/kvmd/extras", type=_as_path, rename="extras_path"), + "meta": Option("/etc/kvmd/meta.yaml", type=valid_abs_path_exists, unpack_as="meta_path"), + "extras": Option("/usr/share/kvmd/extras", type=valid_abs_path_exists, unpack_as="extras_path"), }, "hid": { - "reset_pin": Option(0, type=_as_pin), - "reset_delay": Option(0.1), - - "device": Option("", type=_as_path, rename="device_path"), - "speed": Option(115200), - "read_timeout": Option(2.0), - "read_retries": Option(10), - "common_retries": Option(100), - "retries_delay": Option(0.1), - "noop": Option(False), - - "state_poll": Option(0.1), + "reset_pin": Option(-1, type=valid_gpio_pin), + "reset_delay": Option(0.1, type=valid_float_f01), + + "device": Option("", type=valid_abs_path_exists, unpack_as="device_path"), + "speed": Option(115200, type=valid_tty_speed), + "read_timeout": Option(2.0, type=valid_float_f01), + "read_retries": Option(10, type=valid_int_f1), + "common_retries": Option(100, type=valid_int_f1), + "retries_delay": Option(0.1, type=valid_float_f01), + "noop": Option(False, type=valid_bool), + + "state_poll": Option(0.1, type=valid_float_f01), }, "atx": { - "enabled": Option(True), + "enabled": Option(True, type=valid_bool), - "power_led_pin": Option(-1, type=_as_optional_pin), - "hdd_led_pin": Option(-1, type=_as_optional_pin), - "power_switch_pin": Option(-1, type=_as_optional_pin), - "reset_switch_pin": Option(-1, type=_as_optional_pin), + "power_led_pin": Option(-1, type=valid_gpio_pin, only_if="enabled"), + "hdd_led_pin": Option(-1, type=valid_gpio_pin, only_if="enabled"), + "power_switch_pin": Option(-1, type=valid_gpio_pin, only_if="enabled"), + "reset_switch_pin": Option(-1, type=valid_gpio_pin, only_if="enabled"), - "click_delay": Option(0.1), - "long_click_delay": Option(5.5), + "click_delay": Option(0.1, type=valid_float_f01), + "long_click_delay": Option(5.5, type=valid_float_f01), - "state_poll": Option(0.1), + "state_poll": Option(0.1, type=valid_float_f01), }, "msd": { - "enabled": Option(True), + "enabled": Option(True, type=valid_bool), - "target_pin": Option(-1, type=_as_optional_pin), - "reset_pin": Option(-1, type=_as_optional_pin), + "target_pin": Option(-1, type=valid_gpio_pin, only_if="enabled"), + "reset_pin": Option(-1, type=valid_gpio_pin, only_if="enabled"), - "device": Option("", type=_as_optional_path, rename="device_path"), - "init_delay": Option(2.0), - "reset_delay": Option(1.0), - "write_meta": Option(True), - "chunk_size": Option(65536), + "device": Option("", type=valid_abs_path, only_if="enabled", unpack_as="device_path"), + "init_delay": Option(2.0, type=valid_float_f01), + "reset_delay": Option(1.0, type=valid_float_f01), + "write_meta": Option(True, type=valid_bool), + "chunk_size": Option(65536, type=(lambda arg: valid_number(arg, min=1024))), }, "streamer": { - "cap_pin": Option(0, type=_as_optional_pin), - "conv_pin": Option(0, type=_as_optional_pin), + "cap_pin": Option(-1, type=valid_gpio_pin_optional), + "conv_pin": Option(-1, type=valid_gpio_pin_optional), - "sync_delay": Option(1.0), - "init_delay": Option(1.0), - "init_restart_after": Option(0.0), - "shutdown_delay": Option(10.0), - "state_poll": Option(1.0), + "sync_delay": Option(1.0, type=valid_float_f01), + "init_delay": Option(1.0, type=valid_float_f01), + "init_restart_after": Option(0.0, type=(lambda arg: valid_number(arg, min=0.0, type=float))), + "shutdown_delay": Option(10.0, type=valid_float_f01), + "state_poll": Option(1.0, type=valid_float_f01), - "quality": Option(80), - "desired_fps": Option(0), + "quality": Option(80, type=valid_stream_quality), + "desired_fps": Option(0, type=valid_stream_fps), - "host": Option("localhost"), - "port": Option(0), - "unix": Option("", type=_as_optional_path, rename="unix_path"), - "timeout": Option(2.0), + "host": Option("localhost", type=valid_ip_or_host), + "port": Option(0, type=valid_port), + "unix": Option("", type=valid_abs_path, only_if="!port", unpack_as="unix_path"), + "timeout": Option(2.0, type=valid_float_f01), - "cmd": Option(["/bin/true"], type=_as_string_list), + "cmd": Option(["/bin/true"]), # TODO: Validator }, }, |