techbloc/asterisk-misc
1
0
Fork 0
Code Issues Pull requests Releases Activity

FIX bytes/str blah

Browse source
This commit is contained in:
boyska 2022-03-15 19:09:16 +01:00
parent b654700ccf
commit 68e6805bd9
1 changed files with 14 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
ring_mon/apply-state
View file

@ -17,15 +17,15 @@ you can just run your favourite inotify-like tool (entr, watchman, you name it)
time you need it. time you need it.
""" """
import sys
import os
import logging import logging
import time import os
import signal import signal
import sys
import time
from argparse import ArgumentParser from argparse import ArgumentParser
from multiprocessing import Pipe, Process
from pathlib import Path from pathlib import Path
from subprocess import Popen, check_output, CalledProcessError from subprocess import CalledProcessError, Popen, check_output
from multiprocessing import Process, Pipe
def rotate(lst: list, n: int) -> list: def rotate(lst: list, n: int) -> list:
@ -61,7 +61,7 @@ class LightManager:
self.stop_pipe_r, self.stop_pipe_w = Pipe(False) self.stop_pipe_r, self.stop_pipe_w = Pipe(False)
def set_light(self, on: bool): def set_light(self, on: bool):
self.log.debug('set %s', on) self.log.debug("set %s", on)
env = os.environ.copy() env = os.environ.copy()
env["DEVICE_NUMBER"] = str(self.device_number) env["DEVICE_NUMBER"] = str(self.device_number)
@ -87,11 +87,13 @@ class LightManager:
def get_status(self) -> str: def get_status(self) -> str:
try: try:
status = check_output([str(self.args.bin_state.resolve()), self.device_name]) status = check_output(
[str(self.args.bin_state.resolve()), self.device_name]
)
except CalledProcessError: except CalledProcessError:
status = 'off' status = "off"
status = status.lower().strip().decode('ascii', errors='ignore') status = status.lower().strip()
if status not in self.waveforms: if status not in self.waveforms:
return "off" return "off"
return status return status
@ -134,7 +136,9 @@ def main():
p.add_argument("--devices", type=parse_devices, default="1") p.add_argument("--devices", type=parse_devices, default="1")
p.add_argument("--bin-state", type=Path, default=(EXE_DIR / "get-status")) p.add_argument("--bin-state", type=Path, default=(EXE_DIR / "get-status"))
p.add_argument("--bin-set-light", type=Path, default=(EXE_DIR / "set-light")) p.add_argument("--bin-set-light", type=Path, default=(EXE_DIR / "set-light"))
p.add_argument("--disable-polling", dest='enable_polling', default=True, action='store_false') p.add_argument(
"--disable-polling", dest="enable_polling", default=True, action="store_false"
)
p.add_argument( p.add_argument(
"--log-level", "--log-level",
choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],