Merge pull request #230 from my-game-plan/feature/add_interception_event

Add InterceptionEvent

kloppy/domain/models/event.py

@@ -156,6 +156,28 @@ def is_success(self):
         return self == self.WON
 
 
+class InterceptionResult(ResultType):
+    """
+    InterceptionResult
+
+    Attributes:
+        SUCCESS (InterceptionResult): An interception that gains possession of the ball (without going out of bounds)
+        LOST (InterceptionResult): An interception by the defending team that knocked the ball to an attacker
+        OUT (InterceptionResult): An interception that knocked the ball out of bounds
+    """
+
+    SUCCESS = "SUCCESS"
+    LOST = "LOST"
+    OUT = "OUT"
+
+    @property
+    def is_success(self):
+        """
+        Returns if the interception was successful
+        """
+        return self == self.SUCCESS
+
+
 class CardType(Enum):
     """
     CardType
@@ -180,6 +202,7 @@ class EventType(Enum):
         TAKE_ON (EventType):
         CARRY (EventType):
         CLEARANCE (EventType):
+        INTERCEPTION (EventType):
         DUEL (EventType):
         SUBSTITUTION (EventType):
         CARD (EventType):
@@ -200,6 +223,7 @@ class EventType(Enum):
     TAKE_ON = "TAKE_ON"
     CARRY = "CARRY"
     CLEARANCE = "CLEARANCE"
+    INTERCEPTION = "INTERCEPTION"
     DUEL = "DUEL"
     SUBSTITUTION = "SUBSTITUTION"
     CARD = "CARD"
@@ -768,6 +792,21 @@ class CarryEvent(Event):
     event_name: str = "carry"
 
 
+@dataclass(repr=False)
+@docstring_inherit_attributes(Event)
+class InterceptionEvent(Event):
+    """
+    InterceptionEvent
+
+    Attributes:
+        event_type (EventType): `EventType.INTERCEPTION` (See [`EventType`][kloppy.domain.models.event.EventType])
+        event_name (str): `"interception"`
+    """
+
+    event_type: EventType = EventType.INTERCEPTION
+    event_name: str = "interception"
+
+
 @dataclass(repr=False)
 @docstring_inherit_attributes(Event)
 class ClearanceEvent(Event):
@@ -1044,6 +1083,8 @@ def generic_record_converter(event: Event):
     "TakeOnEvent",
     "CarryEvent",
     "ClearanceEvent",
+    "InterceptionEvent",
+    "InterceptionResult",
     "SubstitutionEvent",
     "PlayerOnEvent",
     "PlayerOffEvent",

kloppy/domain/services/event_factory.py

@@ -12,6 +12,7 @@
     MiscontrolEvent,
     CarryEvent,
     DuelEvent,
+    InterceptionEvent,
     ClearanceEvent,
     FormationChangeEvent,
     BallOutEvent,
@@ -89,6 +90,9 @@ def build_take_on(self, **kwargs) -> TakeOnEvent:
     def build_carry(self, **kwargs) -> CarryEvent:
         return create_event(CarryEvent, **kwargs)
 
+    def build_interception(self, **kwargs) -> InterceptionEvent:
+        return create_event(InterceptionEvent, **kwargs)
+
     def build_clearance(self, **kwargs) -> ClearanceEvent:
         return create_event(ClearanceEvent, **kwargs)
 

kloppy/infra/serializers/event/opta/deserializer.py

@@ -1,8 +1,9 @@
-from typing import Tuple, Dict, List, NamedTuple, IO
+from typing import Tuple, Dict, List, NamedTuple, IO, Optional
 import logging
 from datetime import datetime
 import pytz
 from lxml import objectify
+from lxml.objectify import ObjectifiedElement
 
 from kloppy.domain import (
     EventDataset,
@@ -28,6 +29,7 @@
     Metadata,
     Player,
     Position,
+    InterceptionResult,
     RecoveryEvent,
     BallOutEvent,
     FoulCommittedEvent,
@@ -63,6 +65,7 @@
 EVENT_TYPE_TACKLE = 7
 EVENT_TYPE_AERIAL = 44
 EVENT_TYPE_50_50 = 67
+EVENT_TYPE_INTERCEPTION = 8
 EVENT_TYPE_CLEARANCE = 12
 EVENT_TYPE_SHOT_MISS = 13
 EVENT_TYPE_SHOT_POST = 14
@@ -75,6 +78,7 @@
 EVENT_TYPE_RECOVERY = 49
 EVENT_TYPE_FORMATION_CHANGE = 40
 EVENT_TYPE_BALL_TOUCH = 61
+EVENT_TYPE_BLOCKED_PASS = 74
 
 EVENT_TYPE_SAVE = 10
 EVENT_TYPE_CLAIM = 11
@@ -382,6 +386,27 @@ def _parse_duel(raw_qualifiers: List, type_id: int, outcome: int) -> Dict:
     )
 
 
+def _parse_interception(
+    raw_qualifiers: List, team: Team, next_event: ObjectifiedElement
+) -> Dict:
+    qualifiers = _get_event_qualifiers(raw_qualifiers)
+    result = InterceptionResult.SUCCESS
+
+    if next_event is not None:
+        next_event_type_id = int(next_event.attrib["type_id"])
+        if next_event_type_id in BALL_OUT_EVENTS:
+            result = InterceptionResult.OUT
+        elif (next_event_type_id in BALL_OWNING_EVENTS) and (
+            next_event.attrib["team_id"] != team.team_id
+        ):
+            result = InterceptionResult.LOST
+
+    return dict(
+        result=result,
+        qualifiers=qualifiers,
+    )
+
+
 def _parse_team_players(
     f7_root, team_ref: str
 ) -> Tuple[str, Dict[str, Dict[str, str]]]:
@@ -618,7 +643,17 @@ def deserialize(self, inputs: OptaInputs) -> EventDataset:
             ]
             possession_team = None
             events = []
-            for event_elm in game_elm.iterchildren("Event"):
+            events_list = [
+                event
+                for event in list(game_elm.iterchildren("Event"))
+                if int(event.attrib["type_id"]) != EVENT_TYPE_DELETED_EVENT
+            ]
+            for idx, event_elm in enumerate(events_list):
+                next_event_elm = (
+                    events_list[idx + 1]
+                    if (idx + 1) < len(events_list)
+                    else None
+                )
                 event_id = event_elm.attrib["id"]
                 type_id = int(event_elm.attrib["type_id"])
                 timestamp = _parse_f24_datetime(event_elm.attrib["timestamp"])
@@ -642,11 +677,6 @@ def deserialize(self, inputs: OptaInputs) -> EventDataset:
                         f"Set end of period {period.id} to {timestamp}"
                     )
                     period.end_timestamp = timestamp
-                elif type_id == EVENT_TYPE_DELETED_EVENT:
-                    logger.debug(
-                        f"Skipping event {event_id} because it is a deleted event (type id - {type_id})"
-                    )
-                    continue
                 else:
                     if not period.start_timestamp:
                         # not started yet
@@ -712,6 +742,7 @@ def deserialize(self, inputs: OptaInputs) -> EventDataset:
                         event = self.event_factory.build_take_on(
                             **take_on_event_kwargs,
                             **generic_event_kwargs,
+                            qualifiers=None,
                         )
                     elif type_id in (
                         EVENT_TYPE_SHOT_MISS,
@@ -761,6 +792,17 @@ def deserialize(self, inputs: OptaInputs) -> EventDataset:
                             **duel_event_kwargs,
                             **generic_event_kwargs,
                         )
+                    elif type_id in (
+                        EVENT_TYPE_INTERCEPTION,
+                        EVENT_TYPE_BLOCKED_PASS,
+                    ):
+                        interception_event_kwargs = _parse_interception(
+                            raw_qualifiers, team, next_event_elm
+                        )
+                        event = self.event_factory.build_interception(
+                            **interception_event_kwargs,
+                            **generic_event_kwargs,
+                        )
                     elif type_id in KEEPER_EVENTS:
                         goalkeeper_event_kwargs = _parse_goalkeeper_events(
                             raw_qualifiers, type_id

kloppy/infra/serializers/event/statsbomb/deserializer.py

@@ -16,6 +16,7 @@
     TakeOnResult,
     CarryResult,
     DuelResult,
+    InterceptionResult,
     DuelQualifier,
     DuelType,
     Metadata,
@@ -51,6 +52,7 @@
 SB_EVENT_TYPE_RECOVERY = 2
 SB_EVENT_TYPE_DUEL = 4
 SB_EVENT_TYPE_CLEARANCE = 9
+SB_EVENT_TYPE_INTERCEPTION = 10
 SB_EVENT_TYPE_DRIBBLE = 14
 SB_EVENT_TYPE_SHOT = 16
 SB_EVENT_TYPE_OWN_GOAL_AGAINST = 20
@@ -79,6 +81,8 @@
 SB_PASS_OUTCOME_OFFSIDE = 76
 SB_PASS_OUTCOME_UNKNOWN = 77
 
+SB_PASS_TYPE_ONE_TOUCH_INTERCEPTION = 64
+
 SB_PASS_HEIGHT_GROUND = 1
 SB_PASS_HEIGHT_LOW = 2
 SB_PASS_HEIGHT_HIGH = 3
@@ -92,8 +96,17 @@
 SB_SHOT_OUTCOME_SAVED_OFF_TARGET = 115
 SB_SHOT_OUTCOME_SAVED_TO_POST = 116
 
-SB_EVENT_TYPE_AERIAL_LOST = 10
-SB_EVENT_TYPE_TACKLE = 11
+SB_DUEL_TYPE_AERIAL_LOST = 10
+SB_DUEL_TYPE_TACKLE = 11
+
+SB_INTERCEPTION_OUTCOME_LOST = 1
+SB_INTERCEPTION_OUTCOME_WON = 4
+SB_INTERCEPTION_OUTCOME_LOST_IN_PLAY = 13
+SB_INTERCEPTION_OUTCOME_LOST_OUT = 14
+SB_INTERCEPTION_OUTCOME_SUCCESS = 15
+SB_INTERCEPTION_OUTCOME_SUCCESS_IN_PLAY = 16
+SB_INTERCEPTION_OUTCOME_SUCCESS_OUT = 17
+
 
 DUEL_WON_NAMES = [
     "Won",
@@ -471,6 +484,42 @@ def _parse_carry(carry_dict: Dict, fidelity_version: int) -> Dict:
     }
 
 
+def _parse_interception(raw_event: Dict) -> Dict:
+    event_type = raw_event["type"]["id"]
+    # Note: passes with interception qualifier, are always successful. Otherwise, interception that is LOST or OUT
+    result = InterceptionResult.SUCCESS
+
+    if event_type == SB_EVENT_TYPE_INTERCEPTION:
+        outcome = raw_event.get("interception", {}).get("outcome", {})
+        outcome_id = outcome.get("id")
+        if outcome_id in [
+            SB_INTERCEPTION_OUTCOME_LOST_OUT,
+            SB_INTERCEPTION_OUTCOME_SUCCESS_OUT,
+        ]:
+            result = InterceptionResult.OUT
+        elif outcome_id in [
+            SB_INTERCEPTION_OUTCOME_WON,
+            SB_INTERCEPTION_OUTCOME_SUCCESS,
+            SB_INTERCEPTION_OUTCOME_SUCCESS_IN_PLAY,
+        ]:
+            result = InterceptionResult.SUCCESS
+        elif outcome_id in [
+            SB_INTERCEPTION_OUTCOME_LOST,
+            SB_INTERCEPTION_OUTCOME_LOST_IN_PLAY,
+        ]:
+            result = InterceptionResult.LOST
+        else:
+            raise DeserializationError(
+                f"Unknown interception outcome: {raw_event['outcome']['name']}({outcome_id})"
+            )
+
+    qualifiers = []
+    body_part_qualifiers = _get_body_part_qualifiers(raw_event)
+    qualifiers.extend(body_part_qualifiers)
+
+    return {"result": result, "qualifiers": qualifiers}
+
+
 def _parse_clearance(clearance_dict: Dict) -> Dict:
     qualifiers = []
     body_part_qualifiers = _get_body_part_qualifiers(clearance_dict)
@@ -518,12 +567,12 @@ def _parse_duel(
     if event_type == SB_EVENT_TYPE_DUEL:
         duel_dict = raw_event.get("duel", {})
         type_id = duel_dict.get("type", {}).get("id")
-        if type_id == SB_EVENT_TYPE_AERIAL_LOST:
+        if type_id == SB_DUEL_TYPE_AERIAL_LOST:
             duel_qualifiers = [
                 DuelQualifier(value=DuelType.LOOSE_BALL),
                 DuelQualifier(value=DuelType.AERIAL),
             ]
-        elif type_id == SB_EVENT_TYPE_TACKLE:
+        elif type_id == SB_DUEL_TYPE_TACKLE:
             duel_qualifiers = [DuelQualifier(value=DuelType.GROUND)]
     elif event_type == SB_EVENT_TYPE_50_50:
         duel_dict = raw_event.get("50_50", {})
@@ -836,6 +885,20 @@ def deserialize(self, inputs: StatsBombInputs) -> EventDataset:
                         **pass_event_kwargs,
                         **generic_event_kwargs,
                     )
+                    # if pass is an interception, insert interception prior to pass event
+                    if "type" in raw_event["pass"]:
+                        type_id = raw_event["pass"]["type"]["id"]
+                        if type_id == SB_PASS_TYPE_ONE_TOUCH_INTERCEPTION:
+                            interception_event_kwargs = _parse_interception(
+                                raw_event=raw_event
+                            )
+                            interception_event = (
+                                self.event_factory.build_interception(
+                                    **interception_event_kwargs,
+                                    **generic_event_kwargs,
+                                )
+                            )
+                            new_events.append(interception_event)
                     new_events.append(pass_event)
                 elif event_type == SB_EVENT_TYPE_SHOT:
                     shot_event_kwargs = _parse_shot(
@@ -846,6 +909,15 @@ def deserialize(self, inputs: StatsBombInputs) -> EventDataset:
                         **generic_event_kwargs,
                     )
                     new_events.append(shot_event)
+                elif event_type == SB_EVENT_TYPE_INTERCEPTION:
+                    interception_event_kwargs = _parse_interception(
+                        raw_event=raw_event
+                    )
+                    interception_event = self.event_factory.build_interception(
+                        **interception_event_kwargs,
+                        **generic_event_kwargs,
+                    )
+                    new_events.append(interception_event)
                 elif event_type == SB_EVENT_TYPE_OWN_GOAL_AGAINST:
                     shot_event = self.event_factory.build_shot(
                         result=ShotResult.OWN_GOAL,