streamd/src/streamer/localize/repository_configuration.py
Konstantin Fickel 260400fa34
feat: add overwrite configuration
Signed-off-by: Konstantin Fickel <mail@konstantinfickel.de>
2026-02-01 12:09:38 +01:00

108 lines
3 KiB
Python

from __future__ import annotations
from typing import Optional
from pydantic import BaseModel
class Dimension(BaseModel):
display_name: str
comment: Optional[str] = None
propagate: bool = False
class MarkerPlacement(BaseModel):
if_with: set[str] = set()
dimension: str
value: str | None = None
overwrites: bool = True
class Marker(BaseModel):
display_name: str
placements: list[MarkerPlacement] = []
class RepositoryConfiguration(BaseModel):
dimensions: dict[str, Dimension]
markers: dict[str, Marker]
def merge_single_dimension(base: Dimension, second: Dimension) -> Dimension:
second_fields_set = getattr(second, "model_fields_set", set())
return Dimension(
display_name=second.display_name or base.display_name,
comment=base.comment if second.comment is None else second.comment,
propagate=second.propagate
if "propagate" in second_fields_set
else base.propagate,
)
def merge_dimensions(
base: dict[str, Dimension], second: dict[str, Dimension]
) -> dict[str, Dimension]:
merged: dict[str, Dimension] = dict(base)
for key, second_dimension in second.items():
if key in merged:
merged[key] = merge_single_dimension(merged[key], second_dimension)
else:
merged[key] = second_dimension
return merged
def _placement_identity(p: MarkerPlacement) -> tuple[frozenset[str], str]:
return (frozenset(p.if_with), p.dimension)
def merge_single_marker(base: Marker, second: Marker) -> Marker:
merged_display_name = second.display_name or base.display_name
merged_placements: list[MarkerPlacement] = []
seen: dict[tuple[frozenset[str], str], int] = {}
for placement in base.placements:
ident = _placement_identity(placement)
seen[ident] = len(merged_placements)
merged_placements.append(placement)
for placement in second.placements:
ident = _placement_identity(placement)
if ident in seen:
merged_placements[seen[ident]] = placement
else:
seen[ident] = len(merged_placements)
merged_placements.append(placement)
return Marker(display_name=merged_display_name, placements=merged_placements)
def merge_markers(
base: dict[str, Marker], second: dict[str, Marker]
) -> dict[str, Marker]:
merged: dict[str, Marker] = dict(base)
for key, second_marker in second.items():
if key in merged:
merged[key] = merge_single_marker(merged[key], second_marker)
else:
merged[key] = second_marker
return merged
def merge_repository_configuration(
base: RepositoryConfiguration, second: RepositoryConfiguration
) -> RepositoryConfiguration:
return RepositoryConfiguration(
dimensions=merge_dimensions(base.dimensions, second.dimensions),
markers=merge_markers(base.markers, second.markers),
)
__all__ = [
"Dimension",
"Marker",
"MarkerPlacement",
"RepositoryConfiguration",
"merge_repository_configuration",
]