1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
# ========================================================================== #
# #
# KVMD - The main PiKVM daemon. #
# #
# Copyright (C) 2018-2021 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/>. #
# #
# ========================================================================== #
import argparse
import errno
import time
from typing import List
from typing import Optional
from ...logging import get_logger
from ...yamlconf import Section
from ... import env
from .. import init
# =====
class RtcIsNotAvailableError(Exception):
pass
# =====
def _join_rtc(device: int, key: str) -> str:
return f"{env.SYSFS_PREFIX}/sys/class/rtc/rtc{device}/{key}"
def _read_int(device: int, key: str) -> int:
with open(_join_rtc(device, key)) as value_file:
return int(value_file.read().strip() or "0")
def _write_int(device: int, key: str, value: int) -> None:
with open(_join_rtc(device, key), "w") as value_file:
value_file.write(str(value))
def _reset_alarm(device: int, timeout: int) -> None:
now = _read_int(device, "since_epoch")
if now == 0:
raise RtcIsNotAvailableError("Current UNIX time == 0")
try:
for wake in [0, now + timeout]:
_write_int(device, "wakealarm", wake)
except OSError as err:
if err.errno != errno.EIO:
raise
raise RtcIsNotAvailableError("IO error, probably the supercapacitor is not charged")
# =====
def _cmd_run(config: Section) -> None:
logger = get_logger(0)
logger.info("Running watchdog loop on RTC%d ...", config.device)
fail = False
try:
while True:
try:
_reset_alarm(config.device, config.timeout)
except RtcIsNotAvailableError as err:
if not fail:
logger.error("RTC%d is not available now: %s; waiting ...", config.device, err)
fail = True
else:
if fail:
logger.info("RTC%d is available, working ...", config.device)
fail = False
time.sleep(config.interval)
except (SystemExit, KeyboardInterrupt):
if not fail:
_reset_alarm(config.device, config.timeout)
logger.info("The watchdog remains alarmed. Use 'kvmd-watchdog cancel' to disarm it")
logger.info("Bye-bye")
def _cmd_cancel(config: Section) -> None:
_write_int(config.device, "wakealarm", 0)
# =====
def main(argv: Optional[List[str]]=None) -> None:
(parent_parser, argv, config) = init(add_help=False, argv=argv)
parser = argparse.ArgumentParser(
prog="kvmd-watchdog",
description="RTC-based hardware watchdog",
parents=[parent_parser],
)
parser.set_defaults(cmd=(lambda *_: parser.print_help()))
subparsers = parser.add_subparsers()
cmd_run_parser = subparsers.add_parser("run", help="Run watchdog loop")
cmd_run_parser.set_defaults(cmd=_cmd_run)
cmd_cancel_parser = subparsers.add_parser("cancel", help="Cancel armed timeout")
cmd_cancel_parser.set_defaults(cmd=_cmd_cancel)
options = parser.parse_args(argv[1:])
options.cmd(config.watchdog)
|
