Support StatsPerform MA5 and MA12 feeds

socceraction/data/opta/loader.py

@@ -21,6 +21,8 @@
     F24XMLParser,
     MA1JSONParser,
     MA3JSONParser,
+    MA5JSONParser,
+    MA12JSONParser,
     OptaParser,
     WhoScoredParser,
 )
@@ -38,6 +40,8 @@
     "f24": F24JSONParser,
     "ma1": MA1JSONParser,
     "ma3": MA3JSONParser,
+    "ma5": MA5JSONParser,
+    "ma12": MA12JSONParser,
 }
 
 _xmlparsers = {

socceraction/data/opta/parsers/__init__.py

@@ -9,6 +9,8 @@
     'F24XMLParser',
     'MA1JSONParser',
     'MA3JSONParser',
+    'MA5JSONParser',
+    'MA12JSONParser',
     'WhoScoredParser',
 ]
 
@@ -20,4 +22,6 @@
 from .f24_xml import F24XMLParser
 from .ma1_json import MA1JSONParser
 from .ma3_json import MA3JSONParser
+from .ma5_json import MA5JSONParser
+from .ma12_json import MA12JSONParser
 from .whoscored import WhoScoredParser

socceraction/data/opta/parsers/ma12_json.py

@@ -0,0 +1,168 @@
+"""JSON parser for Stats Perform MA1 feeds."""
+from typing import Any, Dict, List, Optional, Tuple
+
+import pandas as pd
+
+from ...base import MissingDataError
+from .base import OptaJSONParser, assertget
+
+
+class MA12JSONParser(OptaJSONParser):
+    """Extract data from a Stats Perform MA12 data stream.
+
+    Parameters
+    ----------
+    path : str
+        Path of the data file.
+    """
+
+    def _get_matches(self) -> List[Dict[str, Any]]:
+        if 'matchInfo' in self.root:
+            return [self.root]
+        if 'match' in self.root:
+            return self.root['match']
+        raise MissingDataError
+
+    def _get_match_info(self) -> Dict[str, Any]:
+        if "matchInfo" in self.root:
+            return self.root["matchInfo"]
+        raise MissingDataError
+
+    def _get_live_data(self) -> Dict[str, Any]:
+        if "liveData" in self.root:
+            return self.root["liveData"]
+        raise MissingDataError
+
+    def _get_name(self, obj: Dict[str, Any]) -> Optional[str]:
+        if "name" in obj:
+            return assertget(obj, "name")
+        if "firstName" in obj:
+            return f"{assertget(obj, 'firstName')} {assertget(obj, 'lastName')}"
+        return None
+
+    @staticmethod
+    def _extract_team_id(teams: List[Dict[str, str]], side: str) -> Optional[str]:
+        for team in teams:
+            team_side = assertget(team, "position")
+            if team_side == side:
+                team_id = assertget(team, "id")
+                return team_id
+        raise MissingDataError
+
+    def extract_players(self) -> Dict[Tuple[str, str], Dict[str, Any]]:  # noqa: C901
+        """Return a dictionary with all available players.
+
+        Returns
+        -------
+        dict
+            A mapping between (game ID, player ID) tuples and the information
+            available about each player in the data stream.
+        """
+        match_info = self._get_match_info()
+        game_id = assertget(match_info, "id")
+        live_data = self._get_live_data()
+        line_up = assertget(live_data, "lineUp")
+
+        players_data: Dict[str, List[Any]] = {
+            # "game_id": [],
+            "team_id": [],
+            "player_id": [],
+            "player_name": [],
+            # "is_starter": [],
+            "starting_position": [],
+            "minutes_played": [],
+            "jersey_number": [],
+            "position_side": [],
+            "xG_non_penalty": [],
+        }
+
+        team_info = assertget(match_info, "contestant")
+        home_team_info = team_info[0]
+        away_team_info = team_info[1]
+
+        home_team_id = assertget(home_team_info, "id")
+        away_team_id = assertget(away_team_info, "id")
+
+        team_home = line_up[0]
+        team_away = line_up[1]
+        for player in assertget(team_home, "player"):
+            # players_data["game_id"] += [game_id]
+            players_data["team_id"] += [home_team_id]
+            players_data["player_id"] += [assertget(player, "playerId")]
+            players_data["player_name"] += [assertget(player, "matchName")]
+
+            starting_position = assertget(player, "position")
+            players_data["starting_position"] += [starting_position]
+            # players_data["is_starter"] += (starting_position != "Substitute")
+
+            players_data["jersey_number"] += [assertget(player, "shirtNumber")]
+
+            if starting_position != "Substitute":
+                players_data["position_side"] += [assertget(player, "positionSide")]
+            else:
+                players_data["position_side"] += [""]
+
+            xG_non_penalty = 0.0
+            minutesPlayed = 0
+            if "stat" in player and player["stat"] is not None:
+                player_stats = assertget(player, "stat")
+                for stat in player_stats:
+                    stat_type = assertget(stat, "type")
+                    if stat_type == "expectedGoalsNonpenalty":
+                        xG_non_penalty = assertget(stat, "value")
+                        break
+                    elif stat_type == "minsPlayed":
+                        minutesPlayed = assertget(stat, "value")
+            players_data["xG_non_penalty"] += [xG_non_penalty]
+            players_data["minutes_played"] += [minutesPlayed]
+
+        for player in assertget(team_away, "player"):
+            # players_data["game_id"] += game_id
+            players_data["team_id"] += [away_team_id]
+            players_data["player_id"] += [assertget(player, "playerId")]
+            players_data["player_name"] += [assertget(player, "matchName")]
+
+            starting_position = assertget(player, "position")
+            players_data["starting_position"] += [starting_position]
+            # players_data["is_starter"] += (starting_position != "Substitute")
+
+            players_data["jersey_number"] += [assertget(player, "shirtNumber")]
+
+            if starting_position != "Substitute":
+                players_data["position_side"] += [assertget(player, "positionSide")]
+            else:
+                players_data["position_side"] += [""]
+            xG_non_penalty = 0.0
+            minutesPlayed = 0
+            if "stat" in player:
+                player_stats = assertget(player, "stat")
+                for stat in player_stats:
+                    stat_type = assertget(stat, "type")
+                    if stat_type == "expectedGoalsNonpenalty":
+                        xG_non_penalty = assertget(stat, "value")
+                        break
+                    elif stat_type == "minsPlayed":
+                        minutesPlayed = assertget(stat, "value")
+            players_data["xG_non_penalty"] += [xG_non_penalty]
+            players_data["minutes_played"] += [minutesPlayed]
+
+        df_players_data = pd.DataFrame.from_dict(players_data)
+
+        players = {}
+        for _, player in df_players_data.iterrows():
+            is_starter = player.starting_position != "Substitute"
+            players[(game_id, player.player_id)] = {
+                # Fields required by the base schema
+                "game_id": game_id,
+                "team_id": player.team_id,
+                "player_id": player.player_id,
+                "player_name": player.player_name,
+                "is_starter": is_starter,
+                "minutes_played": int(player.minutes_played),
+                "jersey_number": player.jersey_number,
+                # Fields required by the opta schema
+                "starting_position": player.starting_position,
+                "position_side": player.position_side,
+                "xG_non_penalty": player.xG_non_penalty,
+            }
+        return players

