Merge pull request #233 from probberechts/feat/opta-shot-result-coord…

…inates

Add Opta shot result coordinates

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

@@ -10,6 +10,7 @@
     Team,
     Period,
     Point,
+    Point3D,
     BallState,
     DatasetFlag,
     Orientation,
@@ -261,11 +262,7 @@ def _parse_pass(raw_qualifiers: Dict[int, str], outcome: int) -> Dict:
         result = PassResult.COMPLETE
     else:
         result = PassResult.INCOMPLETE
-    receiver_coordinates = Point(
-        x=float(raw_qualifiers[140]) if 140 in raw_qualifiers else 0,
-        y=float(raw_qualifiers[141]) if 141 in raw_qualifiers else 0,
-    )
-
+    receiver_coordinates = _get_end_coordinates(raw_qualifiers)
     qualifiers = _get_event_qualifiers(raw_qualifiers)
 
     return dict(
@@ -277,13 +274,11 @@ def _parse_pass(raw_qualifiers: Dict[int, str], outcome: int) -> Dict:
     )
 
 
-def _parse_offside_pass(raw_qualifiers: List) -> Dict:
+def _parse_offside_pass(raw_qualifiers: Dict[int, str]) -> Dict:
     qualifiers = _get_event_qualifiers(raw_qualifiers)
     return dict(
         result=PassResult.OFFSIDE,
-        receiver_coordinates=Point(
-            x=float(raw_qualifiers[140]), y=float(raw_qualifiers[141])
-        ),
+        receiver_coordinates=_get_end_coordinates(raw_qualifiers),
         receiver_player=None,
         receive_timestamp=None,
         qualifiers=qualifiers,
@@ -298,11 +293,11 @@ def _parse_take_on(outcome: int) -> Dict:
     return dict(result=result)
 
 
-def _parse_clearance(raw_qualifiers: List) -> Dict:
+def _parse_clearance(raw_qualifiers: Dict[int, str]) -> Dict:
     return dict(qualifiers=_get_event_qualifiers(raw_qualifiers))
 
 
-def _parse_card(raw_qualifiers: List) -> Dict:
+def _parse_card(raw_qualifiers: Dict[int, str]) -> Dict:
     qualifiers = _get_event_qualifiers(raw_qualifiers)
 
     if EVENT_QUALIFIER_RED_CARD in qualifiers:
@@ -317,7 +312,7 @@ def _parse_card(raw_qualifiers: List) -> Dict:
     return dict(result=None, qualifiers=qualifiers, card_type=card_type)
 
 
-def _parse_formation_change(raw_qualifiers: List) -> Dict:
+def _parse_formation_change(raw_qualifiers: Dict[int, str]) -> Dict:
     formation_id = int(raw_qualifiers[EVENT_QUALIFIER_TEAM_FORMATION])
     formation = formations[formation_id]
 
@@ -347,19 +342,41 @@ def _parse_shot(
         result = None
 
     qualifiers = _get_event_qualifiers(raw_qualifiers)
+    result_coordinates = _get_end_coordinates(raw_qualifiers)
+    if result == ShotResult.OWN_GOAL:
+        if isinstance(result_coordinates, Point3D):
+            result_coordinates = Point3D(
+                x=100 - result_coordinates.x,
+                y=100 - result_coordinates.y,
+                z=result_coordinates.z,
+            )
+        elif isinstance(result_coordinates, Point):
+            result_coordinates = Point(
+                x=100 - result_coordinates.x,
+                y=100 - result_coordinates.y,
+            )
 
-    return dict(coordinates=coordinates, result=result, qualifiers=qualifiers)
+    return dict(
+        coordinates=coordinates,
+        result=result,
+        result_coordinates=result_coordinates,
+        qualifiers=qualifiers,
+    )
 
 
-def _parse_goalkeeper_events(raw_qualifiers: List, type_id: int) -> Dict:
+def _parse_goalkeeper_events(
+    raw_qualifiers: Dict[int, str], type_id: int
+) -> Dict:
     qualifiers = _get_event_qualifiers(raw_qualifiers)
     goalkeeper_qualifiers = _get_goalkeeper_qualifiers(type_id)
     qualifiers.extend(goalkeeper_qualifiers)
 
     return dict(result=None, qualifiers=qualifiers)
 
 
-def _parse_duel(raw_qualifiers: List, type_id: int, outcome: int) -> Dict:
+def _parse_duel(
+    raw_qualifiers: Dict[int, str], type_id: int, outcome: int
+) -> Dict:
     qualifiers = _get_event_qualifiers(raw_qualifiers)
     if type_id == EVENT_TYPE_TACKLE:
         qualifiers.extend([DuelQualifier(value=DuelType.GROUND)])
@@ -387,7 +404,7 @@ def _parse_duel(raw_qualifiers: List, type_id: int, outcome: int) -> Dict:
 
 
 def _parse_interception(
-    raw_qualifiers: List, team: Team, next_event: ObjectifiedElement
+    raw_qualifiers: Dict[int, str], team: Team, next_event: ObjectifiedElement
 ) -> Dict:
     qualifiers = _get_event_qualifiers(raw_qualifiers)
     result = InterceptionResult.SUCCESS
@@ -472,7 +489,32 @@ def _team_from_xml_elm(team_elm, f7_root) -> Team:
     return team
 
 
-def _get_event_qualifiers(raw_qualifiers: List) -> List[Qualifier]:
+def _get_end_coordinates(raw_qualifiers: Dict[int, str]) -> Optional[Point]:
+    x, y, z = None, None, None
+    # pass
+    if 140 in raw_qualifiers:
+        x = float(raw_qualifiers[140])
+    if 141 in raw_qualifiers:
+        y = float(raw_qualifiers[141])
+    # blocked shot
+    if 146 in raw_qualifiers:
+        x = float(raw_qualifiers[146])
+    if 147 in raw_qualifiers:
+        y = float(raw_qualifiers[147])
+    # passed the goal line
+    if 102 in raw_qualifiers:
+        x = float(100)
+        y = float(raw_qualifiers[102])
+    if 103 in raw_qualifiers:
+        z = float(raw_qualifiers[103])
+    if x is not None and y is not None and z is not None:
+        return Point3D(x=x, y=y, z=z)
+    if x is not None and y is not None:
+        return Point(x=x, y=y)
+    return None
+
+
+def _get_event_qualifiers(raw_qualifiers: Dict[int, str]) -> List[Qualifier]:
     qualifiers = []
     qualifiers.extend(_get_event_setpiece_qualifiers(raw_qualifiers))
     qualifiers.extend(_get_event_bodypart_qualifiers(raw_qualifiers))
@@ -482,7 +524,9 @@ def _get_event_qualifiers(raw_qualifiers: List) -> List[Qualifier]:
     return qualifiers
 
 
-def _get_event_pass_qualifiers(raw_qualifiers: List) -> List[Qualifier]:
+def _get_event_pass_qualifiers(
+    raw_qualifiers: Dict[int, str]
+) -> List[Qualifier]:
     qualifiers = []
     if EVENT_QUALIFIER_CROSS in raw_qualifiers:
         qualifiers.append(PassQualifier(value=PassType.CROSS))
@@ -503,7 +547,9 @@ def _get_event_pass_qualifiers(raw_qualifiers: List) -> List[Qualifier]:
     return qualifiers
 
 
-def _get_event_setpiece_qualifiers(raw_qualifiers: List) -> List[Qualifier]:
+def _get_event_setpiece_qualifiers(
+    raw_qualifiers: Dict[int, str]
+) -> List[Qualifier]:
     qualifiers = []
     if EVENT_QUALIFIER_CORNER_KICK in raw_qualifiers:
         qualifiers.append(SetPieceQualifier(value=SetPieceType.CORNER_KICK))
@@ -523,7 +569,9 @@ def _get_event_setpiece_qualifiers(raw_qualifiers: List) -> List[Qualifier]:
     return qualifiers
 
 
-def _get_event_bodypart_qualifiers(raw_qualifiers: List) -> List[Qualifier]:
+def _get_event_bodypart_qualifiers(
+    raw_qualifiers: Dict[int, str]
+) -> List[Qualifier]:
     qualifiers = []
     if EVENT_QUALIFIER_HEAD_PASS in raw_qualifiers:
         qualifiers.append(BodyPartQualifier(value=BodyPart.HEAD))
@@ -538,7 +586,9 @@ def _get_event_bodypart_qualifiers(raw_qualifiers: List) -> List[Qualifier]:
     return qualifiers
 
 
-def _get_event_card_qualifiers(raw_qualifiers: List) -> List[Qualifier]:
+def _get_event_card_qualifiers(
+    raw_qualifiers: Dict[int, str]
+) -> List[Qualifier]:
     qualifiers = []
     if EVENT_QUALIFIER_RED_CARD in raw_qualifiers:
         qualifiers.append(CardQualifier(value=CardType.RED))
@@ -571,7 +621,7 @@ def _get_goalkeeper_qualifiers(type_id: int) -> List[Qualifier]:
 
 
 def _get_event_counter_attack_qualifiers(
-    raw_qualifiers: List,
+    raw_qualifiers: Dict[int, str],
 ) -> List[Qualifier]:
     qualifiers = []
     if EVENT_QUALIFIER_COUNTER_ATTACK in raw_qualifiers:

kloppy/tests/test_opta.py

@@ -17,7 +17,11 @@
     GoalkeeperActionType,
     DuelQualifier,
     DuelType,
+    ShotResult,
+    SetPieceQualifier,
     CounterAttackQualifier,
+    BodyPartQualifier,
+    Point3D,
 )
 
 from kloppy.domain.models.event import EventType
@@ -159,6 +163,38 @@ def test_correct_deserialization(self, f7_data: str, f24_data: str):
             == DuelType.GROUND
         )
 
