Coverage for src / anpr2mqtt / settings.py: 98%

63 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-01-11 15:35 +0000

1import re 

2from pathlib import Path 

3from typing import Final, Literal 

4 

5from pydantic import BaseModel, Field 

6from pydantic_settings import ( 

7 BaseSettings, 

8 CliSettingsSource, 

9 NestedSecretsSettingsSource, 

10 PydanticBaseSettingsSource, 

11 SettingsConfigDict, 

12 YamlConfigSettingsSource, 

13) 

14 

15TARGET_TYPE_PLATE: Final[str] = "plate" 

16 

17 

18class MQTTSettings(BaseModel): 

19 topic_root: str = "anpr2mqtt" 

20 host: str = Field(default="localhost", description="MQTT broker IP address or hostname") 

21 port: int = Field(default=1883, description="MQTT broker port number") 

22 user: str 

23 password: str = Field(alias="pass", description="MQTT account password") 

24 

25 

26class EventSettings(BaseModel): 

27 camera: str = Field(default="driveway", description="Camera Identifier, used to build MQTT topic names") 

28 event: str = Field(default="anpr", description="Identifier of the event, used in MQTT topic description") 

29 area: str | None = Field(default=None, description="Home Assistant area ID") 

30 description: str | None = Field(default=None, description="Free text description of event") 

31 target_type: str = Field(default="plate", description="Type of target for this event, 'plate' if ANPR") 

32 watch_path: Path = Field(default=Path(), description="File system directory to watch") 

33 image_name_re: re.Pattern[str] = Field( 

34 default=re.compile(r"(?P<dt>[0-9]{17})_(?P<target>[A-Z0-9]+)_(?P<event>VEHICLE_DETECTION)\.(?P<ext>jpg|png|gif|jpeg)"), 

35 description="Regular expression to find datetime, file extension and target from image name", 

36 ) 

37 image_url_base: str | None = Field(default=None, description="Base URL to turn a file name into a web link") 

38 ocr_field_ids: list[str] = Field( 

39 default_factory=lambda: ["hik_direction"], description="OCR field definitions to find in image" 

40 ) 

41 

42 

43class HomeAssistantSettings(BaseModel): 

44 discovery_topic_root: str = "homeassistant" 

45 device_creation: bool = True 

46 

47 

48class DVLASettings(BaseModel): 

49 api_key: str | None = None 

50 cache_ttl: int = 86400 

51 

52 

53class TrackerSettings(BaseModel): 

54 data_dir: Path = Path("/data") 

55 

56 

57class ImageSettings(BaseModel): 

58 jpeg_opts: dict = Field(default={"quality": 30, "progressive": True, "optimize": True}) 

59 png_opts: dict = Field(default={"quality": 30, "dpi": (60, 90), "optimize": True}) 

60 

61 

62class DimensionSettings(BaseModel): 

63 x: int = Field(description="Horizontal position of crop box, measured from image left side") 

64 y: int = Field(description="Vertical position of crop box, measured from image bottom side") 

65 h: int = Field(description="Height of crop box in pixels") 

66 w: int = Field(description="Width of crop box in pixels") 

67 

68 

69class OCRFieldSettings(BaseModel): 

70 label: str = "ocr_field" 

71 crop: DimensionSettings | None = None 

72 invert: bool = True 

73 correction: dict[str, list[re.Pattern | str]] = Field(default_factory=lambda: {}) 

74 values: list[str] | None = None 

75 

76 

77class OCRSettings(BaseModel): 

78 """Defaults for reading `direction` for the Hikvision DS-2CD4A25FWD-IZS""" 

79 

80 fields: dict[str, OCRFieldSettings] = Field( 

81 default_factory=lambda: { 

82 "hik_direction": OCRFieldSettings( 

83 label="vehicle_direction", 

84 invert=True, 

85 crop=DimensionSettings(x=850, y=0, h=30, w=650), 

86 values=["Forward", "Reverse"], 

87 correction={"Forward": [r"Fo.*rd"], "Reverse": [r"Re.*rse", r"Bac.*rd"]}, 

88 ) 

89 } 

90 ) 

91 

92 

93class TargetSettings(BaseModel): 

94 known: dict[str, str | None] = Field(default_factory=lambda: {}) 

95 dangerous: dict[str, str | None] = Field(default_factory=lambda: {}) 

96 ignore: list[str] = Field(default_factory=list) 

97 correction: dict[str, list[re.Pattern | str]] = Field(default_factory=lambda: {}) 

98 

99 

100class Settings(BaseSettings): 

101 model_config = SettingsConfigDict( 

102 secrets_dir="/run/secrets", 

103 yaml_file="/config/anpr2mqtt.yaml", 

104 env_nested_delimiter="__", 

105 env_ignore_empty=True, 

106 cli_avoid_json=False, 

107 ) 

108 

109 log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO" 

110 events: list[EventSettings] = Field(default_factory=lambda: []) 

111 image: ImageSettings = ImageSettings() 

112 targets: dict[str, TargetSettings] = Field(default_factory=lambda: {}) 

113 tracker: TrackerSettings = TrackerSettings() 

114 mqtt: MQTTSettings 

115 dvla: DVLASettings = DVLASettings() 

116 homeassistant: HomeAssistantSettings = HomeAssistantSettings() 

117 ocr: OCRSettings = OCRSettings() 

118 

119 @classmethod 

120 def settings_customise_sources( 

121 cls, 

122 settings_cls: type[BaseSettings], 

123 init_settings: PydanticBaseSettingsSource, 

124 env_settings: PydanticBaseSettingsSource, 

125 dotenv_settings: PydanticBaseSettingsSource, 

126 file_secret_settings: PydanticBaseSettingsSource, 

127 ) -> tuple[PydanticBaseSettingsSource, ...]: 

128 return ( 

129 CliSettingsSource(settings_cls, cli_parse_args=True), 

130 env_settings, 

131 dotenv_settings, 

132 NestedSecretsSettingsSource(file_secret_settings, secrets_dir_missing="ok"), 

133 YamlConfigSettingsSource(settings_cls), 

134 init_settings, 

135 )