socceraction/data/opta/parsers/ma5_json.py

@@ -0,0 +1,77 @@
+"""JSON parser for Stats Perform MA5 feeds."""
+from datetime import datetime
+from typing import Any, Dict, List
+
+from ...base import MissingDataError
+from .base import OptaJSONParser, assertget
+
+
+class MA5JSONParser(OptaJSONParser):
+    """Extract data from a Stats Perform MA5 data stream.
+
+    Parameters
+    ----------
+    path : str
+        Path of the data file.
+    """
+
+    def _get_matches(self) -> List[Dict[str, Any]]:
+        if 'matchInfo' in self.root:
+            return [self.root]
+        if 'match' in self.root:
+            return self.root['match']
+        raise MissingDataError
+
+    def _get_match_info(self) -> Dict[str, Any]:
+        if "matchInfo" in self.root:
+            return self.root["matchInfo"]
+        raise MissingDataError
+
+    def _get_live_data(self) -> Dict[str, Any]:
+        if "liveData" in self.root:
+            return self.root["liveData"]
+        raise MissingDataError
+
+    def extract_games(self) -> Dict[str, Dict[str, Any]]:
+        """Return a dictionary with all available games.
+
+        Returns
+        -------
+        dict
+            A mapping between game IDs and the information available about
+            each game in the data stream.
+        """
+        match_info = self._get_match_info()
+        live_data = self._get_live_data()
+        game_id = assertget(match_info, "id")
+        season = assertget(match_info, "tournamentCalendar")
+        competition = assertget(match_info, "competition")
+
+        contestants = assertget(match_info, "contestant")
+        home_team = contestants[0]
+        away_team = contestants[1]
+        home_team_id = assertget(home_team, "id")
+        away_team_id = assertget(away_team, "id")
+
+        possession_data = assertget(live_data, "possession")
+        possession_wave_data = assertget(possession_data, "possessionWave")
+        overall_percentage = assertget(possession_wave_data[0], "overall")
+        away_possession = assertget(overall_percentage, "away")
+        home_possession = assertget(overall_percentage, "home")
+
+        game_date = assertget(match_info, "date")[0:10]
+        game_time = assertget(match_info, "time")[0:8]
+        game_datetime = f"{game_date}T{game_time}"
+        return {
+            game_id: dict(
+                game_id=game_id,
+                season_id=assertget(season, "id"),
+                competition_id=assertget(competition, "id"),
+                game_day=int(match_info["week"]) if "week" in match_info else None,
+                game_date=datetime.strptime(game_datetime, "%Y-%m-%dT%H:%M:%S"),
+                home_team_id=home_team_id,
+                away_team_id=away_team_id,
+                home_possession=float(home_possession),
+                away_possession=float(away_possession),
+            )
+        }

