feat: add merge for repository configuration

Signed-off-by: Konstantin Fickel <mail@konstantinfickel.de>

src/streamer/localize/repository_configuration.py

@@ -0,0 +1,107 @@
+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
+
+
+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",
+]