blakeblackshear.frigate/frigate/util/config.py
Nicolas Mowen c2eac10925
Tweaks and fixes (#11541)
* Update config version to be stored inside of the config

* Don't remove items from list when navigating back

* Use video api instead of webps for live current hour filmstrip

* Check that the config file is writable

* Show camera name when camera is offline

* Show camera name when offline

* Cleanup
2024-05-26 16:49:12 -05:00

197 lines
6.7 KiB
Python

"""configuration utils."""
import logging
import os
import shutil
from typing import Optional, Union
from ruamel.yaml import YAML
from frigate.const import CONFIG_DIR, EXPORT_DIR
logger = logging.getLogger(__name__)
CURRENT_CONFIG_VERSION = 0.14
def migrate_frigate_config(config_file: str):
"""handle migrating the frigate config."""
logger.info("Checking if frigate config needs migration...")
if not os.access(config_file, mode=os.W_OK):
logger.error("Config file is read-only, unable to migrate config file.")
return
yaml = YAML()
yaml.indent(mapping=2, sequence=4, offset=2)
with open(config_file, "r") as f:
config: dict[str, dict[str, any]] = yaml.load(f)
previous_version = config.get("version", 0.13)
if previous_version == CURRENT_CONFIG_VERSION:
logger.info("frigate config does not need migration...")
return
logger.info("copying config as backup...")
shutil.copy(config_file, os.path.join(CONFIG_DIR, "backup_config.yaml"))
if previous_version < 0.14:
logger.info(f"Migrating frigate config from {previous_version} to 0.14...")
new_config = migrate_014(config)
with open(config_file, "w") as f:
yaml.dump(new_config, f)
previous_version = 0.14
logger.info("Migrating export file names...")
for file in os.listdir(EXPORT_DIR):
if "@" not in file:
continue
new_name = file.replace("@", "_")
os.rename(
os.path.join(EXPORT_DIR, file), os.path.join(EXPORT_DIR, new_name)
)
logger.info("Finished frigate config migration...")
def migrate_014(config: dict[str, dict[str, any]]) -> dict[str, dict[str, any]]:
"""Handle migrating frigate config to 0.14"""
# migrate record.events.required_zones to review.alerts.required_zones
new_config = config.copy()
global_required_zones = (
config.get("record", {}).get("events", {}).get("required_zones", [])
)
if global_required_zones:
# migrate to new review config
if not new_config.get("review"):
new_config["review"] = {}
if not new_config["review"].get("alerts"):
new_config["review"]["alerts"] = {}
if not new_config["review"]["alerts"].get("required_zones"):
new_config["review"]["alerts"]["required_zones"] = global_required_zones
# remove record required zones config
del new_config["record"]["events"]["required_zones"]
# remove record altogether if there is not other config
if not new_config["record"]["events"]:
del new_config["record"]["events"]
if not new_config["record"]:
del new_config["record"]
if new_config.get("ui", {}).get("use_experimental"):
del new_config["ui"]["experimental"]
if not new_config["ui"]:
del new_config["ui"]
# remove rtmp
if new_config.get("ffmpeg", {}).get("output_args", {}).get("rtmp"):
del new_config["ffmpeg"]["output_args"]["rtmp"]
if new_config.get("rtmp"):
del new_config["rtmp"]
for name, camera in config.get("cameras", {}).items():
camera_config: dict[str, dict[str, any]] = camera.copy()
required_zones = (
camera_config.get("record", {}).get("events", {}).get("required_zones", [])
)
if required_zones:
# migrate to new review config
if not camera_config.get("review"):
camera_config["review"] = {}
if not camera_config["review"].get("alerts"):
camera_config["review"]["alerts"] = {}
if not camera_config["review"]["alerts"].get("required_zones"):
camera_config["review"]["alerts"]["required_zones"] = required_zones
# remove record required zones config
del camera_config["record"]["events"]["required_zones"]
# remove record altogether if there is not other config
if not camera_config["record"]["events"]:
del camera_config["record"]["events"]
if not camera_config["record"]:
del camera_config["record"]
# remove rtmp
if camera_config.get("ffmpeg", {}).get("output_args", {}).get("rtmp"):
del camera_config["ffmpeg"]["output_args"]["rtmp"]
if camera_config.get("rtmp"):
del camera_config["rtmp"]
new_config["cameras"][name] = camera_config
new_config["version"] = 0.14
return new_config
def get_relative_coordinates(
mask: Optional[Union[str, list]], frame_shape: tuple[int, int]
) -> Union[str, list]:
# masks and zones are saved as relative coordinates
# we know if any points are > 1 then it is using the
# old native resolution coordinates
if mask:
if isinstance(mask, list) and any(x > "1.0" for x in mask[0].split(",")):
relative_masks = []
for m in mask:
points = m.split(",")
if any(x > "1.0" for x in points):
rel_points = []
for i in range(0, len(points), 2):
x = int(points[i])
y = int(points[i + 1])
if x > frame_shape[1] or y > frame_shape[0]:
logger.error(
f"Not applying mask due to invalid coordinates. {x},{y} is outside of the detection resolution {frame_shape[1]}x{frame_shape[0]}. Use the editor in the UI to correct the mask."
)
continue
rel_points.append(
f"{round(x / frame_shape[1], 3)},{round(y / frame_shape[0], 3)}"
)
relative_masks.append(",".join(rel_points))
else:
relative_masks.append(m)
mask = relative_masks
elif isinstance(mask, str) and any(x > "1.0" for x in mask.split(",")):
points = mask.split(",")
rel_points = []
for i in range(0, len(points), 2):
x = int(points[i])
y = int(points[i + 1])
if x > frame_shape[1] or y > frame_shape[0]:
logger.error(
f"Not applying mask due to invalid coordinates. {x},{y} is outside of the detection resolution {frame_shape[1]}x{frame_shape[0]}. Use the editor in the UI to correct the mask."
)
return []
rel_points.append(
f"{round(x / frame_shape[1], 3)},{round(y / frame_shape[0], 3)}"
)
mask = ",".join(rel_points)
return mask
return mask