Merge pull request #86 from kaulketh/develop

Develop

bot_framework/__init__.py → app/__init__.py

@@ -6,4 +6,4 @@
 __maintainer__ = "Thomas Kaulke"
 __status__ = "Production"
 
-from .telepot_bot import *
+from .framework import *

bot_framework/telepot_bot.py → app/framework.py

@@ -18,7 +18,7 @@
 from config import AUTO_REBOOT_ENABLED, AUTO_REBOOT_TIME, ID_CHAT_THK, \
     RUNNING, TOKEN_TELEGRAM_BOT, commands, m_not_allowed, m_pls_select, \
     m_rebooted, m_restarted, m_started, m_stopped, m_updated, m_wrong_id, \
-    AUTO_START
+    AUTO_START, AUTO_REBOOT_MSG, AUTO_START_MSG
 from control import peripheral_functions, run_thread, service, \
     stop_threads
 from control.reboot import AutoReboot
@@ -87,9 +87,7 @@ def __btn_grp(self, choices: list) -> list:
         return btn_list
 
     def __send(self, ch_id, text, reply_markup, parse_mode='Markdown'):
-        self.__log.debug(
-            f"Message posted: "
-            f"{ch_id}|{text}|{reply_markup}|{parse_mode}".replace("\n", " "))
+        self.__log.info(text.split(sep='/n'))
         self.__bot.sendMessage(ch_id, text, reply_markup=reply_markup,
                                parse_mode=parse_mode)
 
@@ -159,7 +157,7 @@ def sos() -> bool:
 
         if content_type == 'text':
             command = msg['text']
-            self.__log.info(f"Got command '{command}'")
+            self.__log.info(command)
 
             # Bot menu respectively Telegram-in-app-commands
             if execution_possible("/start"):
@@ -197,47 +195,45 @@ def sos() -> bool:
             self.__reply_wrong_command(chat_id, content_type)
 
     def start(self):
-        self.__log.info(RUNNING)
+        self.__log.debug(RUNNING)
         for a in self.__admins:
             self.__send(a, m_started, reply_markup=self.__remove_keyboard)
-
-        MessageLoop(self.__bot,
-                    {'chat': self.__handle}).run_as_thread()
-        # TODO: nfo string/text as constant w/ translations (II)
-        as_nfo = f"Autostart"
-        self.__log.info(f"{as_nfo} = {AUTO_START}")
-        with open(HISTORY, "r") as f:
-            # FIXME: if no HISTORY, impossible to find line in file
-            lines = f.readlines()
-            # if len(f.readlines()) > 0 else ["new file\n"]
-            line = lines[-1]
-            self.__log.warning(line.replace("\n", ""))
-            # TODO: implement considering of translation of stored command after language change
-            #  - search key of value/stored string and gather translations with this key
-            #  - depending of set language execute/set command text
-            cmd = line.partition(" HISTORY ")[2].replace("\n", "")
-            _stop = (cmd == STOP)
-            self.__log.warning(_stop)
+        MessageLoop(self.__bot, {'chat': self.__handle}).run_as_thread()
+        try:
+            with open(HISTORY, "r") as f:
+                lines = f.readlines()
+                if lines:
+                    pass
+                else:
+                    raise ValueError("Empty file")
+        except (FileNotFoundError, ValueError):
+            with open(HISTORY, "w") as f:
+                f.write("new file\n")
+            with open(HISTORY, "r") as f:
+                lines = f.readlines()
+        line = lines[-1]
+        self.__log.debug(f"History: {line.strip()}")
+        # TODO: implement considering of translation of stored command after language change
+        #  - search key of value/stored string and gather translations with this key
+        #  - depending of set language execute/set command text
+        cmd = line.partition(" HISTORY ")[2].rstrip()
         if AUTO_START:
-            if not _stop:
+            self.__log.info(AUTO_START_MSG)
+            if not (cmd == STOP):
                 self.__func_thread = run_thread(cmd, ID_CHAT_THK, self)
                 for a in self.__admins:
