feat: add initial support for positioning

Signed-off-by: Konstantin Fickel <mail@konstantinfickel.de>
This commit is contained in:
Konstantin Fickel 2025-06-22 18:02:42 +02:00
parent 28dc40ebf0
commit 0c61067db0
Signed by: kfickel
GPG key ID: A793722F9933C1A5
9 changed files with 186 additions and 3 deletions

View file

@ -0,0 +1,43 @@
from typing import Optional
from streamer.parse.shard import Shard, StreamFile
from .repostory_configuration import RepositoryConfiguration
from .localized_shard import LocalizedShard
from .extract_datetime import extract_date_from_file_name
def localize_shard(
shard: Shard, config: RepositoryConfiguration, propagated: dict[str, str]
) -> LocalizedShard:
position = {**propagated}
private_position = {}
for marker in shard.markers:
normalized_marker = marker.lower()
if marker_definition := config.markers[normalized_marker]:
dimension_name = marker_definition.dimension
dimension = config.dimensions[marker_definition.dimension]
if dimension.propagate:
position[dimension_name] = normalized_marker
else:
private_position[dimension_name] = normalized_marker
children = [localize_shard(child, config, position) for child in shard.children]
(position.update(private_position),)
return LocalizedShard(
**shard.model_dump(exclude={"children"}), location=position, children=children
)
def localize_stream_file(
stream_file: StreamFile, config: RepositoryConfiguration
) -> Optional[LocalizedShard]:
shard_date = extract_date_from_file_name(stream_file.filename)
return localize_shard(stream_file.shard, config, {"moment": shard_date.isoformat()})
__all__ = ["localize_stream_file"]