pikvm/pikvm#457: custom commands driver

This commit is contained in:
Maxim Devaev 2021-10-24 12:00:45 +03:00
parent 278b4b335c
commit 434acc5de5
2 changed files with 114 additions and 0 deletions

92
kvmd/plugins/ugpio/cmd.py Normal file
View File

@ -0,0 +1,92 @@
# ========================================================================== #
# #
# KVMD - The main PiKVM daemon. #
# #
# Copyright (C) 2018-2021 Maxim Devaev <mdevaev@gmail.com> #
# #
# 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.OUTPUT])
@classmethod
def get_pin_validator(cls) -> Callable[[Any], Any]:
return str
async def read(self, pin: str) -> bool:
_ = pin
return False
async def write(self, pin: str, state: bool) -> None:
_ = pin
if not state:
return
try:
proc = await aioproc.log_process(self.__cmd, logger=get_logger(0))
if proc.returncode != 0:
raise RuntimeError(f"Custom command error: pid={proc.pid}; retcode={proc.returncode}")
except Exception as err:
get_logger(0).error("Can't run custom command %s: %s", self.__cmd, tools.efmt(err))
raise GpioDriverOfflineError(self)
def __str__(self) -> str:
return f"CMD({self._instance_name})"
__repr__ = __str__

View File

@ -44,6 +44,12 @@ kvmd:
relay: relay:
type: hidrelay type: hidrelay
device: /dev/hidraw0 device: /dev/hidraw0
cmd1:
type: cmd
cmd: [/bin/sleep, 5]
cmd2:
type: cmd
cmd: [/bin/ls, -l]
scheme: scheme:
__v3_usb_breaker__: __v3_usb_breaker__:
@ -86,6 +92,18 @@ kvmd:
delay: 2 delay: 2
max_delay: 5 max_delay: 5
cmd1:
pin: 0
mode: output
driver: cmd1
switch: false
cmd2:
pin: 0
mode: output
driver: cmd2
switch: false
view: view:
header: header:
title: Switches title: Switches
@ -99,6 +117,10 @@ kvmd:
- [] - []
- ["#Relay #1:", "relay1|confirm|Boop 0.1"] - ["#Relay #1:", "relay1|confirm|Boop 0.1"]
- ["#Relay #2:", "relay2|Boop 2.0"] - ["#Relay #2:", "relay2|Boop 2.0"]
- []
- ["#Commands"]
- ["#Cmd #1:", "cmd1|confirm|Run 'sleep 5'"]
- ["#Cmd #2:", "cmd2|Run 'ls -l'"]
vnc: vnc:
keymap: /usr/share/kvmd/keymaps/ru keymap: /usr/share/kvmd/keymaps/ru