feat(Signality): add support for Signality tracking data (#356)

Author: DriesDeprest

kloppy/_providers/signality.py

@@ -0,0 +1,51 @@
+from typing import Optional, Iterable
+
+from kloppy.domain import TrackingDataset, CoordinateSystem
+from kloppy.infra.serializers.tracking.signality import (
+    SignalityDeserializer,
+    SignalityInputs,
+)
+from kloppy.io import FileLike, open_as_file, expand_inputs
+
+
+def load(
+    meta_data: FileLike,
+    raw_data_feeds: Iterable[FileLike],
+    venue_information: FileLike,
+    sample_rate: Optional[float] = None,
+    limit: Optional[int] = None,
+    coordinates: Optional[str] = None,
+) -> TrackingDataset:
+    """
+    Load and deserialize tracking data from multiple input files.
+
+    Args:
+        meta_data (FileLike): json feed containing metadata information of the game.
+        raw_data_feeds (Iterable[FileLike]): json feeds containing raw tracking data feeds.
+        venue_information (FileLike): json feed containing venue information where the game was played.
+        sample_rate (Optional[float]): Sampling rate to be applied during deserialization (default: None).
+        limit (Optional[int]): Limit on the number of frames to process (default: None).
+        coordinates (Optional[str]): Coordinate system to use for deserialization (default: None).
+
+    Returns:
+        TrackingDataset: A deserialized tracking dataset object.
+    """
+
+    raw_data_feeds = expand_inputs(raw_data_feeds)
+
+    deserializer = SignalityDeserializer(
+        sample_rate=sample_rate,
+        limit=limit,
+        coordinate_system=coordinates,
+    )
+
+    with open_as_file(meta_data) as meta_data_fp, open_as_file(
+        venue_information
+    ) as venue_information_fp:
+        return deserializer.deserialize(
+            inputs=SignalityInputs(
+                meta_data=meta_data_fp,
+                venue_information=venue_information_fp,
+                raw_data_feeds=raw_data_feeds,
+            )
+        )

kloppy/domain/models/common.py

@@ -112,6 +112,7 @@ class Provider(Enum):
     STATSPERFORM = "statsperform"
     HAWKEYE = "hawkeye"
     SPORTVU = "sportvu"
+    SIGNALITY = "signality"
     OTHER = "other"
 
     def __str__(self):
@@ -866,6 +867,43 @@ def pitch_dimensions(self) -> PitchDimensions:
             )
 
 
+class SignalityCoordinateSystem(ProviderCoordinateSystem):
+    @property
+    def provider(self) -> Provider:
+        return Provider.SIGNALITY
+
+    @property
+    def origin(self) -> Origin:
+        return Origin.CENTER
+
+    @property
+    def vertical_orientation(self) -> VerticalOrientation:
+        return VerticalOrientation.BOTTOM_TO_TOP
+
+    @property
+    def pitch_dimensions(self) -> PitchDimensions:
+        if self._pitch_length is not None and self._pitch_width is not None:
+            return MetricPitchDimensions(
+                x_dim=Dimension(
+                    -1 * self._pitch_length / 2, self._pitch_length / 2
+                ),
+                y_dim=Dimension(
+                    -1 * self._pitch_width / 2, self._pitch_width / 2
+                ),
+                pitch_length=self._pitch_length,
+                pitch_width=self._pitch_width,
+                standardized=False,
+            )
+        else:
+            return MetricPitchDimensions(
+                x_dim=Dimension(None, None),
+                y_dim=Dimension(None, None),
+                pitch_length=None,
+                pitch_width=None,
+                standardized=False,
+            )
+
+
 class DatafactoryCoordinateSystem(ProviderCoordinateSystem):
     @property
     def provider(self) -> Provider:
@@ -1012,6 +1050,7 @@ def build_coordinate_system(
         Provider.SECONDSPECTRUM: SecondSpectrumCoordinateSystem,
         Provider.HAWKEYE: HawkEyeCoordinateSystem,
         Provider.SPORTVU: SportVUCoordinateSystem,
+        Provider.SIGNALITY: SignalityCoordinateSystem,
     }
 
     if provider in coordinate_systems:

kloppy/domain/services/__init__.py

@@ -1,6 +1,7 @@
-from typing import List
+from collections import defaultdict, Counter
+from typing import List, Dict
 
-from kloppy.domain import AttackingDirection, Frame, Ground
+from kloppy.domain import AttackingDirection, Frame, Ground, Period
 
 from .event_factory import EventFactory, create_event
 from .transformers import DatasetTransformer, DatasetTransformerBuilder
@@ -37,10 +38,42 @@ def attacking_direction_from_frame(frame: Frame) -> AttackingDirection:
         return AttackingDirection.RTL
 
 
+def attacking_directions_from_multi_frames(
+    frames: List[Frame], periods: List[Period]
+) -> Dict[int, AttackingDirection]:
+    """
+    with only partial tracking data we cannot rely on a single frame to
+    infer the attacking directions as a simple average of only some players
+    x-coords might not reflect the attacking direction.
+    """
+    attacking_directions = {}
+
+    # Group attacking directions by period ID
+    period_direction_map = defaultdict(list)
+    for frame in frames:
+        if len(frame.players_data) > 0:
+            direction = attacking_direction_from_frame(frame)
+        else:
+            direction = AttackingDirection.NOT_SET
+        period_direction_map[frame.period.id].append(direction)
+
+    # Determine the most common attacking direction for each period
+    for period in periods:
+        period_id = period.id
+        if period_id in period_direction_map:
+            count = Counter(period_direction_map[period_id])
+            attacking_directions[period_id] = count.most_common(1)[0][0]
+        else:
+            attacking_directions[period_id] = AttackingDirection.NOT_SET
+
+    return attacking_directions
+
+
 __all__ = [
     "DatasetTransformer",
     "DatasetTransformerBuilder",
     "EventFactory",
     "create_event",
     "attacking_direction_from_frame",
+    "attacking_directions_from_multi_frames",
 ]