12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667 |
- import enum
- import queue
- 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_prefix: str,
- topic_levels: typing.Iterable[_MQTTTopicLevel],
- mac_address: str,
- ) -> str:
- return topic_prefix + "/".join(
- mac_address if l == _MQTTTopicPlaceholder.MAC_ADDRESS else l
- for l in topic_levels
- )
- def _parse_mqtt_topic(
- *,
- topic: str,
- expected_prefix: str,
- expected_levels: typing.Collection[_MQTTTopicLevel],
- ) -> typing.Dict[_MQTTTopicPlaceholder, str]:
- if not topic.startswith(expected_prefix):
- raise ValueError(f"expected topic prefix {expected_prefix}, got topic {topic}")
- attrs: typing.Dict[_MQTTTopicPlaceholder, str] = {}
- topic_split = topic[len(expected_prefix) :].split("/")
- if len(topic_split) != len(expected_levels):
- raise ValueError(f"unexpected topic {topic}")
- for given_part, expected_part in zip(topic_split, expected_levels):
- if expected_part == _MQTTTopicPlaceholder.MAC_ADDRESS:
- attrs[_MQTTTopicPlaceholder(expected_part)] = given_part
- elif expected_part != given_part:
- raise ValueError(f"unexpected topic {topic}")
- return attrs
|