Source code for nirahmq.utils

 1#  SPDX-License-Identifier: AGPL-3.0-or-later
 2#  Copyright (C) 2025  Dionisis Toulatos
 3
 4import re
 5from enum import Enum
 6from typing import Annotated
 7
 8from pydantic import AfterValidator, BaseModel as PydanticBaseModel, ConfigDict
 9
10
11# Why in gods green earth does pydantic not allow to have an optional, non-nullable field
12# that is not defined by default? Hence, this monstrosity of a custom singleton type hint below.
13class _Unset(Enum):
14    TOKEN = None
15
16
17Unset = _Unset.TOKEN
18"""A custom singleton type used by to indicate a value is unset."""
19
20type Optional[T] = T | _Unset
21
22type Required = None  # Required fields should always be defined like `Annotated[<actual type>, Required]`
23
24
25def _validate_regex_string(value: str) -> str:
26    try:
27        re.compile(value)
28        return value
29    except re.error:
30        raise ValueError(f"Invalid regex string: {value}")
31
32
33type Regex = Annotated[str, AfterValidator(_validate_regex_string)]
34"""A custom type alias used by :py:mod:`nirahmq` to validate a string is a valid regex pattern."""
35
36
[docs] 37def sanitize_string(string: str) -> str: 38 """Sanitizes a string for use as an MQTT topic. 39 40 .. note:: The sanitization is not an MQTT restriction but a Home Assistant one. 41 42 :param str string: The string to be sanitized 43 :return: The sanitized string 44 :rtype: str 45 """ 46 # Replace whitespace character between non-whitespace ones with single underscore 47 string = re.sub(r"(\S)\s+(\S)", r"\1_\2", string) 48 # Replace non-valid characters with nothing (remove them) 49 # This handles any whitespaces that are at the beginning or at the end 50 string = re.sub(r"[^a-zA-Z0-9_-]+", '', string) 51 return string
52 53
[docs] 54def clamp[T](val: T, min_: T, max_: T) -> T: 55 """Clamp a value in the range :math:`[min\\_, max\\_]`. 56 57 :param T val: The value to be clamped 58 :param T min\\_: The minimum allowed value (inclusive) 59 :param T max\\_: The maximum allowed value (inclusive) 60 :return: The clamped value 61 :rtype: T 62 """ 63 return min(max(val, min_), max_)
64 65
[docs] 66class BaseModel(PydanticBaseModel): 67 """A customized :py:class:`pydantic.BaseModel` class for use by the rest of the library.""" 68 69 model_config = ConfigDict( 70 validate_by_alias=True, 71 validate_by_name=True, 72 # use_enum_values=True, 73 validate_default=True, 74 strict=True, 75 extra='forbid' 76 )