pyppms.common

  1"""Common functions related to Stratocore's PPMS Utility Management API."""
  2
  3# pylint: disable-msg=fixme
  4
  5from datetime import datetime, timedelta
  6import csv
  7from io import StringIO
  8
  9from loguru import logger as log
 10
 11from .exceptions import NoDataError
 12
 13
 14def process_response_values(values):
 15    """Process (in-place) a list of strings, remove quotes, detect boolean etc.
 16
 17    Check all (str) elements of the given list, remove surrounding double-quotes
 18    and convert 'true' / 'false' strings into Python booleans.
 19
 20    Parameters
 21    ----------
 22    values : list(str)
 23        The list of strings that should be processed.
 24
 25    Returns
 26    -------
 27    None
 28        Nothing is returned, the list's element are processed in-place.
 29    """
 30    # tell pylint that there is no real gain using enumerate here:
 31    # pylint: disable-msg=consider-using-enumerate
 32    for i in range(len(values)):
 33        values[i] = values[i].strip('"')
 34        if values[i] == "true":
 35            values[i] = True
 36        if values[i] == "false":
 37            values[i] = False
 38
 39
 40def dict_from_single_response(text, graceful=True):
 41    """Parse a two-line CSV response from PUMAPI and create a dict from it.
 42
 43    Parameters
 44    ----------
 45    text : str
 46        The PUMAPI response with two lines: a header line and one data line.
 47    graceful : bool, optional
 48        Whether to continue in case the response text is inconsistent, i.e.
 49        having different number of fields in the header line and the data line,
 50        by default True. In graceful mode, any inconsistency detected in the
 51        data will be logged as a warning, in non-graceful mode they will raise
 52        an Exception.
 53
 54    Returns
 55    -------
 56    dict
 57        A dict with the fields of the header line being the keys and the fields
 58        of the data line being the values. Values are stripped from quotes
 59        and converted to Python boolean values where applicable.
 60
 61    Raises
 62    ------
 63    ValueError
 64        Raised when the response text is inconsistent and the `graceful`
 65        parameter has been set to false, or if parsing fails for any other
 66        unforeseen reason.
 67    """
 68
 69    # check if we got an empty response (only two newlines) and return a dict
 70    # with two empty strings only
 71    # TODO: this should probably rather raise a ValueError but we need to test
 72    # all effects on existing code first!
 73    if text == "\n\n":
 74        return {"": ""}
 75    try:
 76        lines = list(csv.reader(StringIO(text), delimiter=","))
 77        if len(lines) != 2:
 78            log.warning("Response expected to have exactly two lines: {}", text)
 79            if not graceful:
 80                raise ValueError("Invalid response format!")
 81        header = lines[0]
 82        data = lines[1]
 83        process_response_values(data)
 84        if len(header) != len(data):
 85            msg = "Parsing CSV failed, mismatch of header vs. data fields count"
 86            log.warning("{} ({} vs. {})", msg, len(header), len(data))
 87            if not graceful:
 88                raise ValueError(msg)
 89            minimum = min(len(header), len(data))
 90            if minimum < len(header):
 91                log.warning("Discarding header-fields: {}", header[minimum:])
 92                header = header[:minimum]
 93            else:
 94                log.warning("Discarding data-fields: {}", data[minimum:])
 95                data = data[:minimum]
 96
 97    except Exception as err:
 98        msg = f"Unable to parse data returned by PUMAPI: {text} - ERROR: {err}"
 99        log.error(msg)
100        raise ValueError(msg) from err
101
102    parsed = dict(zip(header, data))
103    return parsed
104
105
106def parse_multiline_response(text, graceful=True):
107    """Parse a multi-line CSV response from PUMAPI.
108
109    Parameters
110    ----------
111    text : str
112        The PUMAPI response with two or more lines, where the first line
113        contains the header field names and the subsequent lines contain data.
114    graceful : bool, optional
115        Whether to continue in case the response text is inconsistent, i.e.
116        having different number of fields in the header line and the data lines,
117        by default True. In graceful mode, any inconsistency detected in the
118        data will be logged as a warning, in non-graceful mode they will raise
119        an Exception.
120
121    Returns
122    -------
123    list(dict)
124        A list with dicts where the latter ones have the same form as produced
125        by the dict_from_single_response() function. May be empty in case the
126        PUMAPI response didn't contain any useful data. Note that when graceful
127        mode is requested, consistency among the dicts is not guaranteed.
128
129    Raises
130    ------
131    NoDataError
132        Raised when the response text was too short (less than two lines) and
133        the `graceful` parameter has been set to false.
134    ValueError
135        Raised when the response text is inconsistent and the `graceful`
136        parameter has been set to false, or if parsing fails for any other
137        unforeseen reason.
138    """
139    parsed = []
140    try:
141        lines = text.splitlines()
142        if len(lines) < 2:
143            log.debug("Response has less than TWO lines: >>>{}<<<", text)
144            if not graceful:
145                raise NoDataError("Invalid response format!")
146            return []
147
148        header = lines[0].split(",")
149        for i, entry in enumerate(header):
150            header[i] = entry.strip()
151
152        lines_max = lines_min = len(header)
153        for line in lines[1:]:
154            data = line.split(",")
155            process_response_values(data)
156            lines_max = max(lines_max, len(data))
157            lines_min = min(lines_min, len(data))
158            if len(header) != len(data):
159                msg = "Parsing CSV failed, mismatch of header vs. data fields count"
160                log.warning("{} ({} vs. {})", msg, len(header), len(data))
161                if not graceful:
162                    raise ValueError(msg)
163
164                minimum = min(len(header), len(data))
165                if minimum < len(header):
166                    log.warning("Discarding header-fields: {}", header[minimum:])
167                    header = header[:minimum]
168                else:
169                    log.warning("Discarding data-fields: {}", data[minimum:])
170                    data = data[:minimum]
171
172            details = dict(zip(header, data))
173            # log.debug(details)
174            parsed.append(details)
175
176        if lines_min != lines_max:
177            msg = (
178                "Inconsistent data detected, not all dicts will have the "
179                "same number of elements!"
180            )
181            log.warning(msg)
182
183    except NoDataError as err:
184        raise err
185
186    except Exception as err:
187        msg = f"Unable to parse data returned by PUMAPI: {text} - ERROR: {err}"
188        log.error(msg)
189        raise ValueError(msg) from err
190
191    return parsed
192
193
194def time_rel_to_abs(minutes_from_now):
195    """Convert a relative time given in minutes from now to a datetime object.
196
197    Parameters
198    ----------
199    minutes_from_now : int or int-like
200        The relative time in minutes to be converted.
201
202    Returns
203    -------
204    datetime
205        The absolute time point as a datetime object.
206    """
207    now = datetime.now().replace(second=0, microsecond=0)
208    abstime = now + timedelta(minutes=int(minutes_from_now))
209    return abstime
210
211
212def fmt_time(time):
213    """Format a `datetime` or `None` object to string.
214
215    This is useful to apply it to booking times as they might be `None` e.g. in
216    case they have been created from a "nextbooking" response.
217
218    Parameters
219    ----------
220    time : datetime.datetime or None
221
222    Returns
223    -------
224    str
225        The formatted time, or a specific string in case the input was `None`.
226    """
227    if time is None:
228        return "===UNDEFINED==="
229    return datetime.strftime(time, "%Y-%m-%d %H:%M")