avwx.forecast.base

  1"""Forecast report shared resources."""
  2
  3# stdlib
  4from __future__ import annotations
  5
  6from datetime import datetime, timedelta, timezone
  7from typing import TYPE_CHECKING
  8
  9# module
 10from avwx.base import ManagedReport
 11from avwx.parsing import core
 12from avwx.structs import Code, Number, ReportData, Timestamp
 13
 14if TYPE_CHECKING:
 15    from collections.abc import Callable
 16
 17    from avwx.service import Service
 18
 19
 20def _trim_lines(lines: list[str], target: int) -> list[str]:
 21    """Trim all lines to match the trimmed length of the target line."""
 22    length = len(lines[target].strip())
 23    return [line[:length] for line in lines]
 24
 25
 26def _split_line(line: str, size: int = 3, prefix: int = 4, strip: str = " |") -> list[str]:
 27    """Evenly split a string while stripping elements."""
 28    line = line[prefix:]
 29    ret = []
 30    while len(line) >= size:
 31        ret.append(line[:size].strip(strip))
 32        line = line[size:]
 33    if line := line.strip(strip):
 34        ret.append(line)
 35    return ret
 36
 37
 38def _timestamp(line: str) -> Timestamp:
 39    """Return the report timestamp from the first line."""
 40    start = line.find("GUIDANCE") + 11
 41    text = line[start : start + 16].strip()
 42    timestamp = datetime.strptime(text, r"%m/%d/%Y  %H%M").replace(tzinfo=timezone.utc)
 43    return Timestamp(text, timestamp.replace(tzinfo=timezone.utc))
 44
 45
 46def _find_time_periods(line: list[str], timestamp: datetime | None) -> list[dict]:
 47    """Find and create the empty time periods."""
 48    periods: list[Timestamp | None] = []
 49    if timestamp is None:
 50        periods = [None] * len(line)
 51    else:
 52        previous = timestamp.hour
 53        for hourstr in line:
 54            if not hourstr:
 55                continue
 56            hour = int(hourstr)
 57            previous, difference = hour, hour - previous
 58            if difference < 0:
 59                difference += 24
 60            timestamp += timedelta(hours=difference)
 61            periods.append(Timestamp(hourstr, timestamp))
 62    return [{"time": time} for time in periods]
 63
 64
 65def _init_parse(report: str) -> tuple[ReportData, list[str]]:
 66    """Return the meta data and lines from a report string."""
 67    report = report.strip()
 68    lines = report.split("\n")
 69    struct = ReportData(
 70        raw=report,
 71        sanitized=report,
 72        station=report[:4],
 73        time=_timestamp(lines[0]),
 74        remarks=None,
 75    )
 76    return struct, lines
 77
 78
 79def _numbers(
 80    line: str,
 81    size: int = 3,
 82    prefix: str = "",
 83    postfix: str = "",
 84    *,
 85    decimal: int | None = None,
 86    literal: bool = False,
 87    special: dict | None = None,
 88) -> list[Number | None]:
 89    """Parse line into Number objects.
 90
 91    Prefix, postfix, and decimal location are applied to value, not repr.
 92
 93    Decimal is applied after prefix and postfix.
 94    """
 95    ret = []
 96    for item in _split_line(line, size=size):
 97        value = None
 98        if item:
 99            value = prefix + item + postfix
100            if decimal is not None:
101                if abs(decimal) > len(value):
102                    value = value.zfill(abs(decimal))
103                value = f"{value[:decimal]}.{value[decimal:]}"
104        ret.append(core.make_number(value, repr=item, literal=literal, special=special))
105    return ret
106
107
108def _decimal_10(line: str, size: int = 3) -> list[Number | None]:
109    """Parse line into Number objects with 10ths decimal location."""
110    return _numbers(line, size, decimal=-1)
111
112
113def _decimal_100(line: str, size: int = 3) -> list[Number | None]:
114    """Parse line into Number objects with 100ths decimal location."""
115    return _numbers(line, size, decimal=-2)
116
117
118def _number_10(line: str, size: int = 3) -> list[Number | None]:
119    """Parse line into Number objects in tens."""
120    return _numbers(line, size, postfix="0")
121
122
123def _number_100(line: str, size: int = 3) -> list[Number | None]:
124    """Parse line into Number objects in hundreds."""
125    return _numbers(line, size, postfix="00")
126
127
128def _direction(line: str, size: int = 3) -> list[Number | None]:
129    """Parse line into Number objects in hundreds."""
130    return _numbers(line, size, postfix="0", literal=True)
131
132
133def _code(mapping: dict) -> Callable:
134    """Generate a conditional code mapping function."""
135
136    def func(line: str, size: int = 3) -> list[Code | str | None]:
137        ret: list[Code | str | None] = []
138        for key in _split_line(line, size=size):
139            try:
140                ret.append(Code(key, mapping[key]))
141            except KeyError:
142                ret.append(key or None)
143        return ret
144
145    return func
146
147
148def _parse_lines(
149    periods: list[dict],
150    lines: list[str],
151    handlers: dict | Callable,
152    size: int = 3,
153) -> None:
154    """Add data to time periods by parsing each line (element type).
155
156    Adds data in place.
157    """
158    for line in lines:
159        try:
160            key = line[:3]
161            *keys, handler = handlers[key] if isinstance(handlers, dict) else handlers(key)
162        except (IndexError, KeyError):
163            continue
164        values = handler(line, size=size)
165        values += [None] * (len(periods) - len(values))
166        for i in range(len(periods)):
167            value = values[i]
168            if not value:
169                continue
170            if isinstance(value, tuple):
171                for j, key in enumerate(keys):
172                    if value[j]:
173                        periods[i][key] = value[j]
174            else:
175                periods[i][keys[0]] = value
176
177
178class Forecast(ManagedReport):
179    """Forecast base class."""
180
181    report_type: str
182    _service_class: Service
183
184    def __init__(self, code: str):
185        super().__init__(code)
186        self.service: Service = self._service_class(self.report_type)  # type: ignore