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

inverted output and initial state

Browse Source
This commit is contained in:
Devaev Maxim 2020-08-31 03:58:07 +03:00
parent cb1e5c10cb
commit 9baf507453
2 changed files with 36 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
kvmd/apps/__init__.py
View File

@ -177,16 +177,17 @@ def _patch_dynamic( # pylint: disable=too-many-locals
try: try:
mode = valid_gpio_mode(params.get("mode", "")) mode = valid_gpio_mode(params.get("mode", ""))
except Exception: except Exception:
mode = "" pass
channel_scheme: Dict = { finally:
ch_scheme: Dict = {
"pin": Option(-1, type=valid_gpio_pin), "pin": Option(-1, type=valid_gpio_pin),
"mode": Option("", type=valid_gpio_mode), "mode": Option("", type=valid_gpio_mode),
"title": Option(""), "title": Option(""),
"inverted": Option(False, type=valid_bool),
} }
if mode == "input": if mode == "output":
channel_scheme["inverted"] = Option(False, type=valid_bool) ch_scheme.update({
else: # output "initial": Option(False, type=valid_bool),
channel_scheme.update({
"switch": Option(True, type=valid_bool), "switch": Option(True, type=valid_bool),
"pulse": { "pulse": {
"delay": Option(0.1, type=valid_float_f0), "delay": Option(0.1, type=valid_float_f0),
@ -194,7 +195,7 @@ def _patch_dynamic( # pylint: disable=too-many-locals
"max_delay": Option(0.1, type=valid_float_f01), "max_delay": Option(0.1, type=valid_float_f01),
}, },
}) })
scheme["kvmd"]["gpio"]["scheme"][channel] = channel_scheme scheme["kvmd"]["gpio"]["scheme"][channel] = ch_scheme
return rebuild return rebuild

24
kvmd/apps/kvmd/ugpio.py
View File

@ -84,13 +84,14 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes
def __init__(self, channel: str, config: Section, notifier: aiotools.AioNotifier) -> None: def __init__(self, channel: str, config: Section, notifier: aiotools.AioNotifier) -> None:
self.__channel = channel self.__channel = channel
self.__title: str = config.title self.__title: str = config.title
self.__pin: int = gpio.set_output(config.pin) self.__pin: int = gpio.set_output(config.pin, (config.initial ^ config.inverted))
self.__inverted: bool = config.inverted
self.__switch: bool = config.switch self.__switch: bool = config.switch
self.__pulse_delay: float = config.pulse.delay self.__pulse_delay: float = config.pulse.delay
self.__min_pulse_delay: float = config.pulse.min_delay self.__min_pulse_delay: float = config.pulse.min_delay
self.__max_pulse_delay: float = config.pulse.max_delay self.__max_pulse_delay: float = config.pulse.max_delay
self.__state = False self.__state = config.initial
self.__region = aiotools.AioExclusiveRegion(GpioChannelIsBusyError, notifier) self.__region = aiotools.AioExclusiveRegion(GpioChannelIsBusyError, notifier)
def get_scheme(self) -> Dict: def get_scheme(self) -> Dict:
@ -122,9 +123,9 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes
raise GpioSwitchNotSupported() raise GpioSwitchNotSupported()
async with self.__region: async with self.__region:
# Состояние проверяется только при изменении # Состояние проверяется только при изменении
real_state = gpio.read(self.__pin) real_state = self.__read()
if state != real_state: if state != real_state:
gpio.write(self.__pin, state) self.__write(state)
self.__state = state self.__state = state
get_logger(0).info("Switched GPIO %s to %d", self, state) get_logger(0).info("Switched GPIO %s to %d", self, state)
return True return True
@ -144,15 +145,24 @@ class _GpioOutput: # pylint: disable=too-many-instance-attributes
@aiotools.atomic @aiotools.atomic
async def __inner_pulse(self, delay: float) -> None: async def __inner_pulse(self, delay: float) -> None:
try: try:
gpio.write(self.__pin, True) self.__write(True)
await asyncio.sleep(delay) await asyncio.sleep(delay)
finally: finally:
gpio.write(self.__pin, False) self.__write(False)
await asyncio.sleep(1) await asyncio.sleep(1)
get_logger(0).info("Pulsed GPIO %s", self) get_logger(0).info("Pulsed GPIO %s", self)
def __read(self) -> bool:
return (gpio.read(self.__pin) ^ self.__inverted)
def __write(self, state: bool) -> None:
gpio.write(self.__pin, (state ^ self.__inverted))
def __str__(self) -> str: def __str__(self) -> str:
return f"Output({self.__channel}, pin={self.__pin}, switch={self.__switch}, pulse={bool(self.__max_pulse_delay)})" return (
f"Output({self.__channel}, pin={self.__pin}, inverted={self.__inverted},"
f" switch={self.__switch}, pulse={bool(self.__max_pulse_delay)})"
)
__repr__ = __str__ __repr__ = __str__