refactor; unify emoji style; use recent python features

.github/workflows/code-quality.yaml

@@ -19,15 +19,13 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install -q -U pip
-        pip install -q -U flake8 pytest
-        if [ -f requirements.txt ]; then pip install -q -U -r requirements.txt; fi
-    - name: Lint with flake8
-      continue-on-error: true
-      run: |
-        # stop the build if there are Python syntax errors or undefined names
-        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
-        # exit-zero treats all errors as warnings
-        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=120 --statistics
-    - name: Test with pytest
+        pip install -q -U -r requirements.txt
+        pip install -q -U pytest pytest-asyncio coverage
+
+    - name: Run tests with coverage
       run: |
-        pytest
+        coverage run -m pytest
+        coverage report -m
+
+    - name: Run Prospector
+      uses: jpetrucciani/prospector-check@master

.gitignore

@@ -20,6 +20,8 @@ wheels/
 .installed.cfg
 *.egg
 
+.coverage
+
 # dotenv
 .envrc
 

config/config.yaml

@@ -21,6 +21,8 @@ meshtastic:
 
 logging:
   level: 'info'
+  level_telegram: 'warn'
+  level_httpx: 'warn'
   use_syslog: false
   syslog_host: "${SYSLOG_HOST}"
   syslog_port: 514

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+pythonpath = src
+testpaths = tests

src/config_manager.py

@@ -1,12 +1,12 @@
 import logging
 import re
-from typing import Any, Optional, List
+from typing import Any, Optional, List, Dict
 from envyaml import EnvYAML
 
 class ConfigManager:
     def __init__(self, config_path: str = 'config/config.yaml'):
         try:
-            self.config = EnvYAML(config_path)
+            self.config: Dict[str, Any] = EnvYAML(config_path)
         except Exception as e:
             raise ValueError(f"Failed to load configuration from {config_path}: {e}")
         self.setup_logging()
@@ -23,19 +23,23 @@ def get_authorized_users(self) -> List[int]:
 
     def setup_logging(self) -> None:
         log_level = self._parse_log_level(self.get('logging.level', 'INFO'))
-        formatter = SensitiveFormatter('%(asctime)s %(levelname)s [%(name)s] %(message)s',)
+        log_level_telegram = self._parse_log_level(self.get('logging.level_telegram', 'INFO'))
+        log_level_httpx = self._parse_log_level(self.get('logging.level_telegram', 'WARN'))
+
+        formatter = SensitiveFormatter('%(asctime)s %(levelname)s %(name)s - %(message)s')
+
+        handlers = [logging.StreamHandler()]
+        if self.get('logging.file_log', False):
+            handlers.append(logging.FileHandler(self.get('logging.file_path', 'meshgram.log')))
 
-        logging.basicConfig(
-            level=log_level,
-            handlers=[
-                logging.StreamHandler(),
-                logging.FileHandler('meshgram.log')
-            ]
-        )
+        logging.basicConfig(level=log_level, handlers=handlers, format='%(asctime)s %(levelname)s %(name)s - %(message)s')
 
-        for handler in logging.getLogger().handlers:
+        for handler in logging.root.handlers:
             handler.setFormatter(formatter)
 
+        logging.getLogger('httpx').setLevel(log_level_httpx)
+        logging.getLogger('telegram').setLevel(log_level_telegram)
+
         if self.get('logging.use_syslog', False):
             self._setup_syslog_handler()
 
@@ -52,11 +56,13 @@ def _parse_log_level(self, level: Any) -> int:
             logging.warning(f"Invalid log level type: {type(level)}. Defaulting to INFO.")
             return logging.INFO
 
+    # TODO: seperate logging level for syslog handler
     def _setup_syslog_handler(self) -> None:
         try:
-            syslog_handler = logging.handlers.SysLogHandler(
+            from logging.handlers import SysLogHandler
+            syslog_handler = SysLogHandler(
                 address=(self.get('logging.syslog_host'), self.get('logging.syslog_port', 514)),
-                socktype=logging.handlers.socket.SOCK_DGRAM if self.get('logging.syslog_protocol', 'udp') == 'udp' else logging.handlers.socket.SOCK_STREAM
+                socktype=SysLogHandler.UDP_SOCKET if self.get('logging.syslog_protocol', 'udp').lower() == 'udp' else SysLogHandler.TCP_SOCKET
             )
             syslog_handler.setFormatter(SensitiveFormatter('%(name)s - %(levelname)s - %(message)s'))
             logging.getLogger().addHandler(syslog_handler)
@@ -70,16 +76,15 @@ def validate_config(self) -> None:
             'meshtastic.connection_type',
             'meshtastic.device',
         ]
-        for key in required_keys:
-            if not self.get(key):
-                raise ValueError(f"Missing required configuration: {key}")
+        missing_keys = [key for key in required_keys if not self.get(key)]
+        if missing_keys:
+            raise ValueError(f"Missing required configuration: {', '.join(missing_keys)}")
 
 class SensitiveFormatter(logging.Formatter):
     def __init__(self, fmt: Optional[str] = None, datefmt: Optional[str] = None):
         super().__init__(fmt, datefmt)
         self.sensitive_patterns = [
-            (re.compile(r'(bot\d+):(AAH[\w-]{34})'), r'\1:[REDACTED]'),
-            (re.compile(r'(token=)([A-Za-z0-9-_]{35,})'), r'\1[REDACTED]'),
+            (re.compile(r'(https://api\.telegram\.org/bot)([A-Za-z0-9:_-]{35,})(/\w+)'), r'\1[redacted]\3')
         ]
 
     def format(self, record: logging.LogRecord) -> str:

src/meshgram.py

@@ -1,6 +1,6 @@
 import argparse
 import asyncio
-from typing import Optional, List
+from typing import Optional, Sequence
 from meshtastic_interface import MeshtasticInterface
 from telegram_interface import TelegramInterface
 from message_processor import MessageProcessor
@@ -14,69 +14,72 @@ def __init__(self, config: ConfigManager) -> None:
         self.meshtastic: Optional[MeshtasticInterface] = None
         self.telegram: Optional[TelegramInterface] = None
         self.message_processor: Optional[MessageProcessor] = None
-        self.tasks: List[Task] = []
+        self.tasks: Sequence[Task] = ()
 
     async def setup(self) -> None:
         self.logger.info("Setting up meshgram...")
         try:
-            self.meshtastic = MeshtasticInterface(self.config)
-            await self.meshtastic.setup()
-
-            self.telegram = TelegramInterface(self.config)
-            await self.telegram.setup()
-
+            self.meshtastic = await self._setup_meshtastic()
+            self.telegram = await self._setup_telegram()
             self.message_processor = MessageProcessor(self.meshtastic, self.telegram, self.config)
             self.logger.info("Meshgram setup complete.")
         except Exception as e:
-            self.logger.error(f"Error during setup: {e}")
+            self.logger.error(f"Error during setup: {e}", exc_info=True)
             await self.shutdown()
             raise
 
+    async def _setup_meshtastic(self) -> MeshtasticInterface:
+        meshtastic = MeshtasticInterface(self.config)
+        await meshtastic.setup()
+        return meshtastic
+
+    async def _setup_telegram(self) -> TelegramInterface:
+        telegram = TelegramInterface(self.config)
+        await telegram.setup()
+        return telegram
+
+    async def shutdown(self) -> None:
+        self.logger.info("Shutting down meshgram...")
+        components = [self.message_processor, self.telegram, self.meshtastic]
+        for component in components:
+            if component:
+                try:
+                    await component.close()
+                except Exception as e:
+                    self.logger.error(f"Error closing {component.__class__.__name__}: {e}", exc_info=True)
+
+        for task in self.tasks:
+            if not task.done():
+                task.cancel()
+        await asyncio.gather(*self.tasks, return_exceptions=True)
+        self.logger.info("Meshgram shutdown complete.")
+
     async def run(self) -> None:
         try:
             await self.setup()
         except Exception as e:
-            self.logger.error(f"Failed to set up Meshgram: {e}")
+            self.logger.error(f"Failed to set up Meshgram: {e}", exc_info=True)
             return
 
-        self.logger.info("Meshgram is running ヽ(´▽`)/")
-        self.tasks = [
+        self.logger.info("Meshgram is running.")
+        self.tasks = (
             asyncio.create_task(self.message_processor.process_messages()),
             asyncio.create_task(self.meshtastic.process_thread_safe_queue()),
             asyncio.create_task(self.meshtastic.process_pending_messages()),
             asyncio.create_task(self.telegram.start_polling()),
-            asyncio.create_task(self.message_processor.check_heartbeats())
-        ]
+        )
         try:
             await asyncio.gather(*self.tasks)
         except asyncio.CancelledError:
             self.logger.info("Received cancellation signal.")
         except Exception as e:
-            self.logger.error(f"An error occurred: {e}", exc_info=True)
+            self.logger.error(f"Unexpected error: {e}", exc_info=True)
         finally:
             await self.shutdown()
 
-    async def shutdown(self) -> None:
-        self.logger.info("Shutting down meshgram...")
-        for task in self.tasks:
-            if not task.done():
-                task.cancel()
-        await asyncio.gather(*self.tasks, return_exceptions=True)
-        if self.meshtastic:
-            await self.meshtastic.close()
-        if self.telegram:
-            await self.telegram.close()
-        if self.message_processor:
-            if hasattr(self.message_processor, 'close'):
-                await self.message_processor.close()
-            else:
-                self.logger.warning("MessageProcessor does not have a close method.")
-        self.logger.info("Meshgram shutdown complete.")
-
 async def main() -> None:
     parser = argparse.ArgumentParser(description='Meshgram: Meshtastic-Telegram Bridge')
     parser.add_argument('-c', '--config', default='config/config.yaml', help='Path to configuration file')
-    parser.add_argument('--version', action='version', version='%(prog)s 1.0.0')
     args = parser.parse_args()
 
     config = ConfigManager(args.config)
@@ -86,15 +89,14 @@ async def main() -> None:
     app = Meshgram(config)
     try:
         await app.run()
-    except KeyboardInterrupt:
-        logger.info("Received keyboard interrupt. Shutting down gracefully...")
-    except Exception as e:
-        logger.error(f"Unhandled exception: {e}", exc_info=True)
+    except ExceptionGroup as eg:
+        for i, e in enumerate(eg.exceptions, 1):
+            logger.error(f"Exception {i}: {e}", exc_info=e)
     finally:
         await app.shutdown()
 
 if __name__ == '__main__':
     try:
         asyncio.run(main())
     except KeyboardInterrupt:
-        print("\nShutdown complete.")
+        print("\nShutdown complete.")