socceraction/data/opta/schema.py

@@ -37,13 +37,24 @@ class OptaGameSchema(GameSchema):
     """The name of the manager of the home team."""
     away_manager: Optional[Series[str]] = pa.Field(nullable=True)
     """The name of the manager of the away team."""
+    home_possession: Optional[Series[int]]
+    """The possession percentage of the home team in the game."""
+    away_possession: Optional[Series[int]]
+    """The possession percentage of the away team in the game."""
 
 
 class OptaPlayerSchema(PlayerSchema):
     """Definition of a dataframe containing the list of players of a game."""
 
     starting_position: Series[str]
     """The starting position of the player."""
+    position_side: Optional[Series[str]]
+    """The side of the pitch where the player started the game"""
+    xG_non_penalty: Optional[Series[float]]
+    """The xG of the player without taking penalties into account"""
+
+    # minutes_played: Optional[Series[int]]
+    # """The amount of minutes played by the player"""
 
 
 class OptaTeamSchema(TeamSchema):

tests/data/opta/parsers/test_ma12_json.py

@@ -0,0 +1,46 @@
+import os
+
+import pandas as pd
+from pytest import fixture
+
+from socceraction.data.opta import OptaPlayerSchema
+from socceraction.data.opta.parsers import MA12JSONParser
+
+
+@fixture()
+def ma12json_parser() -> MA12JSONParser:
+    path = os.path.join(
+        os.path.dirname(__file__),
+        os.pardir,
+        os.pardir,
+        os.pardir,
+        "datasets",
+        "opta",
+        "ma12_drnl1etmyzjz088y0bb4kh4pg.json",
+    )
+    return MA12JSONParser(str(path))
+
+
+def test_extract_players(ma12json_parser: MA12JSONParser) -> None:
+    players = ma12json_parser.extract_players()
+    assert len(players) == 40
+    assert players[("drnl1etmyzjz088y0bb4kh4pg", "4iijb6llnz28unsz4rirr3umt")] == {
+        "game_id": "drnl1etmyzjz088y0bb4kh4pg",
+        "player_id": "4iijb6llnz28unsz4rirr3umt",
+        "player_name": "David Raya",
+        "team_id": "7yx5dqhhphyvfisohikodajhv",
+        "jersey_number": 1,
+        "minutes_played": 90,
+        "starting_position": "Goalkeeper",
+        "is_starter": True,
+        "position_side": "Centre",
+        "xG_non_penalty": 0.0,
+    }
+    # red card
+    assert (
+        players[("drnl1etmyzjz088y0bb4kh4pg", "ds1wjqejslhsbcvzaufxubey2")]["minutes_played"] == 10
+    )
+    assert (
+        players[("drnl1etmyzjz088y0bb4kh4pg", "2o5etqath4k9qaliho4op0hd5")]["minutes_played"] == 0
+    )
+    OptaPlayerSchema.validate(pd.DataFrame.from_dict(players, orient="index"))

tests/data/opta/parsers/test_ma5_json.py

@@ -0,0 +1,39 @@
+import os
+from datetime import datetime
+
+import pandas as pd
+from pytest import fixture
+
+from socceraction.data.opta import OptaGameSchema
+from socceraction.data.opta.parsers import MA5JSONParser
+
+
+@fixture()
+def ma5json_parser() -> MA5JSONParser:
+    path = os.path.join(
+        os.path.dirname(__file__),
+        os.pardir,
+        os.pardir,
+        os.pardir,
+        "datasets",
+        "opta",
+        "ma5_drnl1etmyzjz088y0bb4kh4pg.json",
+    )
+    return MA5JSONParser(str(path))
+
+
+def test_extract_games(ma5json_parser: MA5JSONParser) -> None:
+    games = ma5json_parser.extract_games()
+    assert len(games) == 1
+    assert games["drnl1etmyzjz088y0bb4kh4pg"] == {
+        "game_id": "drnl1etmyzjz088y0bb4kh4pg",
+        "season_id": "8l3o9v8n8tva0bb2cds2dhatw",
+        "competition_id": "2kwbbcootiqqgmrzs6o5inle5",
+        "game_day": 1,
+        "game_date": datetime(2021, 8, 13, 19, 0),
+        "home_team_id": "7yx5dqhhphyvfisohikodajhv",
+        "away_team_id": "4dsgumo7d4zupm2ugsvm4zm4d",
+        "away_possession": 64.7,
+        "home_possession": 35.3,
+    }
+    OptaGameSchema.validate(pd.DataFrame.from_dict(games, orient="index"))