+    def test_shot(self, f7_data: str, f24_data: str):
+        dataset = opta.load(
+            f24_data=f24_data,
+            f7_data=f7_data,
+            event_types=["shot"],
+            coordinates="opta",
+        )
+        assert len(dataset.events) == 2
+
+        shot = dataset.get_event_by_id("2318695229")
+        # A shot event should have a result
+        assert shot.result == ShotResult.GOAL
+        # A shot event should have end coordinates
+        assert shot.result_coordinates == Point3D(100.0, 47.8, 2.5)
+        # A shot event should have a body part
+        assert (
+            shot.get_qualifier_value(BodyPartQualifier) == BodyPart.LEFT_FOOT
+        )
+
+    def test_own_goal(self, f7_data: str, f24_data: str):
+        dataset = opta.load(
+            f24_data=f24_data,
+            f7_data=f7_data,
+            event_types=["shot"],
+            coordinates="opta",
+        )
+
+        own_goal = dataset.get_event_by_id("2318697001")
+        assert own_goal.result == ShotResult.OWN_GOAL
+        # Use the inverse coordinates of the goal location
+        assert own_goal.result_coordinates == Point3D(0.0, 100 - 45.6, 1.9)
+
     def test_correct_normalized_deserialization(
         self, f7_data: str, f24_data: str
     ):