Source code for action0.open_meteo.ensemble.operations

# generated by action0-client-openapi v0.1.2 from ensemble.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 GetV1EnsembleCellSelection
from .models import GetV1EnsembleDailyItem
from .models import GetV1EnsembleHourlyItem
from .models import GetV1EnsembleModelsItem
from .models import GetV1EnsemblePrecipitationUnit
from .models import GetV1EnsembleResponse
from .models import GetV1EnsembleTemperatureUnit
from .models import GetV1EnsembleTemporalResolution
from .models import GetV1EnsembleTimeformat
from .models import GetV1EnsembleWindSpeedUnit
from .models import get_v1_ensemble_error_from_json
from .models import get_v1_ensemble_response_from_json




class GetV1Ensemble(JsonOperation[GetV1EnsembleResponse]):
    """
    ``GET /v1/ensemble`` — Open-Meteo Ensemble Weather Forecast API

    Ensemble weather forecasts from multiple models providing probabilistic forecast data.
    """

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

    #: 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[GetV1EnsembleHourlyItem] | None = query(
        default=None, serialize=lambda values: ",".join(str(item.value) for item in values)
    )
    #: A list of daily weather variable aggregations.
    daily: list[GetV1EnsembleDailyItem] | None = query(
        default=None, serialize=lambda values: ",".join(str(item.value) for item in values)
    )
    elevation: float | None = query(default=None)
    temperature_unit: GetV1EnsembleTemperatureUnit = query(
        default=GetV1EnsembleTemperatureUnit.CELSIUS
    )
    wind_speed_unit: GetV1EnsembleWindSpeedUnit = query(default=GetV1EnsembleWindSpeedUnit.KMH)
    precipitation_unit: GetV1EnsemblePrecipitationUnit = query(
        default=GetV1EnsemblePrecipitationUnit.MM
    )
    #: If format unixtime is selected, all time values are returned in UNIX epoch time.
    timeformat: GetV1EnsembleTimeformat = query(default=GetV1EnsembleTimeformat.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=7)
    past_hours: int | None = query(default=None)
    forecast_hours: int | None = query(default=None)
    #: Start date of the time interval in ISO 8601 format.
    start_date: datetime.date | None = query(default=None)
    #: End date of the time interval in ISO 8601 format.
    end_date: datetime.date | 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: GetV1EnsembleCellSelection | None = query(default=None)
    #: Only required for commercial subscriptions.
    apikey: str | None = query(default=None)
    #: Manually select one or more weather models.
    models: list[GetV1EnsembleModelsItem] | None = query(
        default=None, serialize=lambda values: ",".join(str(item.value) for item in values)
    )
    #: Aggregate data into larger time intervals. Supported: native, hourly, hourly_3, hourly_6.
    temporal_resolution: GetV1EnsembleTemporalResolution | None = query(default=None)



    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_ensemble_error_from_json(data),
                )
        super().check(response)




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