-                    self.__send(a, f"{as_nfo}: {cmd}",
+                    self.__send(a, f"{AUTO_START_MSG}: {cmd}",
                                 reply_markup=self.kb_stop)
             else:
                 open(HISTORY, "w").close()
                 self.__stop_function(ID_CHAT_THK, msg=None)
-
-        # TODO: nfo string/text as constant w/ translations (I)
-        ar_nfo = f"Auto-Reboot"
-        self.__log.info(f"{ar_nfo} = {AUTO_REBOOT_ENABLED}")
         if AUTO_REBOOT_ENABLED:
+            self.__log.info(AUTO_REBOOT_MSG)
             for a in self.__admins:
                 kb = self.kb_stop if AUTO_START else self.__remove_keyboard
-                self.__send(a, f"{ar_nfo}: {AUTO_REBOOT_TIME} CET",
+                self.__send(a, f"{AUTO_REBOOT_MSG}: {AUTO_REBOOT_TIME} CET",
                             reply_markup=kb)
             AutoReboot(reboot_time=AUTO_REBOOT_TIME, bot=self).start()
-
         while True:
             try:
                 signal.pause()

bot.py

@@ -9,13 +9,14 @@
 
 from urllib3.exceptions import HTTPError, ProtocolError
 
-from bot_framework import *
+from app import *
+from control import peripheral_functions
 from logger import LOGGER
 
 if __name__ == '__main__':
 
     try:
-        telepot_bot.main()
+        framework.main()
     except (ConnectionResetError, ProtocolError, HTTPError) as e:
         LOGGER.error(f"Connection error occurs: {e}")
         exit()

config/settings.py

@@ -19,7 +19,9 @@
 
 AUTO_REBOOT_ENABLED = False
 AUTO_REBOOT_TIME = "00:30"
+AUTO_REBOOT_MSG = f"Auto-Reboot"
 AUTO_START = False
+AUTO_START_MSG = f"Autostart"
 LANGUAGE = "German"  # language keys: "German", "English", "French"
 RUNNING = "Bot is running..."
 TRANSLATIONS = 'ui_translations.json'

control/__init__.py

@@ -23,15 +23,15 @@ def stopped():
     return flag
 
 
-def set_stop_flag(_flag):
+def set_stop_flag(new_flag):
     """
     Set a global stop flag for stopping while loops.
 
-    :type _flag: bool
+    :type new_flag: bool
     """
     global flag
-    flag = _flag
-    LOGGER.debug(f"Stop flag: {flag}")
+    flag = new_flag
+    LOGGER.debug(flag)
 
 
 def run_thread(func_name, request_id, bot):

control/light.py

@@ -30,7 +30,6 @@ def __init__(self, function, wreath, name=None, request_id=None, bot=None):
         self.__chat_id = request_id
         self.__bot = bot
         self.__do_run = True
-        self.__stopped = "Stop requested, stopped"
 
     @property
     def __process(self):
@@ -44,10 +43,7 @@ def is_running(self) -> bool:
         return self.__do_run
 
     def run(self):
-        LightFunction.log.info(
-            f"Initialized '{self.__f_name}' from ID:{self.__chat_id}, "
-            f"process: {self.__function}")
-
+        LightFunction.log.info(f"{self.__f_name} {self.__function}")
         p = self.__process
         LightFunction.threads.append(self)
         while self.__do_run:
@@ -60,8 +56,7 @@ def run(self):
 
     def stop(self):
         self.__do_run = False
-        LightFunction.log.info(
-            f"{self.__f_name}: {self.__stopped}: {self.__function}")
+        LightFunction.log.debug(f"{self.__f_name} {self.__function}")
         self.__bot.external_request(m_stop_f.format(self.__f_name),
                                     reply_markup=None,
                                     chat_id=self.__chat_id, bot=self.__bot)

control/service.py

@@ -86,7 +86,7 @@ def execute_os_command(self):
         if self.__command is not None:
             try:
                 if self.__log_msg is not None:
-                    self.__logger.info(self.__log_msg)
+                    self.__logger.info(f"{self.__command} {self.__log_msg}")
                 else:
                     self.__logger.info(self.__command)
                 os.system(self.__command)

functions/colorant.py

@@ -23,7 +23,8 @@ class Colorant:
 
     def __init__(self, fairy_lights: Adafruit_NeoPixel, color_key=None):
         self.__fairy_lights = fairy_lights
-        self.__color = None
+        self.__color_keys = [k for k in list(OwnColors.color.keys()) if
+                             k != "OFF"]
         Colorant.log.debug(f"Init instance of {self.__class__.__name__}.")
         if color_key is not None:
             Colorant.log.debug(f"Run color '{color_key}'")
@@ -32,7 +33,7 @@ def __init__(self, fairy_lights: Adafruit_NeoPixel, color_key=None):
             Colorant.log.debug(
                 f"Call function '{inspect.stack()[1].function}'")
 
-    def __function_loop(self, function):
+    def __loop(self, function):
         Colorant.log.debug(f"Running loop: {inspect.stack()[1].function}")
         from control import stopped
         while not stopped():
@@ -56,12 +57,11 @@ def __start(self, color, brightness=None):
             exit()
 
     def run(self, key, brightness):
-        self.__color = OwnColors.color.get(key)
-        self.__start(self.__color, brightness)
+        self.__start(OwnColors.color.get(key), brightness)
 
     def fade(self):
         def __fade():
-            for c in list(OwnColors.color.keys())[1:]:
+            for c in self.__color_keys:
                 for i in range(
                         wreath_setup(self.__fairy_lights)[1]):
                     self.run(c, brightness=i)
@@ -73,15 +73,15 @@ def __fade():
                     self.run(c, brightness=b)
                     time.sleep(uniform(0.001, 0.05))
 
-        self.__function_loop(__fade)
+        self.__loop(__fade)
 
     def switch(self):
         def __switch():
-            for c in list(OwnColors.color.keys())[1:]:
+            for c in self.__color_keys:
                 self.run(c, None)
                 time.sleep(uniform(0.25, 1))
 
-        self.__function_loop(__switch)
+        self.__loop(__switch)
 
     def strobe(self):
         def __strobe():
@@ -90,7 +90,7 @@ def __strobe():
             self.run('off', 0)
             time.sleep(uniform(0.5, 3))
 
-        self.__function_loop(__strobe)
+        self.__loop(__strobe)
 
 
 def run_red(fairy_lights):

logger/logfiles.py

@@ -0,0 +1,24 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# -----------------------------------------------------------
+# logfiles
+# created 31.12.2023
+# Thomas Kaulke, [email protected]
+# https://github.com/kaulketh
+# -----------------------------------------------------------
+
+APP_NAME = "LedPi"
+DEBUG_LOG = True
+DIRECTORY = "logs"
+EXTENSION = "log"
+
+DEBUG_FILE = f"debug.{EXTENSION}"
+ERROR_FILE = f"error.{EXTENSION}"
+HISTORY_FILE = f"history.{EXTENSION}"
+INFO_FILE = f"info.{EXTENSION}"
+
+FILE_COUNT = 5
+FILE_SIZE = 3_145_728  # 3MB
+
+if __name__ == '__main__':
+    pass

logger/logger.py

@@ -7,6 +7,9 @@
 import os
 import sys
 
+from logger.logfiles import APP_NAME, DIRECTORY, FILE_SIZE, FILE_COUNT, DEBUG_FILE, \
+    INFO_FILE, ERROR_FILE, HISTORY_FILE, DEBUG_LOG
+
 
 class _Singleton(type):
     """ A metaclass that creates a Singleton base class when called. """
@@ -24,33 +27,29 @@ class Singleton(_Singleton('SingletonMeta', (object,), {})):
 
 
 class _LoggerMeta(type, Singleton):
-    NAME = "LedPi"
-    FOLDER_PATH = "../logs"
-    ADDITIONAL_DEBUG_LOG = True
-
+    NAME = APP_NAME
+    FOLDER_PATH = f"../{DIRECTORY}"
+    ADDITIONAL_DEBUG_LOG = DEBUG_LOG
     THIS_FOLDER = os.path.dirname(os.path.abspath(__file__))
     """Runtime location"""
-
     LOG_FOLDER = os.path.join(THIS_FOLDER, FOLDER_PATH)
     """Defined log folder related to location"""
-
-    # define log files, names, formats
-    DEB_LOG = f"{LOG_FOLDER}/debug.log"
-    INF_LOG = f"{LOG_FOLDER}/info.log"
-    ERR_LOG = f"{LOG_FOLDER}/error.log"
-    HIS_LOG = f"{LOG_FOLDER}/history.log"
-    MAX_BYTE = 1024 * 1024 * 3  # 3MB
-    BACK_COUNT = 5
-
+    # log files, names, formats
+    DEB_LOG = f"{LOG_FOLDER}/{DEBUG_FILE}"
+    INF_LOG = f"{LOG_FOLDER}/{INFO_FILE}"
+    ERR_LOG = f"{LOG_FOLDER}/{ERROR_FILE}"
+    HIS_LOG = f"{LOG_FOLDER}/{HISTORY_FILE}"
+    MAX_BYTE = FILE_SIZE
+    BACK_COUNT = FILE_COUNT
+    # log formats
     DAT_FMT = "%Y-%m-%d %H:%M:%S"
-    INF_FMT = "%(asctime)s %(name)s %(levelname)s " \
-              "[%(pathname)s %(funcName)s(lnr.%(lineno)s)] %(message)s"
-    ERR_FMT = "%(asctime)s %(name)s %(levelname)s " \
-              "[%(pathname)s %(funcName)s(lnr.%(lineno)s)] " \
-              "[thread: %(threadName)s] %(message)s"
-    DEB_FMT = "%(asctime)s %(name)s %(levelname)s " \
-              "%(pathname)s %(funcName)s(lnr.%(lineno)s) %(message)s"
-    HIS_FMT = "%(asctime)s %(name)s %(levelname)s %(message)s"
+    __P1 = "%(asctime)s %(name)s %(levelname)s"
+    __P2 = "%(pathname)s %(funcName)s(lnr.%(lineno)s)"
+    __P3 = "%(message)s"
+    INF_FMT = f"{__P1} [{__P2}] {__P3}"
+    ERR_FMT = f"{__P1} [{__P2}] [thread: %(threadName)s] {__P3}"
+    DEB_FMT = f"{__P1} {__P2} {__P3}"
+    HIS_FMT = f"{__P1} {__P3}"
     HISTORY = 39  # before error level
 
     # noinspection PyProtectedMember