diff options
| -rw-r--r-- | kvmd/plugins/ugpio/cmd.py | 10 | ||||
| -rw-r--r-- | kvmd/plugins/ugpio/cmdret.py | 89 |
2 files changed, 91 insertions, 8 deletions
diff --git a/kvmd/plugins/ugpio/cmd.py b/kvmd/plugins/ugpio/cmd.py index 537a68d1..8e1922a4 100644 --- a/kvmd/plugins/ugpio/cmd.py +++ b/kvmd/plugins/ugpio/cmd.py @@ -63,7 +63,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute @classmethod def get_modes(cls) -> Set[str]: - return set([UserGpioModes.OUTPUT, UserGpioModes.INPUT]) + return set([UserGpioModes.OUTPUT]) @classmethod def get_pin_validator(cls) -> Callable[[Any], Any]: @@ -71,13 +71,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute async def read(self, pin: str) -> bool: _ = pin - try: - proc = await aioproc.log_process(self.__cmd, logger=get_logger(0), prefix=str(self)) - return (proc.returncode == 0) - except Exception as err: - get_logger(0).error("Can't run custom command [ %s ]: %s", - tools.cmdfmt(self.__cmd), tools.efmt(err)) - raise GpioDriverOfflineError(self) + return False async def write(self, pin: str, state: bool) -> None: _ = pin diff --git a/kvmd/plugins/ugpio/cmdret.py b/kvmd/plugins/ugpio/cmdret.py new file mode 100644 index 00000000..db55ee05 --- /dev/null +++ b/kvmd/plugins/ugpio/cmdret.py @@ -0,0 +1,89 @@ +# ========================================================================== # +# # +# KVMD - The main PiKVM daemon. # +# # +# Copyright (C) 2018-2022 Maxim Devaev <[email protected]> # +# # +# This program is free software: you can redistribute it and/or modify # +# it under the terms of the GNU General Public License as published by # +# the Free Software Foundation, either version 3 of the License, or # +# (at your option) any later version. # +# # +# This program is distributed in the hope that it will be useful, # +# but WITHOUT ANY WARRANTY; without even the implied warranty of # +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # +# GNU General Public License for more details. # +# # +# You should have received a copy of the GNU General Public License # +# along with this program. If not, see <https://www.gnu.org/licenses/>. # +# # +# ========================================================================== # + + +from typing import List +from typing import Dict +from typing import Set +from typing import Callable +from typing import Any + +from ...logging import get_logger + +from ... import tools +from ... import aiotools +from ... import aioproc + +from ...yamlconf import Option + +from ...validators.os import valid_command + +from . import GpioDriverOfflineError +from . import UserGpioModes +from . import BaseUserGpioDriver + + +# ===== +class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attributes + def __init__( # pylint: disable=super-init-not-called + self, + instance_name: str, + notifier: aiotools.AioNotifier, + + cmd: List[str], + ) -> None: + + super().__init__(instance_name, notifier) + + self.__cmd = cmd + + @classmethod + def get_plugin_options(cls) -> Dict: + return { + "cmd": Option([], type=valid_command), + } + + @classmethod + def get_modes(cls) -> Set[str]: + return set([UserGpioModes.INPUT]) + + @classmethod + def get_pin_validator(cls) -> Callable[[Any], Any]: + return str + + async def read(self, pin: str) -> bool: + _ = pin + try: + proc = await aioproc.log_process(self.__cmd, logger=get_logger(0), prefix=str(self)) + return (proc.returncode == 0) + except Exception as err: + get_logger(0).error("Can't run custom command [ %s ]: %s", + tools.cmdfmt(self.__cmd), tools.efmt(err)) + raise GpioDriverOfflineError(self) + + async def write(self, pin: str, state: bool) -> None: + _ = pin + _ = state + + def __str__(self) -> str: + return f"CMDRET({self._instance_name})" + + __repr__ = __str__ |