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

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 = Field(description="MQTT account user name") 

23 protocol: str = Field(default="3.11", description="MQTT protocol version, v3 and v5 supported") 

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

25 

26 

27class CameraSettings(BaseModel): 

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

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

30 live_url: str | None = Field(default=None, description="URL to watch camera feed live") 

31 

32 

33class EventSettings(BaseModel): 

34 camera: str = Field(default="driveway", description="Camera name") 

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

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

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

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

39 watch_tree: bool = Field( 

40 default=False, description="Watch directory tree at path, or false for only the root watch_path directory" 

41 ) 

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

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

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

45 ) 

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

47 ocr_field_ids: list[str] = Field( 

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

49 ) 

50 

51 

52class HomeAssistantSettings(BaseModel): 

53 discovery_topic_root: str = "homeassistant" 

54 status_topic: str = Field( 

55 default="homeassistant/status", description="HomeAssistant status topic, where birth and will messages are posted" 

56 ) 

57 device_creation: bool = Field( 

58 default=True, description="Create a Home Assistant Device and associate the event sensors and images to it" 

59 ) 

60 image_entity: bool = Field(default=True, description="Create an Image entity via MQTT discovery") 

61 camera_entity: bool = Field(default=True, description="Create a Camera entity via MQTT discovery") 

62 

63 

64class DVLASettings(BaseModel): 

65 api_key: str | None = Field(default=None, description="DVLA issued API key") 

66 cache_ttl: int = Field(default=86400, description="Time to live for cached DVLA API results, in seconds") 

67 

68 

69class TrackerSettings(BaseModel): 

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

71 

72 

73class ImageSettings(BaseModel): 

74 jpeg_opts: dict[str, int | bool | float | str | tuple[int | float, int | float]] = Field( 

75 default={"quality": 30, "progressive": True, "optimize": True} 

76 ) 

77 png_opts: dict[str, int | bool | float | str | tuple] = Field(default={"quality": 30, "dpi": (60, 90), "optimize": True}) 

78 

79 

80class DimensionSettings(BaseModel): 

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

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

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

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

85 

86 

87class OCRFieldSettings(BaseModel): 

88 label: str = "ocr_field" 

89 crop: DimensionSettings | None = None 

90 invert: bool = True 

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

92 values: list[str] | None = None 

93 

94 

95class OCRSettings(BaseModel): 

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

97 

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

99 default_factory=lambda: { 

100 "hik_direction": OCRFieldSettings( 

101 label="vehicle_direction", 

102 invert=True, 

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

104 values=["Forward", "Reverse", "Unknown"], 

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

106 ) 

107 } 

108 ) 

109 

110 

111class TargetSettings(BaseModel): 

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

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

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

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

116 

117 

118class Settings(BaseSettings): 

119 model_config = SettingsConfigDict( 

120 secrets_dir="/run/secrets", 

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

122 env_nested_delimiter="__", 

123 env_ignore_empty=True, 

124 cli_avoid_json=False, 

125 ) 

126 

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

128 cameras: list[CameraSettings] = Field(default_factory=list) 

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

130 image: ImageSettings = ImageSettings() 

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

132 tracker: TrackerSettings = TrackerSettings() 

133 mqtt: MQTTSettings 

134 dvla: DVLASettings = DVLASettings() 

135 homeassistant: HomeAssistantSettings = HomeAssistantSettings() 

136 ocr: OCRSettings = OCRSettings() 

137 

138 @classmethod 

139 def settings_customise_sources( 

140 cls, 

141 settings_cls: type[BaseSettings], 

142 init_settings: PydanticBaseSettingsSource, 

143 env_settings: PydanticBaseSettingsSource, 

144 dotenv_settings: PydanticBaseSettingsSource, 

145 file_secret_settings: PydanticBaseSettingsSource, 

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

147 return ( 

148 CliSettingsSource(settings_cls, cli_parse_args=True), 

149 env_settings, 

150 dotenv_settings, 

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

152 YamlConfigSettingsSource(settings_cls), 

153 init_settings, 

154 )