From 86e27153f2b1b127dfd1bad45f7e8b42d3f97882 Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 16 Apr 2026 20:27:32 -0700 Subject: [PATCH] refactor(packet_capture_service): update type hints for consistency and clarity Revised type hints in the PacketCaptureService class to use lowercase 'list' and 'dict' instead of 'List' and 'Dict', enhancing code readability and consistency with Python's typing conventions. Adjusted optional parameters to use the new union syntax for better clarity. No functional changes were made. --- .../service_plugins/packet_capture_service.py | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/modules/service_plugins/packet_capture_service.py b/modules/service_plugins/packet_capture_service.py index af026e9..6d7b5ea 100644 --- a/modules/service_plugins/packet_capture_service.py +++ b/modules/service_plugins/packet_capture_service.py @@ -12,7 +12,7 @@ import logging import os import time from datetime import datetime -from typing import Any, Optional +from typing import Any # Import meshcore from meshcore import EventType @@ -207,7 +207,7 @@ class PacketCaptureService(BaseServicePlugin): config: ConfigParser object containing the configuration. Returns: - List[Dict[str, Any]]: List of configured MQTT broker dictionaries. + list[dict[str, Any]]: List of configured MQTT broker dictionaries. """ brokers = [] @@ -451,7 +451,7 @@ class PacketCaptureService(BaseServicePlugin): self.logger.info("Packet capture event handlers registered") - async def handle_rx_log_data(self, event: Any, metadata: Optional[dict[str, Any]] = None) -> None: + async def handle_rx_log_data(self, event: Any, metadata: dict[str, Any] | None = None) -> None: """Handle RX log data events (matches original script). Args: @@ -489,7 +489,7 @@ class PacketCaptureService(BaseServicePlugin): except Exception as e: self.logger.error(f"Error handling RX log data: {e}") - async def handle_raw_data(self, event: Any, metadata: Optional[dict[str, Any]] = None) -> None: + async def handle_raw_data(self, event: Any, metadata: dict[str, Any] | None = None) -> None: """Handle raw data events. Args: @@ -523,7 +523,7 @@ class PacketCaptureService(BaseServicePlugin): except Exception as e: self.logger.error(f"Error handling raw data: {e}") - def _format_packet_data(self, raw_hex: str, packet_info: dict[str, Any], payload: dict[str, Any], metadata: Optional[dict[str, Any]] = None) -> dict[str, Any]: + def _format_packet_data(self, raw_hex: str, packet_info: dict[str, Any], payload: dict[str, Any], metadata: dict[str, Any] | None = None) -> dict[str, Any]: """Format packet data to match original script's format_packet_data exactly. Args: @@ -533,7 +533,7 @@ class PacketCaptureService(BaseServicePlugin): metadata: Optional metadata dictionary. Returns: - Dict[str, Any]: Formatted packet dictionary. + dict[str, Any]: Formatted packet dictionary. """ current_time = datetime.now() timestamp = current_time.isoformat() @@ -737,7 +737,7 @@ class PacketCaptureService(BaseServicePlugin): return packet_data - async def process_packet(self, raw_hex: str, payload: dict[str, Any], metadata: Optional[dict[str, Any]] = None) -> None: + async def process_packet(self, raw_hex: str, payload: dict[str, Any], metadata: dict[str, Any] | None = None) -> None: """Process a captured packet. Decodes the packet, formats it, writes to file, and publishes to MQTT. @@ -815,7 +815,7 @@ class PacketCaptureService(BaseServicePlugin): except Exception as e: self.logger.error(f"Error processing packet: {e}") - def decode_packet(self, raw_hex: str, payload: dict[str, Any]) -> Optional[dict[str, Any]]: + def decode_packet(self, raw_hex: str, payload: dict[str, Any]) -> dict[str, Any] | None: """Decode a MeshCore packet - matches original packet_capture.py functionality. Args: @@ -823,7 +823,7 @@ class PacketCaptureService(BaseServicePlugin): payload: Payload dictionary from the event (unused in this method but kept for compatibility). Returns: - Optional[Dict[str, Any]]: Decoded packet info, or None if decoding fails. + dict[str, Any] | None: Decoded packet info, or None if decoding fails. """ try: # Remove 0x prefix if present @@ -1254,7 +1254,7 @@ class PacketCaptureService(BaseServicePlugin): else: self.logger.warning("MQTT enabled but no brokers connected") - def _resolve_topic_template(self, template: str, packet_type: str = 'packet') -> Optional[str]: + def _resolve_topic_template(self, template: str, packet_type: str = 'packet') -> str | None: """Resolve topic template with placeholders. Args: @@ -1262,7 +1262,7 @@ class PacketCaptureService(BaseServicePlugin): packet_type: Type of packet ('packet' or 'status'). Returns: - Optional[str]: Resolved topic string, or None if template is empty. + str | None: Resolved topic string, or None if template is empty. """ if not template: return None @@ -1457,7 +1457,7 @@ class PacketCaptureService(BaseServicePlugin): """Get firmware information from meshcore device (matches original script). Returns: - Dict[str, str]: Dictionary containing 'model' and 'version'. + dict[str, str]: Dictionary containing 'model' and 'version'. """ try: # During shutdown, always use cached info - don't query the device @@ -1548,14 +1548,14 @@ class PacketCaptureService(BaseServicePlugin): self.stats_supported = available return available - async def refresh_stats(self, force: bool = False) -> Optional[dict[str, Any]]: + async def refresh_stats(self, force: bool = False) -> dict[str, Any] | None: """Fetch stats from the radio and cache them for status publishing (matches original script). Args: force: Force refresh even if cache is fresh. Returns: - Optional[Dict[str, Any]]: Dictionary of stats or None if unavailable. + dict[str, Any] | None: Dictionary of stats or None if unavailable. """ if not self.stats_status_enabled: if self.debug: