diff --git a/.vscode/settings.json b/.vscode/settings.json index 5edc1e4..0efdd16 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -11,6 +11,7 @@ "distro", "dropin", "errin", + "FSLASH", "fstype", "getloadavg", "getmute", diff --git a/os/linux.py b/os/linux.py index 5ba0f59..d884664 100644 --- a/os/linux.py +++ b/os/linux.py @@ -10,11 +10,29 @@ from dmidecode import DMIDecode +from device.raspberrypi import RaspberryPi from service.log import LOGGER from service.util import get_boot -from device.raspberrypi import RaspberryPi - -SKIP_MOUNTS = ["live", "docker", "subvol", "tmp"] +from service.const import ( + LINUX, + ID, + PC, + LIVE, + DOCKER, + TMP, + VERSION, + MAJOR, + MINOR, + USERS, + HOSTNAME, + STRING_SPACE, + EMPTY_STRING, + STRING_UNDERSCORE, + FSLASH, +) + + +SKIP_MOUNTS = [LIVE, DOCKER, "subvol", TMP] LOG_PREFIX = r"[Linux]" @@ -22,9 +40,9 @@ class AgentPlatform: """OS Module for Linux""" - platform = "linux" + platform = LINUX os = "Linux" - hardware = "pc" + hardware = PC ########################################## def __init__(self): @@ -53,30 +71,30 @@ def update(self): def _get_system_info(self) -> dict: """Build system information and return dict""" - arch = self._cpuinfo.get("arch") + arch = str(self._cpuinfo.get("arch", EMPTY_STRING)).lower() LOGGER.info("[%s] arch: %s", self.platform, arch) - if "ARM" in arch: + if "arm" in arch: self._hardware = RaspberryPi() self.hardware = "raspberrypi" else: self._hardware = DMIDecode() info = distro.info() - _name = info.get("id", "Unknown").title() - _version = info.get("version_parts", info.get("version", "")) + _name = info.get(ID, "Unknown").title() + _version = info.get("version_parts", info.get(VERSION, EMPTY_STRING)) - if "major" in _version: - if len(_version["minor"]) > 0: - _version = f"{_version['major']}.{_version['minor']}" + if MAJOR in _version: + if len(_version[MINOR]) > 0: + _version = f"{_version.get(MAJOR)}.{_version.get(MINOR)}" else: - _version = _version["major"] + _version = _version.get(MAJOR) _release = f"{_name} {_version} ({info.get('codename')})" LOGGER.info("[%s] OS: %s", self.platform, _release) self._sysinfo = { - "hostname": self._uname.node, + HOSTNAME: self._uname.node, "manufacturer": self._hardware.manufacturer(), "model": self._hardware.model(), "serial": self._hardware.serial_number(), @@ -107,11 +125,11 @@ def _update_system_info(self) -> dict: host = "localhost" attribs[user.terminal] = f"{user.name}@{host}" - self._attribs["users"] = attribs + self._attribs[USERS] = attribs memory_usage = psutil.virtual_memory() _data = { - "users": users, + USERS: users, "ip_address": None, "ip4_addresses": [], "ip6_address": None, @@ -144,7 +162,7 @@ def _update_system_info(self) -> dict: stats = nic_stats[iface] nic_io = io_counters[iface] if "." in iface: - iface = iface.replace(".", "_") + iface = iface.replace(".", STRING_UNDERSCORE) key = f"network_{iface}" _data[key] = "Up" if stats.isup else "Down" @@ -180,9 +198,9 @@ def _update_system_info(self) -> dict: _data["mac_addresses"].append(_addr) self._attribs[key]["mac"] = _addr - _data["ip_address"] = next(iter(_data["ip4_addresses"]), "") - _data["ip6_address"] = next(iter(_data["ip6_addresses"]), "") - _data["mac_address"] = next(iter(_data["mac_addresses"]), "") + _data["ip_address"] = next(iter(_data["ip4_addresses"]), EMPTY_STRING) + _data["ip6_address"] = next(iter(_data["ip6_addresses"]), EMPTY_STRING) + _data["mac_address"] = next(iter(_data["mac_addresses"]), EMPTY_STRING) load1, load5, load15 = os.getloadavg() _data["load"] = load1 @@ -191,7 +209,7 @@ def _update_system_info(self) -> dict: for disk in psutil.disk_partitions(): if any(item in str(disk.mountpoint) for item in SKIP_MOUNTS): continue - dev = str(disk.device).split("/", maxsplit=10)[-1] + dev = str(disk.device).split(FSLASH, maxsplit=10)[-1] key = f"disk_{dev}" disk_usage = psutil.disk_usage(disk.mountpoint) _data[key] = int(disk_usage.percent) @@ -203,10 +221,10 @@ def _update_system_info(self) -> dict: "used": bytes2human(disk_usage.used), } - disk_usage = psutil.disk_usage("/") + disk_usage = psutil.disk_usage(FSLASH) _data["disk_root"] = int(disk_usage.percent) self._attribs["disk_root"] = { - "mount": "/", + "mount": FSLASH, "total": bytes2human(disk_usage.total), "used": bytes2human(disk_usage.used), } @@ -218,7 +236,9 @@ def _update_system_info(self) -> dict: if len(sensor.label) == 0: _label = f"temp_{count}" else: - _label = sensor.label.lower().replace(" ", "_") + _label = sensor.label.lower().replace( + STRING_SPACE, STRING_UNDERSCORE + ) _data[f"{_type}_{_label}"] = int(sensor.current) count += 1 @@ -227,7 +247,7 @@ def _update_system_info(self) -> dict: for _type, _fans in fans.items(): count = 0 for fan in _fans: - _label = fan.label.lower().replace(" ", "_") + _label = fan.label.lower().replace(STRING_SPACE, STRING_UNDERSCORE) if len(_label) == 0: _label = f"fan_{count}" diff --git a/service/agent.py b/service/agent.py index 794eb5e..aea60a3 100644 --- a/service/agent.py +++ b/service/agent.py @@ -52,7 +52,10 @@ CONNECTED, RESET, GPS, + EMPTY_STRING, STRING_SPACE, + STRING_UNDERSCORE, + FSLASH, ) @@ -590,7 +593,7 @@ def message_receive(self, _data: dict): def process_cmd(self, _data: dict): """Process message from Home Assistant""" - topic = _data.get(TOPIC, "").split("/") + topic = _data.get(TOPIC, EMPTY_STRING).split(FSLASH) command = topic[-1] payload = _data.get(PAYLOAD) LOGGER.debug( @@ -634,7 +637,11 @@ def process_cmd(self, _data: dict): ) elif command == SET: - sensor = _data.get(TOPIC, "").split("/")[-2].split("_", 1)[1] + sensor = ( + _data.get(TOPIC, EMPTY_STRING) + .split(FSLASH)[-2] + .split(STRING_UNDERSCORE, 1)[1] + ) LOGGER.info("%s cmd set: %s state: %s", LOG_PREFIX, sensor, payload) if sensor in self._callback: _func = self._callback.get(sensor) @@ -760,7 +767,7 @@ def _setup_module_sensors(self): LOGGER.info("%s Setup callback %s.set()", LOG_PREFIX, sensor) self._callback[sensor] = mod_class.set - _name = sensor.title().replace("_", STRING_SPACE) + _name = sensor.title().replace(STRING_UNDERSCORE, STRING_SPACE) LOGGER.debug("%s Setup module sensor: %s", LOG_PREFIX, _name) data = setup_sensor(self._config, _name) with self._sensors as _sensors: @@ -784,7 +791,7 @@ def _setup_sensors(self): if _state is None: continue - _name = sensor.title().replace("_", STRING_SPACE) + _name = sensor.title().replace(STRING_UNDERSCORE, STRING_SPACE) _data = setup_sensor(self._config, _name) _topic = _data.get(TOPIC).split("/config", 2)[0] diff --git a/service/const.py b/service/const.py index 1ad13af..8f59b0e 100644 --- a/service/const.py +++ b/service/const.py @@ -2,7 +2,21 @@ EMPTY_STRING = "" STRING_SPACE = " " +STRING_UNDERSCORE = r"_" +FSLASH = r"/" + +LINUX = "linux" +PC = "pc" +LIVE = "live" +DOCKER = "docker" +TMP = "tmp" + +ID = "id" +VERSION = "version" +MAJOR = "major" +MINOR = "minor" +USERS = "users" DEBUG = "debug" UPTIME = "uptime" RUNTIME = "runtime" diff --git a/service/log.py b/service/log.py index cae729a..3242b02 100644 --- a/service/log.py +++ b/service/log.py @@ -7,15 +7,21 @@ ######################################### def log(name="homeagent", level=logging.getLevelName("INFO")): """Setup logger""" + logging.raiseExceptions = False _logger = logging.getLogger(name) + _logger.propagate = False + _logger.write = lambda msg: _logger.info(msg) if msg != "\n" else None + _logger.setLevel(level) log_stream = logging.StreamHandler(sys.stdout) - formatter = logging.Formatter( - "[%(filename)18s: %(funcName)18s()] %(levelname)5s %(message)s" - ) - log_stream.setFormatter(formatter) + if level == logging.getLevelName("DEBUG"): + formatter = logging.Formatter( + "[%(filename)18s: %(funcName)18s()] %(levelname)5s %(message)s" + ) + log_stream.setFormatter(formatter) + _logger.addHandler(log_stream) return _logger