Intermediate commit

kloppy/domain/models/event.py

@@ -133,6 +133,26 @@ def is_success(self):
  return self == self.COMPLETE
 
 
+class DuelResult(ResultType):
+ """
+ DuelResult
+
+ Attributes:
+ WON (DuelResult): When winning the duel (player touching the ball first)
+ LOST (DuelResult): When losing the duel (opponent touches the ball first)
+ """
+
+ WON = "WON"
+ LOST = "LOST"
+
+ @property
+ def is_success(self):
+ """
+ Returns if the duel was won
+ """
+ return self == self.WON
+
+
 class CardType(Enum):
  """
  CardType
@@ -156,6 +176,7 @@ class EventType(Enum):
  SHOT (EventType):
  TAKE_ON (EventType):
  CARRY (EventType):
+ DUEL (EventType):
  SUBSTITUTION (EventType):
  CARD (EventType):
  PLAYER_ON (EventType):
@@ -172,6 +193,7 @@ class EventType(Enum):
  SHOT = "SHOT"
  TAKE_ON = "TAKE_ON"
  CARRY = "CARRY"
+ DUEL = "DUEL"
  SUBSTITUTION = "SUBSTITUTION"
  CARD = "CARD"
  PLAYER_ON = "PLAYER_ON"
@@ -354,6 +376,30 @@ class GoalkeeperActionQualifier(EnumQualifier):
  value: GoalkeeperAction
 
 
+class DuelType(Enum):
+ """
+ DuelType
+
+ Attributes:
+ AERIAL (DuelType): A duel when the ball is in the air and loose.
+ GROUND (DuelType): A duel when the ball is on the ground.
+ LOOSE_BALL (DuelType): When the ball is not under the control of any particular player or team.
+ SLIDING_TACKLE (DuelType): A duel where the player slides on the ground to kick the ball away from an opponent.
+ STANDING_TACKLE (DuelType): A duel where the player makes a standing tackle.
+ """
+
+ AERIAL = "AERIAL"
+ GROUND = "GROUND"
+ LOOSE_BALL = "LOOSE_BALL"
+ SLIDING_TACKLE = "SLIDING_TACKLE"
+ STANDING_TACKLE = "STANDING_TACKLE"
+
+
+@dataclass
+class DuelQualifier(EnumQualifier):
+ value: DuelType
+
+
 @dataclass
 class CounterAttackQualifier(BoolQualifier):
  pass
@@ -423,6 +469,31 @@ def get_qualifier_value(self, qualifier_type: Type[Qualifier]):
  return qualifier.value
  return None
 
+
+ def get_qualifier_values(self, qualifier_type: Type[Qualifier]):
+ """
+ Returns all Qualifiers of a certain type, or None if qualifier is not present.
+
+ Arguments:
+ qualifier_type: one of the following QualifierTypes: [`SetPieceQualifier`][kloppy.domain.models.event.SetPieceQualifier]
+ [`BodyPartQualifier`][kloppy.domain.models.event.BodyPartQualifier] [`PassQualifier`][kloppy.domain.models.event.PassQualifier]
+
+ Examples:
+ >>> from kloppy.domain import SetPieceQualifier
+ >>> pass_event.get_qualifier_value(SetPieceQualifier)
+ <SetPieceType.GOAL_KICK: 'GOAL_KICK'>
+ """
+ qualifiers = []
+ if self.qualifiers:
+ for qualifier in self.qualifiers:
+ if isinstance(qualifier, qualifier_type):
+ qualifiers.append(qualifier)
+
+ if qualifiers:
+ return qualifiers
+
+ return None
+
  def get_related_events(self) -> List["Event"]:
  if not self.dataset:
  raise OrphanedRecordError()
@@ -650,6 +721,22 @@ class CarryEvent(Event):
  event_name: str = "carry"
 
 
+@dataclass(repr=False)
+@docstring_inherit_attributes(Event)
+class DuelEvent(Event):
+ """
+ DuelEvent
+
+ Attributes:
+ event_type (EventType): `EventType.DUEL` (See [`EventType`][kloppy.domain.models.event.EventType])
+ event_name (str): `"duel"`
+
+ """
+
+ event_type: EventType = EventType.DUEL
+ event_name: str = "duel"
+
+
 @dataclass(repr=False)
 @docstring_inherit_attributes(Event)
 class SubstitutionEvent(Event):
@@ -887,4 +974,8 @@ def generic_record_converter(event: Event):
  "GoalkeeperAction",
  "GoalkeeperActionQualifier",
  "CounterAttackQualifier",
+ "DuelEvent",
+ "DuelType",
+ "DuelQualifier",
+ "DuelResult",
 ]

kloppy/domain/services/event_factory.py

@@ -10,6 +10,7 @@
  TakeOnEvent,
  RecoveryEvent,
  CarryEvent,
+ DuelEvent,
  FormationChangeEvent,
  BallOutEvent,
  PlayerOnEvent,
@@ -82,6 +83,9 @@ def build_take_on(self, **kwargs) -> TakeOnEvent:
  def build_carry(self, **kwargs) -> CarryEvent:
  return create_event(CarryEvent, **kwargs)
 
+ def build_duel(self, **kwargs) -> DuelEvent:
+ return create_event(DuelEvent, **kwargs)
+
  def build_formation_change(self, **kwargs) -> FormationChangeEvent:
  return create_event(FormationChangeEvent, **kwargs)
 

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

@@ -19,6 +19,7 @@
  PassResult,
  ShotResult,
  TakeOnResult,
+ DuelResult,
  Ground,
  Score,
  Provider,
@@ -40,6 +41,8 @@
  BodyPart,
  PassType,
  PassQualifier,
+ DuelType,
+ DuelQualifier
 )
 from kloppy.exceptions import DeserializationError
 from kloppy.infra.serializers.event.deserializer import EventDataDeserializer
@@ -54,6 +57,9 @@
 EVENT_TYPE_PASS = 1
 EVENT_TYPE_OFFSIDE_PASS = 2
 EVENT_TYPE_TAKE_ON = 3
+EVENT_TYPE_TACKLE = 7
+EVENT_TYPE_AERIAL = 44
+EVENT_TYPE_50_50 = 67
 EVENT_TYPE_SHOT_MISS = 13
 EVENT_TYPE_SHOT_POST = 14
 EVENT_TYPE_SHOT_SAVED = 15
@@ -66,6 +72,7 @@
 EVENT_TYPE_FORMATION_CHANGE = 40
 
 BALL_OUT_EVENTS = [EVENT_TYPE_BALL_OUT, EVENT_TYPE_CORNER_AWARDED]
+DUEL_EVENTS = [EVENT_TYPE_TACKLE, EVENT_TYPE_AERIAL, EVENT_TYPE_50_50]
 
 BALL_OWNING_EVENTS = (
  EVENT_TYPE_PASS,
@@ -310,6 +317,22 @@ def _parse_shot(
  return dict(coordinates=coordinates, result=result, qualifiers=qualifiers)
 
 
+def _parse_duel(raw_qualifiers: List, type_id: int, outcome: int) -> Dict:
+ qualifiers = _get_event_qualifiers(raw_qualifiers)
+ duel_qualifiers = _get_duel_qualifiers(type_id)
+ qualifiers.extend(duel_qualifiers)
+
+ if outcome:
+ result = DuelResult.WON
+ else:
+ result = DuelResult.LOST
+
+ return dict(
+ result=result,
+ qualifiers=qualifiers,
+ )
+
+
 def _parse_team_players(
  f7_root, team_ref: str
 ) -> Tuple[str, Dict[str, Dict[str, str]]]:
@@ -452,6 +475,17 @@ def _get_event_card_qualifiers(raw_qualifiers: List) -> List[Qualifier]:
  return qualifiers
 
 
+def _get_duel_qualifiers(type_id: int) -> List[Qualifier]:
+ if type_id == EVENT_TYPE_TACKLE:
+ duel_qualifiers = [DuelQualifier(value=DuelType.GROUND), DuelQualifier(value=DuelType.STANDING_TACKLE)]
+ elif type_id == EVENT_TYPE_AERIAL:
+ duel_qualifiers = [DuelQualifier(value=DuelType.LOOSE_BALL), DuelQualifier(value=DuelType.AERIAL)]
+ elif type_id == EVENT_TYPE_50_50:
+ duel_qualifiers = [DuelQualifier(value=DuelType.LOOSE_BALL), DuelQualifier(value=DuelType.GROUND)]
+
+ return duel_qualifiers
+
+
 def _get_event_type_name(type_id: int) -> str:
  return event_type_names.get(type_id, "unknown")
 
@@ -607,7 +641,6 @@ def deserialize(self, inputs: OptaInputs) -> EventDataset:
  elif type_id == EVENT_TYPE_TAKE_ON:
  take_on_event_kwargs = _parse_take_on(outcome)
  event = self.event_factory.build_take_on(
- qualifiers=None,
  **take_on_event_kwargs,
  **generic_event_kwargs,
  )
@@ -643,7 +676,12 @@ def deserialize(self, inputs: OptaInputs) -> EventDataset:
  qualifiers=None,
  **generic_event_kwargs,
  )
-
+ elif type_id in DUEL_EVENTS:
+ duel_event_kwargs = _parse_duel(raw_qualifiers, type_id, outcome)
+ event = self.event_factory.build_duel(
+ **duel_event_kwargs,
+ **generic_event_kwargs,
+ )
  elif type_id == EVENT_TYPE_FOUL_COMMITTED:
  event = self.event_factory.build_foul_committed(
  result=None,

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

@@ -15,6 +15,9 @@
  ShotResult,
  TakeOnResult,
  CarryResult,
+ DuelResult,
+ DuelQualifier,
+ DuelType,
  Metadata,
  Ground,
  Player,
@@ -43,9 +46,11 @@
 logger = logging.getLogger(__name__)
 
 SB_EVENT_TYPE_RECOVERY = 2
+SB_EVENT_TYPE_DUEL = 4
 SB_EVENT_TYPE_DRIBBLE = 14
 SB_EVENT_TYPE_SHOT = 16
 SB_EVENT_TYPE_PASS = 30
+SB_EVENT_TYPE_50_50 = 33
 SB_EVENT_TYPE_CARRY = 43
 
 SB_EVENT_TYPE_HALF_START = 18
@@ -79,6 +84,28 @@
 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_50_50_OUTCOME_WON = 108
+# SB_50_50_OUTCOME_LOST = 109
+# SB_50_50_OUTCOME_SUCCESS_TO_OPPOSITION = 2 # TODO: Documentation says: 148 - Asking StatsBomb support
+# SB_50_50_OUTCOME_SUCCESS_TO_TEAM = 3 # TODO: Documentation says: 147 - Asking StatsBomb support
+#
+# SB_DUEL_OUTCOME_LOST = 1
+# SB_DUEL_OUTCOME_WON = 4
+# SB_DUEL_OUTCOME_LOST_IN_PLAY = 13
+# SB_DUEL_OUTCOME_LOST_OUT = 14
+# SB_DUEL_OUTCOME_SUCCESS = 15
+# SB_DUEL_OUTCOME_SUCCESS_IN_PLAY = 16
+# SB_DUEL_OUTCOME_SUCCESS_OUT = 17
+DUEL_WON_NAMES = ["Won", "Success To Team", "Success", "Success In Play", "Success Out"]
+DUEL_LOST_NAMES = ["Lost", "Aerial Lost", "Success To Opposition", "Lost In Play", "Lost Out"]
+# DUEL_WON_IDS = [SB_50_50_OUTCOME_WON, SB_50_50_OUTCOME_SUCCESS_TO_TEAM, SB_DUEL_OUTCOME_WON, SB_DUEL_OUTCOME_SUCCESS,
+# SB_DUEL_OUTCOME_SUCCESS_IN_PLAY, SB_DUEL_OUTCOME_SUCCESS_OUT]
+# DUEL_LOST_IDS = [SB_50_50_OUTCOME_LOST, SB_50_50_OUTCOME_SUCCESS_TO_OPPOSITION, SB_DUEL_OUTCOME_LOST,
+# SB_DUEL_OUTCOME_LOST_IN_PLAY, SB_DUEL_OUTCOME_LOST_OUT, SB_EVENT_TYPE_AERIAL_LOST]
+
 SB_EVENT_TYPE_FREE_KICK = 62
 SB_EVENT_TYPE_THROW_IN = 67
 SB_EVENT_TYPE_KICK_OFF = 65
@@ -416,6 +443,58 @@ def _parse_take_on(take_on_dict: Dict) -> Dict:
  }
 
 
+def _parse_duel(raw_event: dict, event_type: int, ) -> Dict:
+ qualifiers = []
+
+ if event_type == SB_EVENT_TYPE_DUEL:
+ duel_dict = raw_event["duel"]
+ if "type" in duel_dict:
+ type_id = duel_dict["type"]["id"]
+ if type_id == SB_EVENT_TYPE_AERIAL_LOST:
+ duel_qualifiers = [DuelQualifier(value=DuelType.LOOSE_BALL), DuelQualifier(value=DuelType.AERIAL)]
+ elif type_id == SB_EVENT_TYPE_TACKLE:
+ duel_qualifiers = [DuelQualifier(value=DuelType.GROUND), DuelQualifier(value=DuelType.STANDING_TACKLE)]
+ elif event_type == SB_EVENT_TYPE_50_50:
+ duel_dict = raw_event["50_50"]
+ duel_qualifiers = [DuelQualifier(value=DuelType.LOOSE_BALL), DuelQualifier(value=DuelType.GROUND)]
+
+ qualifiers.extend(duel_qualifiers)
+ body_part_qualifiers = _get_body_part_qualifiers(duel_dict)
+ qualifiers.extend(body_part_qualifiers)
+
+ if "outcome" in duel_dict:
+ outcome_name = duel_dict["outcome"]["name"]
+ else:
+ outcome_name = duel_dict["type"]["name"]
+
+ result = None
+ if outcome_name in DUEL_WON_NAMES:
+ result = DuelResult.WON
+ elif outcome_name in DUEL_LOST_NAMES:
+ result = DuelResult.LOST
+
+ return {
+ "result": result,
+ "qualifiers": qualifiers
+ }
+
+
+def _parse_aerial_won_duel(raw_event: dict, type_name: str) -> Dict:
+ qualifiers = []
+ aerial_won_dict = raw_event[type_name]
+ duel_qualifiers = [DuelQualifier(value=DuelType.LOOSE_BALL), DuelQualifier(value=DuelType.AERIAL)]
+ result = DuelResult.WON
+
+ qualifiers.extend(duel_qualifiers)
+ body_part_qualifiers = _get_body_part_qualifiers(aerial_won_dict)
+ qualifiers.extend(body_part_qualifiers)
+
+ return {
+ "result": result,
+ "qualifiers": qualifiers
+ }
+
+
 def _parse_substitution(substitution_dict: Dict, team: Team) -> Dict:
  replacement_player = None
  for player in team.players:
@@ -724,6 +803,15 @@ def deserialize(self, inputs: StatsBombInputs) -> EventDataset:
  **generic_event_kwargs,
  )
  new_events.append(carry_event)
+ elif event_type in [SB_EVENT_TYPE_DUEL, SB_EVENT_TYPE_50_50]:
+ duel_event_kwargs = _parse_duel(
+ raw_event=raw_event, event_type=event_type
+ )
+ duel_event = self.event_factory.build_duel(
+ **duel_event_kwargs,
+ **generic_event_kwargs,
+ )
+ new_events.append(duel_event)
 
  # lineup affecting events
  elif event_type == SB_EVENT_TYPE_SUBSTITUTION:
@@ -820,6 +908,19 @@ def deserialize(self, inputs: StatsBombInputs) -> EventDataset:
  )
  new_events.append(generic_event)
 
+ # Add possible aerial won - Last, since applicable to multiple event types
+ for type_name in ["shot", "clearance", "miscontrol", "pass"]:
+ if type_name in raw_event and "aerial_won" in raw_event[type_name]:
+ duel_event_kwargs = _parse_aerial_won_duel(
+ raw_event=raw_event, type_name=type_name
+ )
+ duel_event = self.event_factory.build_duel(
+ **duel_event_kwargs,
+ **generic_event_kwargs,
+ )
+ new_events.append(duel_event)
+
+
  for event in new_events:
  if self.should_include_event(event):
  transformed_event = transformer.transform_event(event)