diff --git a/kloppy/_providers/pff.py b/kloppy/_providers/pff.py index 71c2fbe6e..a2365c77d 100644 --- a/kloppy/_providers/pff.py +++ b/kloppy/_providers/pff.py @@ -1,19 +1,25 @@ -from kloppy.domain import Optional, TrackingDataset +from kloppy.domain import TrackingDataset, EventDataset +from kloppy.domain.services.event_factory import EventFactory from kloppy.infra.serializers.tracking.pff import ( - PFF_TrackingDeserializer, - PFF_TrackingInputs, + PFFTrackingDeserializer, + PFFTrackingInputs, +) +from kloppy.infra.serializers.event.pff import ( + PFFEventDeserializer, + PFFEventInputs, ) from kloppy.io import FileLike, open_as_file +from kloppy.config import get_config def load_tracking( meta_data: FileLike, roster_meta_data: FileLike, raw_data: FileLike, - sample_rate: Optional[float] = None, - limit: Optional[int] = None, - coordinates: Optional[str] = None, - only_alive: Optional[bool] = True, + sample_rate: float | None = None, + limit: int | None = None, + coordinates: str | None = None, + only_alive: bool | None = True, ) -> TrackingDataset: """ Load and deserialize tracking data from the provided metadata, roster metadata, and raw data files. @@ -30,19 +36,63 @@ def load_tracking( Returns: TrackingDataset: A deserialized TrackingDataset object containing the processed tracking data. """ - deserializer = PFF_TrackingDeserializer( + deserializer = PFFTrackingDeserializer( sample_rate=sample_rate, limit=limit, coordinate_system=coordinates, only_alive=only_alive, ) - with open_as_file(meta_data) as meta_data_fp, open_as_file( - roster_meta_data - ) as roster_meta_data_fp, open_as_file(raw_data) as raw_data_fp: + with ( + open_as_file(meta_data) as meta_data_fp, + open_as_file(roster_meta_data) as roster_meta_data_fp, + open_as_file(raw_data) as raw_data_fp, + ): return deserializer.deserialize( - inputs=PFF_TrackingInputs( + inputs=PFFTrackingInputs( meta_data=meta_data_fp, roster_meta_data=roster_meta_data_fp, raw_data=raw_data_fp, ) ) + + +def load_event( + metadata: FileLike, + players: FileLike, + raw_event_data: FileLike, + event_types: list[str] | None = None, + coordinates: str | None = None, + event_factory: EventFactory | None = None, + additional_metadata: dict = {}, +) -> EventDataset: + """ + Load PFF event data into a [`EventDataset`][kloppy.domain.models.event.EventDataset] + + Parameters: + match_metadata (FileLike): A file-like object containing metadata about the match. + roster_metadata (FileLike): filename of json containing the lineup information + raw_event_data (FileLike): filename of json containing the events + event_types (List[str], optional): A list of event types to filter the events. If None, all events are included. Defaults to None. + coordinates (str, optional): The coordinate system to use for the tracking data. Defaults to None. + event_factory: (EventFactory, optional): An optional event factory to use for creating events. If None, the default event factory is used. Defaults to None. + additional_metadata (dict, optional): Additional metadata to include in the deserialization process. Defaults to an empty dictionary. + """ + deserializer = PFFEventDeserializer( + event_types=event_types, + coordinate_system=coordinates, + event_factory=event_factory or get_config("event_factory"), + ) + + with ( + open_as_file(metadata) as metadata_fp, + open_as_file(players) as players_fp, + open_as_file(raw_event_data) as raw_event_data_fp, + ): + return deserializer.deserialize( + inputs=PFFEventInputs( + metadata=metadata_fp, + players=players_fp, + raw_event_data=raw_event_data_fp, + ), + additional_metadata=additional_metadata, + ) diff --git a/kloppy/infra/serializers/event/pff/__init__.py b/kloppy/infra/serializers/event/pff/__init__.py new file mode 100644 index 000000000..6f20f13f1 --- /dev/null +++ b/kloppy/infra/serializers/event/pff/__init__.py @@ -0,0 +1,8 @@ +"""Convert PFF event stream data to a kloppy EventDataset.""" + +from .deserializer import PFFEventDeserializer, PFFEventInputs + +__all__ = [ + "PFFEventDeserializer", + "PFFEventInputs", +] diff --git a/kloppy/infra/serializers/event/pff/deserializer.py b/kloppy/infra/serializers/event/pff/deserializer.py new file mode 100644 index 000000000..cf1fb4618 --- /dev/null +++ b/kloppy/infra/serializers/event/pff/deserializer.py @@ -0,0 +1,172 @@ +from datetime import timedelta +import json +import logging +from itertools import zip_longest +from typing import IO, NamedTuple + +from kloppy.domain import ( + DatasetFlag, + EventDataset, + FormationType, + Ground, + Metadata, + Orientation, + Period, + Player, + Provider, + Team, +) +from kloppy.domain.models.event import Event, EventType +from kloppy.domain.models.pitch import PitchDimensions, Point +from kloppy.exceptions import DeserializationError +from kloppy.infra.serializers.event.deserializer import EventDataDeserializer +from kloppy.utils import performance_logging + +from . import specification as PFF + +logger = logging.getLogger(__name__) + + +class PFFEventInputs(NamedTuple): + metadata: IO[bytes] + players: IO[bytes] + raw_event_data: IO[bytes] + + +class PFFEventDeserializer(EventDataDeserializer[PFFEventInputs]): + @property + def provider(self) -> Provider: + return Provider.PFF + + def deserialize( + self, inputs: PFFEventInputs, additional_metadata: dict + ) -> EventDataset: + # Intialize coordinate system transformer + self.transformer = self.get_transformer() + + with performance_logging("load data", logger=logger): + metadata = json.load(inputs.metadata) + players = json.load(inputs.players) + raw_events = self.load_raw_events(inputs.raw_event_data) + + with performance_logging("parse teams ans players", logger=logger): + teams = self.create_teams_and_players(metadata, players) + + with performance_logging("parse periods", logger=logger): + periods = self.create_periods(raw_events) + + with performance_logging("parse events", logger=logger): + events = [] + for raw_event in raw_events.values(): + new_events = raw_event.set_refs( + periods, teams, raw_events + ).deserialize(self.event_factory) + for event in new_events: + if self.should_include_event(event): + event = self.transformer.transform_event(event) + events.append(event) + + pff_metadata = Metadata( + teams=teams, + periods=periods, + # TODO: get pitch dimensions from a event + pitch_dimensions=self.transformer.get_to_coordinate_system().pitch_dimensions, + frame_rate=None, + orientation=Orientation.ACTION_EXECUTING_TEAM, + flags=DatasetFlag.BALL_OWNING_TEAM | DatasetFlag.BALL_STATE, + score=None, + provider=Provider.PFF, + coordinate_system=self.transformer.get_to_coordinate_system(), + **additional_metadata, + ) + dataset = EventDataset(metadata=pff_metadata, records=events) + + # TODO: add freeze frames + + return dataset + + def load_raw_events( + self, raw_event_data: IO[bytes] + ) -> dict[str, PFF.EVENT]: + raw_events = {} + events = json.load(raw_event_data) + events = sorted(events, key=lambda x: x['eventTime']) + for event in events: + event_id = ( + f"{event['gameEventId']}_{event['possessionEventId']}_{event['gameEvents']['gameEventType']}_{event['eventTime']}" + if event["possessionEventId"] is not None + else f"{event['gameEventId']}" + ) + raw_events[event_id] = PFF.event_decoder(event) + return raw_events + + def create_teams_and_players(self, metadata, players): + def create_team(team_id, team_name, ground_type): + team = Team( + team_id=str(team_id), + name=team_name, + ground=ground_type, + ) + + team.players = [ + Player( + player_id=entry["player"]["id"], + team=team, + name=entry["player"]["nickname"], + jersey_no=int(entry["shirtNumber"]), + # started=entry['started'], + starting_position=PFF.position_types_mapping[ + entry["positionGroupType"] + ], + ) + for entry in players + if entry["team"]["id"] == team_id + ] + + return team + + home_team = metadata["homeTeam"] + away_team = metadata["awayTeam"] + + home = create_team(home_team["id"], home_team["name"], Ground.HOME) + away = create_team(away_team["id"], away_team["name"], Ground.AWAY) + return [home, away] + + def create_periods(self, raw_events: dict[str, PFF.EVENT]) -> list[Period]: + half_start_events = {} + half_end_events = {} + + for event in raw_events.values(): + event_type = PFF.EVENT_TYPE( + event.raw_event["gameEvents"]["gameEventType"] + ) + period = event.raw_event["gameEvents"]["period"] + + if event_type in [ + PFF.EVENT_TYPE.FIRST_HALF_KICKOFF, + PFF.EVENT_TYPE.SECOND_HALF_KICKOFF, + PFF.EVENT_TYPE.THIRD_HALF_KICKOFF, + PFF.EVENT_TYPE.FOURTH_HALF_KICKOFF, + ]: + half_start_events[period] = event.raw_event + elif event_type == PFF.EVENT_TYPE.END_OF_HALF: + half_end_events[period] = event.raw_event + + periods = [] + + for start_event, end_event in zip_longest( + half_start_events.values(), half_end_events.values() + ): + if start_event is None or end_event is None: + raise DeserializationError( + "Failed to determine start and end time of periods." + ) + + period = Period( + id=int(start_event["gameEvents"]["period"]), + start_timestamp=timedelta(seconds=start_event["startTime"]), + end_timestamp=timedelta(seconds=end_event["startTime"]), + ) + periods.append(period) + + return periods diff --git a/kloppy/infra/serializers/event/pff/helpers.py b/kloppy/infra/serializers/event/pff/helpers.py new file mode 100644 index 000000000..8dd6c1c45 --- /dev/null +++ b/kloppy/infra/serializers/event/pff/helpers.py @@ -0,0 +1,139 @@ +from datetime import timedelta +from typing import Dict, List, Optional, Union + +from kloppy.domain import ( + ActionValue, + Event, + Frame, + Period, + Player, + PlayerData, + Point, + Point3D, + PositionType, + Team, +) +from kloppy.domain.models.event import QualifierT +from kloppy.domain.services.frame_factory import create_frame +from kloppy.exceptions import DeserializationError + + +def get_team_by_id(team_id: Optional[int], teams: list[Team]) -> Optional[Team]: + """Get a team by its id.""" + if team_id is None: + return None + if str(team_id) == teams[0].team_id: + return teams[0] + elif str(team_id) == teams[1].team_id: + return teams[1] + else: + raise DeserializationError(f"Unknown team_id {team_id}") + + +def get_period_by_id(period_id: int, periods: list[Period]) -> Period: + """Get a period by its id.""" + for period in periods: + if period.id == period_id: + return period + raise DeserializationError(f"Unknown period_id {period_id}") + + +def find_player(player_id: Union[int, str], teams: list[Team]) -> Optional[Player]: + for team in teams: + player = team.get_player_by_id(player_id) + if player is not None: + return player + +def parse_coordinates( + player: Player | None, raw_event: dict[str, object] +) -> Point | None: + """Parse PFF coordinates into a kloppy Point.""" + if player is None: + return None + + players = raw_event["homePlayers"] + raw_event["awayPlayers"] + + try: + player_dict = next( + player_dict + for player_dict in players + if str(player_dict["playerId"]) == player.player_id + ) + + return Point( + x=player_dict["x"], + y=player_dict["y"], + ) + except StopIteration: + raise DeserializationError(f"Unknown player {player}") + + +def collect_qualifiers(*qualifiers: QualifierT | None) -> list[QualifierT]: + return [q for q in qualifiers if q is not None] + + +def parse_freeze_frame( + freeze_frame: List[Dict], + home_team: Team, + away_team: Team, + event: Event, +) -> Frame: + """Parse a freeze frame into a kloppy Frame.""" + players_data = {} + + def get_player_from_freeze_frame(player_data, team, i): + if "player" in player_data: + return team.get_player_by_id(player_data["player"]["id"]) + elif player_data.get("actor"): + return event.player + elif player_data.get("keeper"): + return team.get_player_by_position( + position=PositionType.Goalkeeper, time=event.time + ) + else: + return Player( + player_id=f"T{team.team_id}-E{event.event_id}-{i}", + team=team, + jersey_no=None, + ) + + for i, freeze_frame_player in enumerate(freeze_frame): + is_teammate = (event.team == home_team) == freeze_frame_player[ + "teammate" + ] + freeze_frame_team = home_team if is_teammate else away_team + + player = get_player_from_freeze_frame( + freeze_frame_player, freeze_frame_team, i + ) + + players_data[player] = PlayerData( + coordinates=parse_coordinates( + freeze_frame_player["location"], fidelity_version + ) + ) + + if event.player not in players_data: + players_data[event.player] = PlayerData(coordinates=event.coordinates) + + FREEZE_FRAME_FPS = 29.97 + + frame_id = int( + event.period.start_timestamp.total_seconds() + + event.timestamp.total_seconds() * FREEZE_FRAME_FPS + ) + + frame = create_frame( + frame_id=frame_id, + ball_coordinates=Point3D( + x=event.coordinates.x, y=event.coordinates.y, z=0 + ), + players_data=players_data, + period=event.period, + timestamp=event.timestamp, + ball_state=event.ball_state, + ball_owning_team=event.ball_owning_team, + other_data={"visible_area": visible_area}, + ) + + return frame diff --git a/kloppy/infra/serializers/event/pff/specification.py b/kloppy/infra/serializers/event/pff/specification.py new file mode 100644 index 000000000..8730bc914 --- /dev/null +++ b/kloppy/infra/serializers/event/pff/specification.py @@ -0,0 +1,1130 @@ +from copy import deepcopy +from datetime import timedelta +from enum import Enum, EnumMeta +from typing import List, Dict, NamedTuple, Union + +from kloppy.domain import ( + BallState, + BodyPart, + BodyPartQualifier, + CardQualifier, + CardType, + CarryResult, + CounterAttackQualifier, + DuelQualifier, + DuelResult, + DuelType, + Event, + EventFactory, + ExpectedGoals, + FormationType, + GoalkeeperActionType, + GoalkeeperQualifier, + InterceptionResult, + PassQualifier, + PassResult, + PassType, + PositionType, + PostShotExpectedGoals, + SetPieceQualifier, + SetPieceType, + ShotResult, + TakeOnResult, +) +from kloppy.domain.models.event import ( + CardEvent, + FoulCommittedEvent, + UnderPressureQualifier, +) +from kloppy.exceptions import DeserializationError +from kloppy.infra.serializers.event.pff.helpers import ( + collect_qualifiers, + find_player, + get_period_by_id, + get_team_by_id, + parse_coordinates, +) + + +position_types_mapping: dict[str, PositionType] = { + "GK": PositionType.Goalkeeper, # Provider: Goalkeeper + "RB": PositionType.RightBack, # Provider: Right Back + "RCB": PositionType.RightCenterBack, # Provider: Right Center Back + "CB": PositionType.CenterBack, # Provider: Center Back + "MCB": PositionType.CenterBack, # Provider: Mid Center Back + "LCB": PositionType.LeftCenterBack, # Provider: Left Center Back + "LB": PositionType.LeftBack, # Provider: Left Back + "LWB": PositionType.LeftWingBack, # Provider: Left Wing Back + "RWB": PositionType.RightWingBack, # Provider: Right Wing Back + "D": PositionType.Defender, # Provider: Defender + "M": PositionType.Midfielder, # Provider: Midfielder + "DM": PositionType.DefensiveMidfield, # Provider: Defensive Midfield + "RM": PositionType.RightMidfield, # Provider: Right Midfield + "CM": PositionType.CenterMidfield, # Provider: Center Midfield + "LM": PositionType.LeftMidfield, # Provider: Left Midfield + "RW": PositionType.RightWing, # Provider: Right Wing + "AM": PositionType.AttackingMidfield, # Provider: Attacking Midfield + "LW": PositionType.LeftWing, # Provider: Left Wing + "CF": PositionType.Striker, # Provider: Center Forward (mapped to Striker) + "F": PositionType.Attacker, # Provider: Forward (mapped to Attacker) +} + + +class TypesEnumMeta(EnumMeta): + def __call__(cls, value, *args, **kw): + if isinstance(value, dict): + if value["id"] not in cls._value2member_map_: + raise DeserializationError( + "Unknown PFF {}: {}/{}".format( + ( + cls.__qualname__.replace("_", " ") + .replace(".", " ") + .title() + ), + value["id"], + value["name"], + ) + ) + value = cls(value["id"]) + elif value not in cls._value2member_map_: + raise DeserializationError( + "Unknown PFF {}: {}".format( + ( + cls.__qualname__.replace("_", " ") + .replace(".", " ") + .title() + ), + value, + ) + ) + return super().__call__(value, *args, **kw) + + +class END_TYPE(Enum, metaclass=TypesEnumMeta): + """ "The list of end of half types used in PFF data.""" + + FIRST_HALF_END = "FIRST" + SECOND_HALF_END = "SECOND" + THIRD_HALF_END = "F" + FOURTH_HALF_END = "S" + GAME_END = "G" + + +class EVENT_TYPE(Enum, metaclass=TypesEnumMeta): + """The list of game event types used in PFF data.""" + + FIRST_HALF_KICKOFF = "FIRSTKICKOFF" + SECOND_HALF_KICKOFF = "SECONDKICKOFF" + THIRD_HALF_KICKOFF = "THIRDKICKOFF" + FOURTH_HALF_KICKOFF = "FOURTHKICKOFF" + GAME_CLOCK_OBSERVATION = "CLK" + END_OF_HALF = "END" + GROUND = "G" + PLAYER_OFF = "OFF" + PLAYER_ON = "ON" + POSSESSION = "OTB" + BALL_OUT_OF_PLAY = "OUT" + PAUSE_OF_GAME_TIME = "PAU" + SUB = "SUB" + VIDEO = "VID" + FOUL = "FOUL" + + +class POSSESSION_EVENT_TYPE(Enum, metaclass=TypesEnumMeta): + """The list of possession event types used in PFF data.""" + + BALL_CARRY = "BC" + CHALLENGE = "CH" + CLEARANCE = "CL" + CROSS = "CR" + FOUL = "FO" + PASS = "PA" + REBOUND = "RE" + SHOT = "SH" + TOUCHES = "TC" + EVT_START = "IT" + + +class PFF_BODYPART(Enum, metaclass=TypesEnumMeta): + """The list of body parts used in PFF data.""" + + BACK = "BA" + BOTTOM = "BO" + TWO_HAND_CATCH = "CA" + CHEST = "CH" + HEAD = "HE" + LEFT_FOOT = "L" + LEFT_ARM = "LA" + LEFT_BACK_HEEL = "LB" + LEFT_SHOULDER = "LC" + LEFT_HAND = "LH" + LEFT_KNEE = "LK" + LEFT_SHIN = "LS" + LEFT_THIGH = "LT" + TWO_HAND_PALM = "PA" + TWO_HAND_PUNCH = "PU" + RIGHT_FOOT = "R" + RIGHT_ARM = "RA" + RIGHT_BACK_HEEL = "RB" + RIGHT_SHOULDER = "RC" + RIGHT_HAND = "RH" + RIGHT_KNEE = "RK" + RIGHT_SHIN = "RS" + RIGHT_THIGH = "RT" + TWO_HANDS = "TWOHANDS" + VIDEO_MISSING = "VM" + + +class PFF_SET_PIECE(Enum, metaclass=TypesEnumMeta): + """The list of set piece types used in PFF data.""" + + CORNER = 'C' + DROP_BALL = 'D' + FREE_KICK = 'F' + GOAL_KICK = 'G' + KICK_OFF = 'K' + PENALTY = 'P' + THROW_IN = 'T' + + +class FOUL_TYPE(Enum, metaclass=TypesEnumMeta): + ADVANTAGE = "A" + INFRIGEMENT = "I" + MISSED_INFRIGEMENT = "M" + + +class FOUL_OUTCOME(Enum, metaclass=TypesEnumMeta): + FIRST_YELLOW = "Y" + SECOND_YELLOW = "S" + RED = "R" + WARNING = "W" + NO_FOUL = "F" + NO_WARNING = "N" + + +class EVENT: + """Base class for PFF events. + + This class is used to deserialize PFF events into kloppy events. + This default implementation is used for all events that do not have a + specific implementation. They are deserialized into a generic event. + + Args: + raw_event: The raw JSON event. + """ + + def __init__(self, raw_event: Dict): + self.raw_event = raw_event + + @property + def game_event(self) -> Dict[str, Union[int, float, str, bool, None]]: + return self.raw_event['gameEvents'] + + @property + def possession_event(self) -> Dict[str, Union[int, float, str, bool, None]]: + return self.raw_event['possessionEvents'] + + def set_refs(self, periods, teams, events): + # temp: some PFF events do not have a 'teamId' assigned but we can get + # the team using the player id. Until this is fixed in the PFF data, + # both teams are being "carried over" in the event. + self.teams = teams + + self.period = get_period_by_id( + self.raw_event["gameEvents"]["period"], periods + ) + + self.team = get_team_by_id( + self.raw_event["gameEvents"]["teamId"], teams + ) + + self.possession_team = get_team_by_id( + self.raw_event["gameEvents"]["teamId"], teams + ) + + self.player = ( + self.team.get_player_by_id( + self.raw_event["gameEvents"]["playerId"] + ) + if self.team + and self.raw_event["gameEvents"]["playerId"] is not None + else None + ) + + self.related_events = [ + events[event_id] + for event_id in events.keys() + if event_id.split("_")[0] == str(self.raw_event["gameEventId"]) + ] + + return self + + def deserialize(self, event_factory: EventFactory) -> list[Event]: + """Deserialize the event. + + Args: + event_factory: The event factory to use to build the event. + + Returns: + A list of kloppy events. + """ + generic_event_kwargs = self._parse_generic_kwargs() + + # create events + base_events = self._create_events( + event_factory, **generic_event_kwargs + ) + + foul_events = self._create_foul( + event_factory, **generic_event_kwargs + ) + + # return events (note: order is important) + return base_events + foul_events + + def _get_set_piece_qualifier(self) -> SetPieceQualifier | None: + pff_set_piece_type = self.game_event['setpieceType'] + + if pff_set_piece_type is None or pff_set_piece_type == 'O': + return None + + pff_to_kloppy_set_piece = { + PFF_SET_PIECE.GOAL_KICK: SetPieceType.GOAL_KICK, + PFF_SET_PIECE.FREE_KICK: SetPieceType.FREE_KICK, + PFF_SET_PIECE.THROW_IN: SetPieceType.THROW_IN, + PFF_SET_PIECE.CORNER: SetPieceType.CORNER_KICK, + PFF_SET_PIECE.PENALTY: SetPieceType.PENALTY, + PFF_SET_PIECE.KICK_OFF: SetPieceType.KICK_OFF, + } + + try: + pff_set_piece = PFF_SET_PIECE(pff_set_piece_type) + set_piece_type = pff_to_kloppy_set_piece[pff_set_piece] + return SetPieceQualifier(value=set_piece_type) + except KeyError: + return None + + def _get_body_part_qualifier(self) -> BodyPartQualifier | None: + """Get the body part qualifier from the PFF body part type.""" + + pff_body_part_type = self.raw_event['possessionEvents']['bodyType'] + + if pff_body_part_type is None: + return None + + pff_to_kloppy_body_part = { + PFF_BODYPART.HEAD: BodyPart.HEAD, + + PFF_BODYPART.LEFT_FOOT: BodyPart.LEFT_FOOT, + PFF_BODYPART.LEFT_BACK_HEEL: BodyPart.LEFT_FOOT, + PFF_BODYPART.LEFT_SHIN: BodyPart.LEFT_FOOT, + PFF_BODYPART.LEFT_THIGH: BodyPart.LEFT_FOOT, + PFF_BODYPART.LEFT_KNEE: BodyPart.LEFT_FOOT, + + PFF_BODYPART.RIGHT_FOOT: BodyPart.RIGHT_FOOT, + PFF_BODYPART.RIGHT_BACK_HEEL: BodyPart.RIGHT_FOOT, + PFF_BODYPART.RIGHT_SHIN: BodyPart.RIGHT_FOOT, + PFF_BODYPART.RIGHT_THIGH: BodyPart.RIGHT_FOOT, + PFF_BODYPART.RIGHT_KNEE: BodyPart.RIGHT_FOOT, + + PFF_BODYPART.LEFT_ARM: BodyPart.LEFT_HAND, + PFF_BODYPART.LEFT_HAND: BodyPart.LEFT_HAND, + PFF_BODYPART.LEFT_SHOULDER: BodyPart.LEFT_HAND, + + PFF_BODYPART.RIGHT_ARM: BodyPart.RIGHT_HAND, + PFF_BODYPART.RIGHT_HAND: BodyPart.RIGHT_HAND, + PFF_BODYPART.RIGHT_SHOULDER: BodyPart.RIGHT_HAND, + + PFF_BODYPART.TWO_HAND_PALM: BodyPart.BOTH_HANDS, + PFF_BODYPART.TWO_HAND_CATCH: BodyPart.BOTH_HANDS, + PFF_BODYPART.TWO_HAND_PUNCH: BodyPart.BOTH_HANDS, + PFF_BODYPART.TWO_HANDS: BodyPart.BOTH_HANDS, + + PFF_BODYPART.BACK: BodyPart.OTHER, + PFF_BODYPART.BOTTOM: BodyPart.OTHER, + + PFF_BODYPART.CHEST: BodyPart.CHEST, + } + + try: + pff_body_part = PFF_BODYPART(pff_body_part_type) + body_part = pff_to_kloppy_body_part[pff_body_part] + return BodyPartQualifier(value=body_part) + except KeyError: + return None + + def _parse_generic_kwargs(self) -> dict: + return { + "period": self.period, + "timestamp": timedelta(seconds=self.raw_event["eventTime"]), + "ball_owning_team": self.possession_team, + "ball_state": BallState.DEAD, + "event_id": self.raw_event["gameEventId"], + "team": self.team, + "player": self.player, + "coordinates": parse_coordinates(self.player, self.raw_event), + "raw_event": self.raw_event, + } + + def _create_foul( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[CardEvent | FoulCommittedEvent]: + foul_type = self.raw_event["fouls"].get("foulType") + + if foul_type != FOUL_TYPE.INFRIGEMENT.value: + return [] + + card_map = { + FOUL_OUTCOME.FIRST_YELLOW: CardType.FIRST_YELLOW, + FOUL_OUTCOME.SECOND_YELLOW: CardType.SECOND_YELLOW, + FOUL_OUTCOME.RED: CardType.RED, + } + + committer_id = self.raw_event["fouls"]["finalCulpritPlayerId"] + team = next(t for t in self.teams if t.get_player_by_id(committer_id)) + + generic_event_kwargs["team"] = team + generic_event_kwargs["player"] = team.get_player_by_id(committer_id) + generic_event_kwargs["ball_state"] = BallState.DEAD + + foul_outcome = self.raw_event["fouls"]["finalFoulOutcomeType"] + card_type = card_map.get(FOUL_OUTCOME(foul_outcome)) + card_qualifier = [CardQualifier(value=card_type)] if card_type else [] + + foul = [ + event_factory.build_foul_committed( + result=None, + qualifiers=card_qualifier, + **generic_event_kwargs, + ) + ] + + if card_type: + card = [ + event_factory.build_card( + result=None, + qualifiers=None, + card_type=card_type, + **generic_event_kwargs, + ) + ] + else: + card = [] + + return foul + card + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + generic_event = event_factory.build_generic( + result=None, + qualifiers=None, + event_name=self.raw_event["gameEvents"]["gameEventType"], + **generic_event_kwargs, + ) + return [generic_event] + + +class SUBSTITUTION(EVENT): + """PFF Substitution event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + # As of now, PFF Substitution events do not set teamId. + # team = generic_event_kwargs['team'] + + player_off_id = self.raw_event["gameEvents"]["playerOffId"] + player_on_id = self.raw_event["gameEvents"]["playerOnId"] + + team = next(t for t in self.teams if t.get_player_by_id(player_off_id)) + + player_off = team.get_player_by_id(player_off_id) + player_on = team.get_player_by_id(player_on_id) + + generic_event_kwargs["team"] = team + generic_event_kwargs["player"] = player_off + + return [ + event_factory.build_substitution( + result=None, + qualifiers=None, + replacement_player=player_on, + **generic_event_kwargs, + ) + ] + + +class PLAYER_OFF(EVENT): + """PFF Player Off event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + return [ + event_factory.build_player_off( + result=None, + qualifiers=None, + **generic_event_kwargs, + ) + ] + + +class PLAYER_ON(EVENT): + """PFF Player On event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + return [ + event_factory.build_player_on( + result=None, + qualifiers=None, + **generic_event_kwargs, + ) + ] + + +class BALL_OUT(EVENT): + """PFF OUT/Ball out of play event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + return [ + event_factory.build_ball_out( + result=None, + qualifiers=None, + **generic_event_kwargs, + ) + ] + + +class POSSESSION_EVENT(EVENT): + def _parse_generic_kwargs(self) -> dict: + event_id = ( + self.raw_event["possessionEventId"] + if self.raw_event["possessionEventId"] is not None + else self.raw_event["gameEventId"] + ) + + return { + "period": self.period, + "timestamp": timedelta(seconds=self.raw_event["eventTime"]), + "ball_owning_team": self.possession_team, + "ball_state": BallState.ALIVE, + "event_id": event_id, + "team": self.team, + "player": self.player, + "coordinates": parse_coordinates(self.player, self.raw_event), + "raw_event": self.raw_event, + } + + +class PASS(POSSESSION_EVENT): + """PFF Pass event.""" + + class TYPE(Enum, metaclass=TypesEnumMeta): + CUTBACK = 'B' + CREATE_CONTEST = 'C' + FLICK_ON = 'F' + LONG_THROW = 'H' + LONG_PASS = 'L' + MISS_HIT = 'M' + BALL_OVER_THE_TOP = 'O' + STANDARD_PASS = 'S' + THROUGH_BALL = 'T' + SWITCH = 'W' + + class OUTCOME(Enum, metaclass=TypesEnumMeta): + BLOCKED = 'B' + COMPLETE = 'C' + DEFENSIVE_INTERCEPTION = 'D' + INADVERTENT_SHOT_OWN_GOAL = 'G' + INADVERTENT_SHOT_GOAL = 'I' + OUT_OF_PLAY = 'O' + STOPPAGE = 'S' + + class CROSS_TYPE(Enum, metaclass=TypesEnumMeta): + DRILLED = 'D' + FLOATED = 'F' + SWING_IN = 'I' + SWING_OUT = 'O' + PLACED = 'P' + + class CROSS_OUTCOME(Enum, metaclass=TypesEnumMeta): + BLOCKED = 'B' + COMPLETE = 'C' + DEFENSIVE_INTERCEPTION = 'D' + INADVERTENT_SHOT_GOAL = 'I' + OUT_OF_PLAY = 'O' + STOPPAGE = 'S' + UNTOUCHED = 'U' + + class HEIGHT(Enum, metaclass=TypesEnumMeta): + ABOVE_HEAD = "A" + GROUND = "G" + BETWEEN_WAIST_AND_HEAD = "H" + OFF_GROUND_BUT_BELOW_WAIST = "L" + VIDEO_MISSING = "M" + HALF_VOLLEY = "V" + + @property + def outcome(self) -> Union[OUTCOME, CROSS_OUTCOME, None]: + try: + return ( + self.OUTCOME(self.possession_event['passOutcomeType']) + or self.CROSS_OUTCOME( + self.possession_event['crossOutcomeType'] + ) + ) + except Exception: + return None + + def _pass_outcome_to_result(self) -> PassResult | None: + if self.outcome is None: + return None + + outcome_mapping = { + PASS.OUTCOME.COMPLETE: PassResult.COMPLETE, + PASS.OUTCOME.BLOCKED: PassResult.INCOMPLETE, + PASS.OUTCOME.DEFENSIVE_INTERCEPTION: PassResult.INCOMPLETE, + PASS.OUTCOME.OUT_OF_PLAY: PassResult.OUT, + PASS.OUTCOME.INADVERTENT_SHOT_OWN_GOAL: None, + PASS.OUTCOME.INADVERTENT_SHOT_GOAL: None, + PASS.OUTCOME.STOPPAGE: None, + + PASS.CROSS_OUTCOME.COMPLETE: PassResult.COMPLETE, + PASS.CROSS_OUTCOME.BLOCKED: PassResult.INCOMPLETE, + PASS.CROSS_OUTCOME.DEFENSIVE_INTERCEPTION: PassResult.INCOMPLETE, + PASS.CROSS_OUTCOME.UNTOUCHED: PassResult.INCOMPLETE, + PASS.CROSS_OUTCOME.OUT_OF_PLAY: PassResult.OUT, + PASS.CROSS_OUTCOME.INADVERTENT_SHOT_GOAL: None, + PASS.CROSS_OUTCOME.STOPPAGE: None, + } + + return outcome_mapping[self.outcome] + + + def _get_pass_qualifiers( + self, body_part: BodyPartQualifier | None + ) -> list[PassQualifier]: + qualifiers = [] + + if self.possession_event['possessionEventType'] == 'CR': + qualifiers.append(PassQualifier(value=PassType.CROSS)) + + pass_type = self.possession_event['passType'] + if pass_type is not None: + pass_type = PASS.TYPE(pass_type) + if pass_type == PASS.TYPE.THROUGH_BALL: + qualifiers.append(PassQualifier(value=PassType.THROUGH_BALL)) + if pass_type == PASS.TYPE.FLICK_ON: + qualifiers.append(PassQualifier(value=PassType.FLICK_ON)) + if pass_type == PASS.TYPE.STANDARD_PASS: + qualifiers.append(PassQualifier(value=PassType.SIMPLE_PASS)) + + if body_part is not None: + if body_part.value in [BodyPart.LEFT_HAND, BodyPart.RIGHT_HAND]: + qualifiers.append(PassQualifier(value=PassType.HAND_PASS)) + + if body_part.value == BodyPart.HEAD: + qualifiers.append(PassQualifier(value=PassType.HEAD_PASS)) + + return qualifiers + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + set_piece = self._get_set_piece_qualifier() + body_part = self._get_body_part_qualifier() + pass_quals = self._get_pass_qualifiers(body_part) + + qualifiers = collect_qualifiers(body_part, set_piece, *pass_quals) + result = self._pass_outcome_to_result() + + return [ + event_factory.build_pass( + result=result, + qualifiers=qualifiers, + **generic_event_kwargs, + ) + ] + + +class SHOT(POSSESSION_EVENT): + """PFF Shot event.""" + + class OUTCOME(Enum, metaclass=TypesEnumMeta): + ON_TARGET_BLOCKED = "B" + OFF_TARGET_BLOCKED = "C" + SAVED_OFF_TARGET = "F" + GOAL = "G" + GOAL_LINE_CLEARANCE = "L" + OFF_TARGET = "O" + ON_TARGET = "S" + + @property + def outcome(self) -> Union[OUTCOME, None]: + try: + return self.OUTCOME(self.possession_event['shotOutcomeType']) + except Exception: + return None + + def _shot_outcome_to_result(self) -> ShotResult | None: + if self.outcome is None: + return None + + outcome_map = { + SHOT.OUTCOME.ON_TARGET_BLOCKED: ShotResult.BLOCKED, + SHOT.OUTCOME.OFF_TARGET_BLOCKED: ShotResult.BLOCKED, + SHOT.OUTCOME.SAVED_OFF_TARGET: ShotResult.SAVED, + SHOT.OUTCOME.GOAL: ShotResult.GOAL, + SHOT.OUTCOME.GOAL_LINE_CLEARANCE: ShotResult.BLOCKED, + SHOT.OUTCOME.OFF_TARGET: ShotResult.OFF_TARGET, + SHOT.OUTCOME.ON_TARGET: ShotResult.SAVED, + } + + return outcome_map.get(self.outcome) + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + body_part = self._get_body_part_qualifier() + set_piece = self._get_set_piece_qualifier() + + qualifiers = collect_qualifiers(body_part, set_piece) + result = self._shot_outcome_to_result() + + return [ + event_factory.build_shot( + result=result, + qualifiers=qualifiers, + **generic_event_kwargs, + ) + ] + + +class BALL_RECEIPT(POSSESSION_EVENT): + """PFF IT event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + return [ + event_factory.build_generic( + result=None, + qualifiers=None, + event_name="RECEIVAL", + **generic_event_kwargs, + ) + ] + + +class CLEARANCE(POSSESSION_EVENT): + """PFF Clearance event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + return [ + event_factory.build_clearance( + result=None, + qualifiers=None, + **generic_event_kwargs, + ) + ] + + +class DUEL(POSSESSION_EVENT): + """PFF Challenge event.""" + + class TYPE(Enum, metaclass=TypesEnumMeta): + AERIAL_DUEL = 'A' + FROM_BEHIND = 'B' + DRIBBLE = 'D' + FIFTY = 'FIFTY' + GK_SMOTHERS = 'G' + SHIELDING = 'H' + HAND_TACKLE = 'K' # GK specific event + SLIDE_TACKLE = 'L' + SHOULDER_TO_SHOULDER = 'S' + STANDING_TACKLE = 'T' + + class OUTCOME(Enum, metaclass=TypesEnumMeta): + DISTRIBUTION_DISRUPTED = 'B' + FORCED_OUT_OF_PLAY = 'C' + DISTRIBUTES_BALL = 'D' + FOUL = 'F' + SHIELDS_IN_PLAY = 'I' + KEEPS_BALL_WITH_CONTACT = 'K' + ROLLS = 'L' + BEATS_MAN_LOSES_BALL = 'M' + NO_WIN_KEEP_BALL = 'N' + OUT_OF_PLAY = 'O' + PLAYER = 'P' + RETAIN = 'R' + SHIELDS_OUT_OF_PLAY = 'S' + + @property + def outcome(self): + try: + return self.OUTCOME( + self.raw_event['possessionEvents']['challengeOutcomeType'] + ) + except Exception: + return None + + @property + def duel_type(self): + try: + return self.TYPE( + self.raw_event['possessionEvents']['challengeType'] + ) + except Exception: + return None + + def _handle_dribble( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + if self.outcome in [ + DUEL.OUTCOME.OUT_OF_PLAY, + DUEL.OUTCOME.FORCED_OUT_OF_PLAY + ]: + result = TakeOnResult.OUT + elif self.outcome in [ + DUEL.OUTCOME.NO_WIN_KEEP_BALL, + DUEL.OUTCOME.ROLLS, + DUEL.OUTCOME.DISTRIBUTION_DISRUPTED, + DUEL.OUTCOME.DISTRIBUTES_BALL, + DUEL.OUTCOME.PLAYER, + ]: + result = TakeOnResult.INCOMPLETE + else: + result = TakeOnResult.COMPLETE + + return [ + event_factory.build_take_on( + result=result, + qualifiers=[DuelQualifier(value=DuelType.GROUND)], + **generic_event_kwargs + ) + ] + + def _handle_aerial( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + events = [] + + qualifiers = [ + DuelQualifier(value=DuelType.LOOSE_BALL), + DuelQualifier(value=DuelType.AERIAL) + ] + + aerialChallengeColumns = [ + 'homeDuelPlayerId', + 'awayDuelPlayerId', + 'challengeKeeperPlayerId', + 'additionalDuelerPlayerId' + ] + + players_involved = [ + find_player(self.possession_event[col], self.teams) + for col in aerialChallengeColumns + if self.raw_event['possessionEvents'][col] is not None + ] + + winner = find_player( + self.raw_event['possessionEvents']['challengeWinnerPlayerId'], + self.teams + ) + + if winner is None: + player_duel_result = [ + (player, DuelResult.NEUTRAL) + for player in players_involved + if player is not None + ] + else: + player_duel_result = [ + ( + player, + ( + DuelResult.WON + if player.team == winner.team + else DuelResult.LOST + ) + ) + for player in players_involved + if player is not None + ] + + for (player, result) in player_duel_result: + kwargs = deepcopy(generic_event_kwargs) + kwargs['team'] = player.team + kwargs['player'] = player + events.append( + event_factory.build_duel( + result=result, + qualifiers=qualifiers, + **kwargs, + ) + ) + + return events + + def _handle_tackle( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + events = [] + + qualifiers = [DuelQualifier(value=DuelType.GROUND)] + + if self.duel_type == DUEL.TYPE.SLIDE_TACKLE: + qualifiers.append(DuelQualifier(value=DuelType.SLIDING_TACKLE)) + elif self.duel_type == DUEL.TYPE.FIFTY: + qualifiers.append(DuelQualifier(value=DuelType.LOOSE_BALL)) + + challengeColumns = [ + 'ballCarrierPlayerId', + 'challengerPlayerId', + 'challengeKeeperPlayerId', + 'additionalDuelerPlayerId', + ] + + players_involved = [ + find_player(self.raw_event['possessionEvents'][col], self.teams) + for col in challengeColumns + if self.raw_event['possessionEvents'][col] is not None + ] + + winner = find_player( + self.raw_event['possessionEvents']['challengeWinnerPlayerId'], + self.teams + ) + + if not winner: + player_duel_result = [ + (player, DuelResult.NEUTRAL) + for player in players_involved + if player is not None + ] + else: + player_duel_result = [ + ( + player, + ( + DuelResult.WON + if player.team == winner.team + else DuelResult.LOST + ) + ) + for player in players_involved + if player is not None + ] + + for (player, result) in player_duel_result: + kwargs = deepcopy(generic_event_kwargs) + kwargs['team'] = player.team + kwargs['player'] = player + events.append( + event_factory.build_duel( + result=result, + qualifiers=qualifiers, + **kwargs, + ) + ) + + return events + + def _handle_gk( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + qualifiers = [ + GoalkeeperQualifier(value=GoalkeeperActionType.SMOTHER) + ] + + return [ + event_factory.build_goalkeeper_event( + result=None, + qualifiers=qualifiers, + **generic_event_kwargs + ) + ] + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + if self.duel_type == DUEL.TYPE.DRIBBLE: + return self._handle_dribble(event_factory, **generic_event_kwargs) + elif self.duel_type == DUEL.TYPE.AERIAL_DUEL: + return self._handle_aerial(event_factory, **generic_event_kwargs) + elif self.duel_type in [ + DUEL.TYPE.SLIDE_TACKLE, + DUEL.TYPE.FIFTY, + DUEL.TYPE.FROM_BEHIND, + DUEL.TYPE.STANDING_TACKLE, + DUEL.TYPE.SHOULDER_TO_SHOULDER, + DUEL.TYPE.SHIELDING, + DUEL.TYPE.HAND_TACKLE # GK Event + ]: + return self._handle_tackle(event_factory, **generic_event_kwargs) + elif self.duel_type == DUEL.TYPE.GK_SMOTHERS: + return self._handle_gk(event_factory, **generic_event_kwargs) + + return [event_factory.build_generic(**generic_event_kwargs)] + + +class CARRY(POSSESSION_EVENT): + """PFF Carry event.""" + + class OUTCOME(Enum, metaclass=TypesEnumMeta): + RETAIN = 'R' + STOPPAGE = 'S' + BALL_LOSS = 'L' + LEADS_INTO_CHALLENGE = 'C' + + @property + def outcome(self): + try: + return self.OUTCOME(self.possession_event['ballCarryOutcome']) + except Exception: + return None + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + if self.outcome == CARRY.OUTCOME.BALL_LOSS: + result = CarryResult.INCOMPLETE + else: + result = CarryResult.COMPLETE + + return [ + event_factory.build_carry( + result=result, + qualifiers=None, + end_timestamp=timedelta(seconds=self.raw_event['endTime']), + end_coordinates=None, + **generic_event_kwargs, + ) + ] + + +class PRESSURE(POSSESSION_EVENT): + """PFF Pressure event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + return [ + event_factory.build_pressure_event( + result=None, + qualifiers=None, + **generic_event_kwargs, + ) + ] + + +class INTERCEPTION(POSSESSION_EVENT): + """PFF Interception event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + return [ + event_factory.build_interception( + result=None, + qualifiers=None, + **generic_event_kwargs, + ) + ] + + +class RECOVERY(POSSESSION_EVENT): + """PFF Recovery event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + return [ + event_factory.build_recovery( + result=None, + qualifiers=None, + **generic_event_kwargs, + ) + ] + + +class MISCONTROL(POSSESSION_EVENT): + """PFF Miscontrol event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + return [ + event_factory.build_miscontrol( + result=None, + qualifiers=None, + **generic_event_kwargs, + ) + ] + + +class GOALKEEPER(POSSESSION_EVENT): + """PFF Goalkeeper event.""" + + def _create_events( + self, event_factory: EventFactory, **generic_event_kwargs + ) -> list[Event]: + return [ + event_factory.build_goalkeeper_event( + result=None, + qualifiers=None, + **generic_event_kwargs, + ) + ] + + +def possession_event_decoder(raw_event: dict) -> POSSESSION_EVENT: + type_to_possession_event = { + POSSESSION_EVENT_TYPE.CROSS: PASS, + POSSESSION_EVENT_TYPE.PASS: PASS, + POSSESSION_EVENT_TYPE.SHOT: SHOT, + POSSESSION_EVENT_TYPE.CLEARANCE: CLEARANCE, + POSSESSION_EVENT_TYPE.BALL_CARRY: CARRY, + POSSESSION_EVENT_TYPE.CHALLENGE: DUEL, + POSSESSION_EVENT_TYPE.REBOUND: POSSESSION_EVENT, + POSSESSION_EVENT_TYPE.TOUCHES: POSSESSION_EVENT, + POSSESSION_EVENT_TYPE.EVT_START: BALL_RECEIPT, + POSSESSION_EVENT_TYPE.FOUL: POSSESSION_EVENT, + } + + p_evt_type = raw_event["possessionEvents"]["possessionEventType"] + + if p_evt_type is None: + return POSSESSION_EVENT(raw_event) + + event_type = POSSESSION_EVENT_TYPE(p_evt_type) + event_creator = type_to_possession_event.get(event_type, POSSESSION_EVENT) + return event_creator(raw_event) + + +def event_decoder(raw_event: dict) -> EVENT: + type_to_event = { + EVENT_TYPE.POSSESSION: possession_event_decoder, + EVENT_TYPE.FIRST_HALF_KICKOFF: possession_event_decoder, + EVENT_TYPE.SECOND_HALF_KICKOFF: possession_event_decoder, + EVENT_TYPE.THIRD_HALF_KICKOFF: possession_event_decoder, + EVENT_TYPE.FOURTH_HALF_KICKOFF: possession_event_decoder, + EVENT_TYPE.GAME_CLOCK_OBSERVATION: EVENT, + EVENT_TYPE.GROUND: EVENT, + EVENT_TYPE.BALL_OUT_OF_PLAY: BALL_OUT, + EVENT_TYPE.SUB: SUBSTITUTION, + EVENT_TYPE.PLAYER_ON: PLAYER_ON, + EVENT_TYPE.PLAYER_OFF: PLAYER_OFF, + EVENT_TYPE.FOUL: EVENT, + } + + event_type = EVENT_TYPE(raw_event["gameEvents"]["gameEventType"]) + event_creator = type_to_event.get(event_type, EVENT) + return event_creator(raw_event) diff --git a/kloppy/infra/serializers/tracking/pff.py b/kloppy/infra/serializers/tracking/pff.py index 901d03747..826547076 100644 --- a/kloppy/infra/serializers/tracking/pff.py +++ b/kloppy/infra/serializers/tracking/pff.py @@ -91,13 +91,13 @@ class GameEventType: VIDEO_MISSING = "VID" -class PFF_TrackingInputs(NamedTuple): +class PFFTrackingInputs(NamedTuple): meta_data: IO[bytes] roster_meta_data: IO[bytes] raw_data: IO[bytes] -class PFF_TrackingDeserializer(TrackingDataDeserializer[PFF_TrackingInputs]): +class PFFTrackingDeserializer(TrackingDataDeserializer[PFFTrackingInputs]): def __init__( self, limit: Optional[int] = None, @@ -218,7 +218,7 @@ def __get_periods(cls, tracking, frame_rate): return periods - def deserialize(self, inputs: PFF_TrackingInputs) -> TrackingDataset: + def deserialize(self, inputs: PFFTrackingInputs) -> TrackingDataset: # Load datasets metadata = json.load(inputs.meta_data)[0] roster_meta_data = json.load(inputs.roster_meta_data) diff --git a/kloppy/pff.py b/kloppy/pff.py index a0ce8a58c..77287fc40 100644 --- a/kloppy/pff.py +++ b/kloppy/pff.py @@ -1,5 +1,5 @@ """Functions for loading PFF FC data.""" -from ._providers.pff import load_tracking +from ._providers.pff import load_tracking, load_event -__all__ = ["load_tracking"] +__all__ = ["load_tracking", "load_event"]