File size: 4,855 Bytes
d6ea71e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
"""Base class for all Opta(-derived) event stream parsers.

A parser reads a single data file and should extend the 'OptaParser' class to
extract data about competitions, games, players, teams and events that is
encoded in the file.

"""

import json  # type: ignore
from typing import Any, Optional

from lxml import objectify


class OptaParser:
    """Extract data from an Opta data stream.

    Parameters
    ----------
    path : str
        Path of the data file.
    """

    def __init__(self, path: str, **kwargs: Any) -> None:  # noqa: ANN401
        raise NotImplementedError

    def extract_competitions(self) -> dict[tuple[Any, Any], dict[str, Any]]:
        """Return a dictionary with all available competitions.

        Returns
        -------
        dict
            A mapping between (competion ID, season ID) tuples and the
            information available about each competition in the data stream.
        """
        return {}

    def extract_games(self) -> dict[Any, 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.
        """
        return {}

    def extract_teams(self) -> dict[Any, dict[str, Any]]:
        """Return a dictionary with all available teams.

        Returns
        -------
        dict
            A mapping between team IDs and the information available about
            each team in the data stream.
        """
        return {}

    def extract_players(self) -> dict[tuple[Any, Any], dict[str, Any]]:
        """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.
        """
        return {}

    def extract_lineups(self) -> dict[Any, dict[str, Any]]:
        """Return a dictionary with the lineup of each team.

        Returns
        -------
        dict
            A mapping between team IDs and the information available about
            each team's lineup in the data stream.
        """
        return {}

    def extract_events(self) -> dict[tuple[Any, Any], dict[str, Any]]:
        """Return a dictionary with all available events.

        Returns
        -------
        dict
            A mapping between (game ID, event ID) tuples and the information
            available about each event in the data stream.
        """
        return {}


class OptaJSONParser(OptaParser):
    """Extract data from an Opta JSON data stream.

    Parameters
    ----------
    path : str
        Path of the data file.
    """

    def __init__(self, path: str, **kwargs: Any) -> None:  # noqa: ANN401
        with open(path, encoding="utf-8") as fh:
            self.root = json.load(fh)


class OptaXMLParser(OptaParser):
    """Extract data from an Opta XML data stream.

    Parameters
    ----------
    path : str
        Path of the data file.
    """

    def __init__(self, path: str, **kwargs: Any) -> None:  # noqa: ANN401
        with open(path, "rb") as fh:
            self.root = objectify.fromstring(fh.read())


def assertget(dictionary: dict[str, Any], key: str) -> Any:  # noqa: ANN401
    """Return the value of the item with the specified key.

    In contrast to the default `get` method, this version will raise an
    assertion error if the given key is not present in the dict.

    Parameters
    ----------
    dictionary : dict
        A Python dictionary.
    key : str
        A key in the dictionary.

    Returns
    -------
    Any
        Returns the value for the specified key if the key is in the dictionary.

    Raises
    ------
    AssertionError
        If the given key could not be found in the dictionary.
    """
    value = dictionary.get(key)
    assert value is not None, "KeyError: " + key + " not found in " + str(dictionary)
    return value


def _get_end_x(qualifiers: dict[int, Any]) -> Optional[float]:
    try:
        # pass
        if 140 in qualifiers:
            return float(qualifiers[140])
        # blocked shot
        if 146 in qualifiers:
            return float(qualifiers[146])
        # passed the goal line
        if 102 in qualifiers:
            return float(100)
        return None
    except ValueError:
        return None


def _get_end_y(qualifiers: dict[int, Any]) -> Optional[float]:
    try:
        # pass
        if 141 in qualifiers:
            return float(qualifiers[141])
        # blocked shot
        if 147 in qualifiers:
            return float(qualifiers[147])
        # passed the goal line
        if 102 in qualifiers:
            return float(qualifiers[102])
        return None
    except ValueError:
        return None