Add constants

slackr31337 · Jul 26, 2023 · 7db556b · 7db556b
1 parent b0da9b4
commit 7db556b
Show file tree

Hide file tree

Showing 5 changed files with 81 additions and 33 deletions.
diff --git 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
@@ -10,21 +10,39 @@
 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]"
 
 
 ##########################################
 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
@@ -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
@@ -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
@@ -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