Source code for action0.open_meteo.historical_weather.operations

# generated by action0-client-openapi v0.1.2 from historical-weather.yml — do not edit

from __future__ import annotations

import datetime
from typing import Any

from action0.client import JsonOperation
from action0.client import query
from action0.req import Method
from action0.req import Response

from .errors import BadRequestError
from .errors import decode_error
from .models import GetV1ArchiveCellSelection
from .models import GetV1ArchiveDailyItem
from .models import GetV1ArchiveHourlyItem
from .models import GetV1ArchiveModelsItem
from .models import GetV1ArchivePrecipitationUnit
from .models import GetV1ArchiveResponse
from .models import GetV1ArchiveTemperatureUnit
from .models import GetV1ArchiveTimeformat
from .models import GetV1ArchiveWindSpeedUnit
from .models import get_v1_archive_error_from_json
from .models import get_v1_archive_response_from_json




class GetV1Archive(JsonOperation[GetV1ArchiveResponse]):
    """
    ``GET /v1/archive`` — Open-Meteo Historical Weather API

    Historical weather data from ERA5 and ERA5-Land reanalysis going back to 1940.
    """

    method = Method.GET
    path = "/v1/archive"

    #: Geographical WGS84 coordinates. Multiple coordinates can be comma separated.
    latitude: str = query()
    longitude: str = query()
    #: A list of weather variables which should be returned.
    hourly: list[GetV1ArchiveHourlyItem] | None = query(
        default=None, serialize=lambda values: ",".join(str(item.value) for item in values)
    )
    #: A list of daily weather variable aggregations.
    daily: list[GetV1ArchiveDailyItem] | None = query(
        default=None, serialize=lambda values: ",".join(str(item.value) for item in values)
    )
    #: Start date in ISO 8601 format. Required.
    start_date: datetime.date = query()
    #: End date in ISO 8601 format. Required.
    end_date: datetime.date = query()
    elevation: float | None = query(default=None)
    temperature_unit: GetV1ArchiveTemperatureUnit = query(
        default=GetV1ArchiveTemperatureUnit.CELSIUS
    )
    wind_speed_unit: GetV1ArchiveWindSpeedUnit = query(default=GetV1ArchiveWindSpeedUnit.KMH)
    precipitation_unit: GetV1ArchivePrecipitationUnit = query(
        default=GetV1ArchivePrecipitationUnit.MM
    )
    #: If format unixtime is selected, all time values are returned in UNIX epoch time.
    timeformat: GetV1ArchiveTimeformat = query(default=GetV1ArchiveTimeformat.ISO8601)
    #: Any IANA time zone name is supported. Use auto to resolve the local time zone.
    timezone: str | None = query(default=None)
    #: Return past days of data.
    past_days: int = query(default=0)
    #: Number of forecast days.
    forecast_days: int = query(default=0)
    past_hours: int | None = query(default=None)
    forecast_hours: int | None = query(default=None)
    #: Slope tilt in degrees for global_tilted_irradiance calculation.
    tilt: float = query(default=0)
    #: Azimuth for global_tilted_irradiance. North=0, East=90, South=180, West=270.
    azimuth: float = query(default=0)
    #: Grid cell selection preference: land, sea, or nearest.
    cell_selection: GetV1ArchiveCellSelection | None = query(default=None)
    #: Only required for commercial subscriptions.
    apikey: str | None = query(default=None)
    #: Manually select one or more weather models.
    models: list[GetV1ArchiveModelsItem] | None = query(
        default=None, serialize=lambda values: ",".join(str(item.value) for item in values)
    )



    def check(self, response: Response) -> None:
        """
        :param response: the response the backend produced
        :raises BadRequestError: for the documented ``400`` answer
        :raises APIError: for any other non-2xx status
        """
        if response.status == 400:
            data = decode_error(response)
            if isinstance(data, dict):
                message = f"{type(self).__name__}: unexpected status {response.status}"
                raise BadRequestError(
                    f"{message} {response.phrase}".rstrip(),
                    response=response,
                    error=get_v1_archive_error_from_json(data),
                )
        super().check(response)




    def load_json(self, data: Any) -> GetV1ArchiveResponse:
        """
        :param data: the decoded JSON payload
        :return: the parsed result
        """
        return get_v1_archive_response_from_json(data)