Coverage for src / anpr2mqtt / settings.py: 94%
194 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-30 16:07 +0000
1import re
2import warnings
3from enum import StrEnum, auto
4from pathlib import Path
5from typing import Final, Literal
7from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
8from pydantic_settings import (
9 BaseSettings,
10 CliSettingsSource,
11 NestedSecretsSettingsSource,
12 PydanticBaseSettingsSource,
13 SettingsConfigDict,
14 YamlConfigSettingsSource,
15)
17TARGET_TYPE_PLATE: Final[str] = "plate"
20class MQTTSettings(BaseModel):
21 model_config = ConfigDict(populate_by_name=True)
23 topic_root: str = "anpr2mqtt"
24 host: str = Field(default="localhost", description="MQTT broker IP address or hostname")
25 port: int = Field(default=1883, description="MQTT broker port number")
26 user: str = Field(description="MQTT account user name")
27 protocol: str = Field(default="3.11", description="MQTT protocol version, v3 and v5 supported")
28 password: str = Field(alias="pass", description="MQTT account password")
31class CameraSettings(BaseModel):
32 name: str = Field(default="driveway", description="Camera Identifier, used to build MQTT topic names")
33 area: str | None = Field(default=None, description="Home Assistant area ID")
34 live_url: str | None = Field(default=None, description="URL to watch camera feed live")
37class AutoClearSettings(BaseModel):
38 enabled: bool = Field(default=True, description="Enable auto-clear of state after last event")
39 post_event: int = Field(default=300, description="Seconds after last event to reset state")
40 state: bool = Field(default=True, description="Auto-clear state")
41 image: bool = Field(default=False, description="Auto-clear image")
44class EventSettings(BaseModel):
45 camera: str = Field(default="driveway", description="Camera name")
46 event: str = Field(default="anpr", description="Identifier of the event, used in MQTT topic description")
47 description: str | None = Field(default=None, description="Free text description of event")
48 target_type: str = Field(default="plate", description="Type of target for this event, 'plate' if ANPR")
49 region: str | None = Field(default="UK", description="Region for applying correction, normalization rules")
50 default_description: str = Field(default="Unknown vehicle", description="Default description if no known match found")
51 icon: str | None = Field(
52 default="mdi:car-back",
53 description="Default icon to provide for Home Assistant sensors",
54 )
55 watch_path: Path = Field(default=Path(), description="File system directory to watch")
56 watch_tree: bool = Field(
57 default=False, description="Watch directory tree at path, or false for only the root watch_path directory"
58 )
59 image_name_re: re.Pattern[str] = Field(
60 default=re.compile(r"(?P<dt>[0-9]{17})_(?P<target>[A-Z0-9]+)_(?P<event>VEHICLE_DETECTION)\.(?P<ext>jpg|png|gif|jpeg)"),
61 description="Regular expression to find datetime, file extension and target from image name",
62 )
63 image_url_base: str | None = Field(default=None, description="Base URL to turn a file name into a web link")
64 ocr_field_ids: list[str] = Field(
65 default_factory=lambda: ["hik_direction"], description="OCR field definitions to find in image"
66 )
67 autoclear: AutoClearSettings = AutoClearSettings()
68 auto_match_tolerance: int = Field(
69 default=1,
70 description="Maximum tolerance for auto matching against known plates, using Levenshtein, 0 to disable fuzzy matching",
71 )
72 good_read_ttl: int = Field(
73 default=60,
74 description="Seconds to retain the last DVLA-confirmed plate per camera for mis-read correction; 0 to disable",
75 )
76 good_read_tolerance: int = Field(
77 default=2,
78 description="Max Levenshtein distance from last known-good plate to trigger correction; 0 to disable",
79 )
81 @field_validator("image_url_base")
82 @classmethod
83 def validate_image_url_base(cls, v: str | None) -> str | None:
84 if v is not None and v.endswith("/"): 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true
85 warnings.warn(
86 f"image_url_base has a trailing slash ({v!r}); this will produce double-slash URLs. "
87 "Remove the trailing slash from image_url_base.",
88 UserWarning,
89 stacklevel=2,
90 )
91 return v
93 @field_validator("image_name_re")
94 @classmethod
95 def validate_image_name_re(cls, v: re.Pattern[str]) -> re.Pattern[str]:
96 required = {"dt", "target"}
97 missing = required - set(v.groupindex)
98 if missing:
99 raise ValueError(f"image_name_re is missing required named groups: {sorted(missing)}")
100 return v
103class HomeAssistantSettings(BaseModel):
104 discovery_topic_root: str = "homeassistant"
105 status_topic: str = Field(
106 default="homeassistant/status", description="HomeAssistant status topic, where birth and will messages are posted"
107 )
108 device_creation: bool = Field(
109 default=True, description="Create a Home Assistant Device and associate the event sensors and images to it"
110 )
111 image_entity: bool = Field(default=True, description="Create an Image entity via MQTT discovery")
112 camera_entity: bool = Field(default=True, description="Create a Camera entity via MQTT discovery")
115class CacheType(StrEnum):
116 MEMORY = auto()
117 FILE = auto()
120class DVLASettings(BaseModel):
121 api_key: str | None = Field(default=None, description="DVLA issued API key")
122 cache_ttl: int = Field(default=86400, description="Time to live for cached DVLA API results, in seconds, default 1 day")
123 cache_type: CacheType = Field(default=CacheType.FILE, description="Cache implementation, MEMORY or FILE")
124 cache_dir: Path | None = Field(default=Path("/data/cache"), description="Cache directory")
125 verify_plate: str | None = Field(default=None, description="Plate to check at startup to verify API")
128class FrigateSettings(BaseModel):
129 enabled: bool = Field(default=False, description="Enable Frigate MQTT event listener")
130 topic: list[str] = Field(
131 default=["frigate/events", "frigate/tracked_object_update"], description="MQTT topic for Frigate events"
132 )
133 min_score: float = Field(default=0.70, description="Minimum plate recognition score to process (0.0->1.0)")
134 url: str | None = Field(
135 default=None, description="Frigate base URL for API snapshot fallback and UI links, e.g. http://frigate:5000"
136 )
137 cameras: list[str] | None = Field(default=None, description="Camera names to process; None means all cameras")
140class TrackerSettings(BaseModel):
141 data_dir: Path = Path("/data")
142 min_visit_gap_seconds: int = Field(
143 default=0,
144 description="Minimum seconds between recorded visits for the same target; 0 to disable (every read is a visit)",
145 )
148class ImageSettings(BaseModel):
149 jpeg_opts: dict[str, int | bool | float | str | tuple[int | float, int | float]] = Field(
150 default={"quality": 30, "progressive": True, "optimize": True}
151 )
152 png_opts: dict[str, int | bool | float | str | tuple[int, int]] = Field(
153 default={"quality": 30, "dpi": (60, 90), "optimize": True}
154 )
157class DimensionSettings(BaseModel):
158 x: int = Field(description="Horizontal position of crop box, measured from image left side")
159 y: int = Field(description="Vertical position of crop box, measured from image bottom side")
160 h: int = Field(description="Height of crop box in pixels")
161 w: int = Field(description="Width of crop box in pixels")
164class OCRFieldSettings(BaseModel):
165 label: str = "ocr_field"
166 crop: DimensionSettings | None = None
167 invert: bool = True
168 correction: dict[str, list[re.Pattern[str]]] = Field(default_factory=lambda: {})
169 values: list[str] | None = None
172class OCRSettings(BaseModel):
173 """Defaults for reading `direction` for the Hikvision DS-2CD4A25FWD-IZS"""
175 fields: dict[str, OCRFieldSettings] = Field(
176 default_factory=lambda: {
177 "hik_direction": OCRFieldSettings(
178 label="vehicle_direction",
179 invert=True,
180 crop=DimensionSettings(x=850, y=0, h=30, w=650),
181 values=["Forward", "Reverse", "Unknown"],
182 correction={"Forward": [re.compile(r"Fo.*rd")], "Reverse": [re.compile(r"Re.*rse"), re.compile(r"Bac.*rd")]},
183 )
184 }
185 )
188class Target(BaseModel):
189 id: str = Field(description="Target identifier, for example reg plate")
190 target_type: str = ""
191 group: str | None = None
192 lookup: bool | None = Field(
193 default=None, description="Lookup registration plate using API, defaults to False for configured plates"
194 )
195 description: str | None = Field(default=None, description="Target description")
196 entity_id: str | None = Field(
197 default=None, description="Entity ID to publish using Home Assistant MQTT Discovery compatible message"
198 )
199 icon: str | None = Field(
200 default=None,
201 description="Name of icon to publish, for Home Assistant should be a Material Design reference like 'mdi:car'",
202 )
203 priority: str | None = Field(default=None, description="Priority to report for use in Home Assistant notifications")
204 correction: list[str | re.Pattern[str]] = Field(default_factory=lambda: [])
206 def as_dict(self) -> dict[str, bool | str | None]:
207 result: dict[str, bool | str | None] = {
208 "dangerous": self.group == "dangerous", # backward compat
209 "description": self.description,
210 "known": self.group == "known", # backward compat
211 "priority": self.priority,
212 "target": self.id,
213 "target_type": self.target_type,
214 "entity_id": self.entity_id,
215 }
216 if self.icon: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 result["icon"] = self.icon
219 return result
222_PRIORITY_BY_GROUP: dict[str, str] = {"known": "medium", "dangerous": "critical"}
225class TargetGroup(BaseModel):
226 name: str = Field(description="Name of the target group, for example 'known'")
227 priority: str = Field(default="medium", description="Priority to report on MQTT message for sightings of this group")
228 lookup: bool = Field(
229 default=False, description="Lookup registration plate using API, defaults to False for configured plates"
230 )
231 members: list[Target] = Field(description="List of target IDs or full target definitions")
232 icon: str | None = Field(
233 default=None,
234 description="Name of icon to publish, for Home Assistant should be a Material Design reference like 'mdi:car'",
235 )
236 entity_id: str | None = Field(
237 default=None, description="Entity ID to publish using Home Assistant MQTT Discovery compatible message"
238 )
240 @field_validator("members", mode="before")
241 @classmethod
242 def coerce_member_strings(cls, v: object) -> object:
243 if not isinstance(v, list): 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true
244 return v
245 return [{"id": item} if isinstance(item, str) else item for item in v]
247 @model_validator(mode="after")
248 def apply_group_defaults(self) -> "TargetGroup":
250 for target in self.members:
251 target.group = self.name
252 if target.lookup is None:
253 target.lookup = self.lookup
254 if target.priority is None:
255 target.priority = self.priority
256 if target.icon is None:
257 target.icon = self.icon
258 if target.entity_id is None:
259 target.entity_id = self.entity_id
260 if target.description is None:
261 target.description = self.name
262 return self
265class TargetSettings(BaseModel):
266 groups: list[TargetGroup] = Field(default_factory=list)
267 known: dict[str, object] = Field(default_factory=dict, deprecated="Replaced by a 'groups' entry with name 'known'")
268 dangerous: dict[str, object] = Field(
269 default_factory=dict, deprecated="Replaced by a 'groups' entry with name 'dangerous' and priority 'critical'"
270 )
271 ignore: list[str | re.Pattern[str]] = Field(default_factory=list)
272 correction: dict[str, list[str | re.Pattern[str]]] = Field(default_factory=lambda: {})
275class Settings(BaseSettings):
276 model_config = SettingsConfigDict(
277 secrets_dir="/run/secrets",
278 yaml_file="/config/anpr2mqtt.yaml",
279 env_nested_delimiter="__",
280 env_ignore_empty=True,
281 cli_avoid_json=False,
282 )
284 log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
285 cameras: list[CameraSettings] = Field(default_factory=list)
286 events: list[EventSettings] = Field(default_factory=lambda: [])
287 image: ImageSettings = ImageSettings()
288 targets: dict[str, TargetSettings] = Field(default_factory=lambda: {})
289 tracker: TrackerSettings = TrackerSettings()
290 mqtt: MQTTSettings
291 dvla: DVLASettings = DVLASettings()
292 frigate: FrigateSettings = FrigateSettings()
293 homeassistant: HomeAssistantSettings = HomeAssistantSettings()
294 ocr: OCRSettings = OCRSettings()
296 @model_validator(mode="before")
297 @classmethod
298 def migrate_and_inject_target_type(cls, data: object) -> object:
299 """Migrate legacy 'known'/'dangerous' to TargetGroup entries and inject target_type into all members."""
300 if not isinstance(data, dict): 300 ↛ 301line 300 didn't jump to line 301 because the condition on line 300 was never true
301 return data
302 for target_type, target_settings in (data.get("targets") or {}).items():
303 if not isinstance(target_settings, dict): 303 ↛ 304line 303 didn't jump to line 304 because the condition on line 303 was never true
304 continue
305 groups: list[dict[str, object] | TargetGroup] = list(target_settings.get("groups") or [])
307 # migration of original known/dangerous fixed groups
308 existing_names = {g["name"] if isinstance(g, dict) else g.name for g in groups}
309 for group_name in ("known", "dangerous"):
310 legacy: dict[str, object] = target_settings.get(group_name) or {}
311 if not legacy or group_name in existing_names:
312 continue
313 members = []
314 for target_id, val in legacy.items():
315 if isinstance(val, str):
316 members.append({"id": target_id, "description": val, "target_type": target_type})
317 elif isinstance(val, dict): 317 ↛ 320line 317 didn't jump to line 320 because the condition on line 317 was always true
318 members.append({"id": target_id, "target_type": target_type, **val})
319 else:
320 members.append({"id": target_id, "target_type": target_type})
321 groups.append(
322 {
323 "name": group_name,
324 "priority": _PRIORITY_BY_GROUP.get(group_name, "medium"),
325 "members": members,
326 "lookup": group_name == "dangerous",
327 }
328 )
329 # normalize strings->dicts
330 target_settings["groups"] = groups
331 for group in groups:
332 if not isinstance(group, dict): 332 ↛ 333line 332 didn't jump to line 333 because the condition on line 332 was never true
333 continue
334 raw_members: list[object] = group.get("members") or [] # type: ignore[assignment]
335 for i, member in enumerate(raw_members):
336 if isinstance(member, str):
337 raw_members[i] = {"id": member, "target_type": target_type}
338 elif isinstance(member, dict) and "target_type" not in member:
339 member["target_type"] = target_type
340 return data
342 @classmethod
343 def settings_customise_sources(
344 cls,
345 settings_cls: type[BaseSettings],
346 init_settings: PydanticBaseSettingsSource,
347 env_settings: PydanticBaseSettingsSource,
348 dotenv_settings: PydanticBaseSettingsSource,
349 file_secret_settings: PydanticBaseSettingsSource,
350 ) -> tuple[PydanticBaseSettingsSource, ...]:
351 return (
352 CliSettingsSource(settings_cls, cli_parse_args=True),
353 env_settings,
354 dotenv_settings,
355 NestedSecretsSettingsSource(file_secret_settings, secrets_dir_missing="ok"),
356 YamlConfigSettingsSource(settings_cls),
357 init_settings,
358 )