refactor: move a few definitions to switchbot_mqtt/_utils.py

switchbot_mqtt/__init__.py

@@ -17,15 +17,8 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import abc
-import argparse
-import collections
-import enum
-import json
 import logging
-import os
-import pathlib
 import queue
-import re
 import shlex
 import typing
 
@@ -33,47 +26,20 @@ import bluepy.btle
 import paho.mqtt.client
 import switchbot
 
-_LOGGER = logging.getLogger(__name__)
-
-_MAC_ADDRESS_REGEX = re.compile(r"^[0-9a-f]{2}(:[0-9a-f]{2}){5}$")
-
-
-class _MQTTTopicPlaceholder(enum.Enum):
-    MAC_ADDRESS = "MAC_ADDRESS"
+from switchbot_mqtt._utils import (
+    _join_mqtt_topic_levels,
+    _mac_address_valid,
+    _MQTTTopicLevel,
+    _MQTTTopicPlaceholder,
+    _QueueLogHandler,
+)
 
+_LOGGER = logging.getLogger(__name__)
 
-_MQTTTopicLevel = typing.Union[str, _MQTTTopicPlaceholder]
 # "homeassistant" for historic reason, may be parametrized in future
 _MQTT_TOPIC_LEVELS_PREFIX: typing.List[_MQTTTopicLevel] = ["homeassistant"]
 
 
-def _join_mqtt_topic_levels(
-    topic_levels: typing.List[_MQTTTopicLevel], mac_address: str
-) -> str:
-    return "/".join(
-        mac_address if l == _MQTTTopicPlaceholder.MAC_ADDRESS else typing.cast(str, l)
-        for l in topic_levels
-    )
-
-
-def _mac_address_valid(mac_address: str) -> bool:
-    return _MAC_ADDRESS_REGEX.match(mac_address.lower()) is not None
-
-
-class _QueueLogHandler(logging.Handler):
-    """
-    logging.handlers.QueueHandler drops exc_info
-    """
-
-    # TypeError: 'type' object is not subscriptable
-    def __init__(self, log_queue: "queue.Queue[logging.LogRecord]") -> None:
-        self.log_queue = log_queue
-        super().__init__()
-
-    def emit(self, record: logging.LogRecord) -> None:
-        self.log_queue.put(record)
-
-
 class _MQTTCallbackUserdata:
     # pylint: disable=too-few-public-methods; @dataclasses.dataclass when python_requires>=3.7
     def __init__(

switchbot_mqtt/_utils.py

@@ -0,0 +1,59 @@
+# switchbot-mqtt - MQTT client controlling SwitchBot button & curtain automators,
+# compatible with home-assistant.io's MQTT Switch & Cover platform
+#
+# Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import enum
+import logging
+import queue  # pylint: disable=unused-import; in type hint
+import re
+import typing
+
+_MAC_ADDRESS_REGEX = re.compile(r"^[0-9a-f]{2}(:[0-9a-f]{2}){5}$")
+
+
+def _mac_address_valid(mac_address: str) -> bool:
+    return _MAC_ADDRESS_REGEX.match(mac_address.lower()) is not None
+
+
+class _MQTTTopicPlaceholder(enum.Enum):
+    MAC_ADDRESS = "MAC_ADDRESS"
+
+
+_MQTTTopicLevel = typing.Union[str, _MQTTTopicPlaceholder]
+
+
+def _join_mqtt_topic_levels(
+    topic_levels: typing.List[_MQTTTopicLevel], mac_address: str
+) -> str:
+    return "/".join(
+        mac_address if l == _MQTTTopicPlaceholder.MAC_ADDRESS else typing.cast(str, l)
+        for l in topic_levels
+    )
+
+
+class _QueueLogHandler(logging.Handler):
+    """
+    logging.handlers.QueueHandler drops exc_info
+    """
+
+    # TypeError: 'type' object is not subscriptable
+    def __init__(self, log_queue: "queue.Queue[logging.LogRecord]") -> None:
+        self.log_queue = log_queue
+        super().__init__()
+
+    def emit(self, record: logging.LogRecord) -> None:
+        self.log_queue.put(record)

tests/test_mac_address.py

@@ -17,8 +17,7 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import pytest
-
-import switchbot_mqtt
+import switchbot_mqtt._utils
 
 
 @pytest.mark.parametrize(
@@ -33,4 +32,4 @@ import switchbot_mqtt
 )
 def test__mac_address_valid(mac_address, valid):
     # pylint: disable=protected-access
-    assert switchbot_mqtt._mac_address_valid(mac_address) == valid
+    assert switchbot_mqtt._utils._mac_address_valid(mac_address) == valid