openaq_v2

OpenAQ V2 REST API Reader

OpenAQV2Reader

Bases: PointReader

Reader for OpenAQ V2 REST API data.

Source code in monetio/readers/openaq_v2.py

@register_reader("openaq_v2")
class OpenAQV2Reader(PointReader):
    """
    Reader for OpenAQ V2 REST API data.
    """

    @_api_key_warning
    def open_dataset(
        self,
        dates: pd.DatetimeIndex | list[datetime] | datetime | str = None,
        parameters: list[str] = None,
        country: str | list[str] = None,
        sites: list[str] = None,
        wide_fmt: bool = True,
        as_xarray: bool = True,
        lazy: bool = False,
        **kwargs: Any,
    ) -> Union[pd.DataFrame, xr.Dataset, "dd.DataFrame"]:
        """
        Retrieves OpenAQ data via the REST API.

        Parameters
        ----------
        dates : Union[pd.DatetimeIndex, List[datetime], datetime, str]
            Dates to retrieve.
        parameters : List[str], optional
            Species to retrieve, by default ['pm25', 'o3'].
        country : Union[str, List[str]], optional
            Country code(s).
        sites : List[str], optional
            Site ID(s).
        wide_fmt : bool, optional
            Whether to return data in wide format, by default True.
        as_xarray : bool, optional
            Whether to return an xarray.Dataset, by default True.
        lazy : bool, optional
            Whether to return a dask-backed object, by default False.
        **kwargs : dict
            Additional arguments passed to the API.

        Returns
        -------
        Union[pd.DataFrame, xr.Dataset, dd.DataFrame]
            The loaded data.

        Examples
        --------
        >>> from monetio.readers.openaq_v2 import OpenAQV2Reader
        >>> reader = OpenAQV2Reader()
        >>> df = reader.open_dataset(dates='2023-01-01', as_xarray=False)
        """
        if dates is None:
            raise ValueError("must provide at least one datetime-like via 'dates'")

        dates = pd.to_datetime(dates)
        if pd.api.types.is_scalar(dates):
            dates = pd.DatetimeIndex([dates])
        dates = dates.dropna()
        if dates.empty:
            raise ValueError("must provide at least one datetime-like")

        if parameters is None:
            parameters = ["pm25", "o3"]
        elif isinstance(parameters, str):
            parameters = [parameters]

        if lazy:
            import dask.dataframe as dd
            from dask import delayed

            query_time_split = kwargs.get("query_time_split", "1D")
            query_dt = pd.to_timedelta(query_time_split)
            date_min, date_max = dates.min(), dates.max()

            def iter_time_slices():
                one_sec = pd.Timedelta(seconds=1)
                if date_min < date_max:
                    t = date_min
                    while t < date_max:
                        t_next = min(t + query_dt, date_max)
                        yield t - one_sec, t_next
                        t = t_next
                else:
                    yield date_min - one_sec, date_max

            delayed_dfs = []
            for t_from, t_to in iter_time_slices():
                for p in parameters:
                    part_kwargs = kwargs.copy()
                    part_kwargs.update(
                        dates=pd.DatetimeIndex([t_from + pd.Timedelta(seconds=1), t_to]),
                        parameters=[p],
                        country=country,
                        sites=sites,
                        query_time_split=None,
                    )
                    delayed_dfs.append(delayed(self._fetch_data)(**part_kwargs))

            if not delayed_dfs:
                df = dd.from_pandas(pd.DataFrame(), npartitions=1)
            else:
                meta = self._get_meta()
                df = dd.from_delayed(delayed_dfs, meta=meta)
        else:
            df = self._fetch_data(
                dates=dates, parameters=parameters, country=country, sites=sites, **kwargs
            )

        df = self.harmonize(df)

        df = force_object_strings(df)

        if as_xarray:
            return self.to_xarray(df, wide_fmt=wide_fmt, **kwargs)

        if wide_fmt:
            from ..util import long_to_wide

            df = long_to_wide(df)

        return df

    def _get_meta(self) -> pd.DataFrame:
        """Returns an empty DataFrame with the expected columns for Dask metadata."""
        cols = {
            "siteid": str,
            "location": str,
            "variable": str,
            "obs": float,
            "units": str,
            "country": str,
            "city": str,
            "is_mobile": bool,
            "is_analysis": object,
            "entity": str,
            "sensor_type": str,
            "time": "datetime64[us]",
            "time_local": "datetime64[us]",
            "utcoffset": "timedelta64[us]",
            "latitude": float,
            "longitude": float,
        }
        df = pd.DataFrame({k: pd.Series(dtype=v) for k, v in cols.items()})
        return df

    def _fetch_data(self, **kwargs) -> pd.DataFrame:
        """Internal fetch logic (Eager)."""
        dates = kwargs.get("dates")
        parameters = kwargs.get("parameters")
        country = kwargs.get("country")
        sites = kwargs.get("sites")
        entity = kwargs.get("entity")
        sensor_type = kwargs.get("sensor_type")
        query_time_split = kwargs.get("query_time_split", "1h")
        search_radius = kwargs.get("search_radius")

        query_dt = (
            pd.to_timedelta(query_time_split) if query_time_split and len(dates) > 1 else None
        )
        date_min, date_max = dates.min(), dates.max()

        def iter_time_slices():
            one_sec = pd.Timedelta(seconds=1)
            if query_dt is not None and date_min < date_max:
                t = date_min
                while t < date_max:
                    t_next = min(t + query_dt, date_max)
                    yield t - one_sec, t_next
                    t = t_next
            else:
                yield date_min - one_sec, date_max

        base_params = {}
        if country is not None:
            base_params.update(country=country)
        if sites is not None:
            base_params.update(location_id=sites)
        if entity is not None:
            base_params.update(entity=entity)
        if sensor_type is not None:
            base_params.update(sensor_type=sensor_type)

        def iter_queries():
            for parameter in parameters:
                for t_from, t_to in iter_time_slices():
                    if search_radius is not None:
                        for coords, radius in search_radius.items():
                            lat, lon = coords
                            yield {
                                **base_params,
                                "parameter": parameter,
                                "date_from": t_from,
                                "date_to": t_to,
                                "coordinates": f"{lat:.8f},{lon:.8f}",
                                "radius": radius,
                            }
                    else:
                        yield {
                            **base_params,
                            "parameter": parameter,
                            "date_from": t_from,
                            "date_to": t_to,
                        }

        # Clean kwargs for _consume
        consume_kwargs = {
            k: v for k, v in kwargs.items() if k in ["timeout", "retry", "limit", "npages"]
        }

        threads = kwargs.get("threads", None)
        if threads is not None:
            import concurrent.futures
            from itertools import chain

            with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as executor:
                data = chain.from_iterable(
                    executor.map(
                        lambda p: _consume(_ENDPOINTS["measurements"], params=p, **consume_kwargs),
                        iter_queries(),
                    )
                )
        else:
            data = []
            for p in iter_queries():
                this_data = _consume(_ENDPOINTS["measurements"], params=p, **consume_kwargs)
                data.extend(this_data)

        df = pd.DataFrame(data)
        if df.empty:
            return self._get_meta()

        to_expand = ["date", "coordinates"]
        to_expand = [c for c in to_expand if c in df.columns]
        new = pd.json_normalize(json.loads(df[to_expand].to_json(orient="records")))

        if "date.utc" in new.columns:
            time = pd.to_datetime(new["date.utc"]).dt.tz_localize(None)
        else:
            time = pd.Series(np.nan, index=new.index, dtype="datetime64[ns]")

        if "date.local" in new.columns:
            time_local = pd.to_datetime(new["date.local"].str.slice(0, 19))
        else:
            time_local = time.copy()

        utcoffset = time_local - time

        lat = new["coordinates.latitude"] if "coordinates.latitude" in new.columns else np.nan
        lon = new["coordinates.longitude"] if "coordinates.longitude" in new.columns else np.nan

        df = df.drop(columns=to_expand).assign(
            time=time,
            time_local=time_local,
            utcoffset=utcoffset,
            latitude=lat,
            longitude=lon,
        )

        df = df.rename(
            columns={
                "locationId": "siteid",
                "isMobile": "is_mobile",
                "isAnalysis": "is_analysis",
                "sensorType": "sensor_type",
                "parameter": "variable",
                "unit": "units",
                "value": "obs",
            },
        )
        df["siteid"] = df.siteid.astype(str)

        meta = self._get_meta()
        for col in meta.columns:
            if col not in df.columns:
                df[col] = pd.Series(dtype=meta[col].dtype)

        # Force exact dtypes from meta to avoid dask mismatch
        for col in meta.columns:
            df[col] = df[col].astype(meta[col].dtype)

        return df[meta.columns]

    def harmonize(
        self, df: Union[pd.DataFrame, "dd.DataFrame"]
    ) -> Union[pd.DataFrame, "dd.DataFrame"]:
        """
        Harmonize the dataset (standard naming, dropping NaNs, unit conversion).

        Parameters
        ----------
        df : Union[pd.DataFrame, dd.DataFrame]
            Input dataframe.

        Returns
        -------
        Union[pd.DataFrame, dd.DataFrame]
            Harmonized dataframe.
        """
        df = super().harmonize(df)

        non_neg_units = [
            "particles/cm³",
            "ppm",
            "ppb",
            "umol/mol",
            "µg/m³",
            "ugm3",
            "ng/m3",
            "iaq",
            "%",
            "m/s",
            "hpa",
            "mb",
        ]

        def _clean_values(df_part):
            if df_part.empty:
                return df_part
            if "units" in df_part.columns and "obs" in df_part.columns:
                mask = df_part.units.isin(non_neg_units) & (df_part.obs < 0)
                df_part.loc[mask, "obs"] = np.nan
            return df_part

        def _convert_units(df_part):
            if df_part.empty:
                return df_part
            for vn, f in _PPM_TO_UGM3.items():
                if "variable" in df_part.columns and "units" in df_part.columns:
                    is_ug = (df_part.variable == vn) & (df_part.units == "µg/m³")
                    df_part.loc[is_ug, "obs"] /= f
                    df_part.loc[is_ug, "units"] = "ppm"
            return df_part

        try:
            import dask.dataframe as dd

            is_dask = isinstance(df, dd.DataFrame)
        except ImportError:
            is_dask = False

        if is_dask:
            df = df.map_partitions(_clean_values)
            df = df.map_partitions(_convert_units)
            df = df.drop_duplicates(subset=["time", "siteid", "variable"])
        else:
            df = _clean_values(df)
            df = _convert_units(df)
            df = df.drop_duplicates(subset=["time", "siteid", "variable"])

        df = update_history(df, "Cleaned negative values, converted units, and dropped duplicates.")
        return df

    def to_xarray(
        self, df: Union[pd.DataFrame, "dd.DataFrame"], wide_fmt: bool = True, **kwargs: Any
    ) -> xr.Dataset:
        """
        Convert to xarray and rename variables for consistency.

        Parameters
        ----------
        df : Union[pd.DataFrame, dd.DataFrame]
            Input dataframe.
        wide_fmt : bool, optional
            Whether to expand to wide format, by default True.
        **kwargs : Any
            Additional arguments.

        Returns
        -------
        xr.Dataset
            The loaded dataset.
        """
        ds = super().to_xarray(df, expand2d=wide_fmt, **kwargs)

        if wide_fmt:
            rename_dict = {}
            for v in _PPM_TO_UGM3:
                if v in ds.data_vars:
                    ds[v].attrs["units"] = "ppm"
                    rename_dict[v] = f"{v}_ppm"
            for v in _NON_MOLEC_PARAMS:
                if v in ds.data_vars:
                    ds[v].attrs["units"] = "ug/m3"
                    rename_dict[v] = f"{v}_ugm3"

            if rename_dict:
                ds = ds.rename(rename_dict)
                from .base import _format_units

                ds = _format_units(ds)
                ds = update_history(ds, f"Renamed variables: {list(rename_dict.values())}")

        return ds

harmonize(df)

Harmonize the dataset (standard naming, dropping NaNs, unit conversion).

Parameters:

Name	Type	Description	Default
df	Union[DataFrame, DataFrame]	Input dataframe.	required

Returns:

Type	Description
Union[DataFrame, DataFrame]	Harmonized dataframe.

Source code in monetio/readers/openaq_v2.py

def harmonize(
    self, df: Union[pd.DataFrame, "dd.DataFrame"]
) -> Union[pd.DataFrame, "dd.DataFrame"]:
    """
    Harmonize the dataset (standard naming, dropping NaNs, unit conversion).

    Parameters
    ----------
    df : Union[pd.DataFrame, dd.DataFrame]
        Input dataframe.

    Returns
    -------
    Union[pd.DataFrame, dd.DataFrame]
        Harmonized dataframe.
    """
    df = super().harmonize(df)

    non_neg_units = [
        "particles/cm³",
        "ppm",
        "ppb",
        "umol/mol",
        "µg/m³",
        "ugm3",
        "ng/m3",
        "iaq",
        "%",
        "m/s",
        "hpa",
        "mb",
    ]

    def _clean_values(df_part):
        if df_part.empty:
            return df_part
        if "units" in df_part.columns and "obs" in df_part.columns:
            mask = df_part.units.isin(non_neg_units) & (df_part.obs < 0)
            df_part.loc[mask, "obs"] = np.nan
        return df_part

    def _convert_units(df_part):
        if df_part.empty:
            return df_part
        for vn, f in _PPM_TO_UGM3.items():
            if "variable" in df_part.columns and "units" in df_part.columns:
                is_ug = (df_part.variable == vn) & (df_part.units == "µg/m³")
                df_part.loc[is_ug, "obs"] /= f
                df_part.loc[is_ug, "units"] = "ppm"
        return df_part

    try:
        import dask.dataframe as dd

        is_dask = isinstance(df, dd.DataFrame)
    except ImportError:
        is_dask = False

    if is_dask:
        df = df.map_partitions(_clean_values)
        df = df.map_partitions(_convert_units)
        df = df.drop_duplicates(subset=["time", "siteid", "variable"])
    else:
        df = _clean_values(df)
        df = _convert_units(df)
        df = df.drop_duplicates(subset=["time", "siteid", "variable"])

    df = update_history(df, "Cleaned negative values, converted units, and dropped duplicates.")
    return df

open_dataset(dates=None, parameters=None, country=None, sites=None, wide_fmt=True, as_xarray=True, lazy=False, **kwargs)

Retrieves OpenAQ data via the REST API.

Parameters:

Name	Type	Description	Default
dates	Union[DatetimeIndex, List[datetime], datetime, str]	Dates to retrieve.	None
parameters	List[str]	Species to retrieve, by default ['pm25', 'o3'].	None
country	Union[str, List[str]]	Country code(s).	None
sites	List[str]	Site ID(s).	None
wide_fmt	bool	Whether to return data in wide format, by default True.	True
as_xarray	bool	Whether to return an xarray.Dataset, by default True.	True
lazy	bool	Whether to return a dask-backed object, by default False.	False
**kwargs	dict	Additional arguments passed to the API.	{}

Returns:

Type	Description
Union[DataFrame, Dataset, DataFrame]	The loaded data.

Examples:

>>> from monetio.readers.openaq_v2 import OpenAQV2Reader
>>> reader = OpenAQV2Reader()
>>> df = reader.open_dataset(dates='2023-01-01', as_xarray=False)

Source code in monetio/readers/openaq_v2.py

@_api_key_warning
def open_dataset(
    self,
    dates: pd.DatetimeIndex | list[datetime] | datetime | str = None,
    parameters: list[str] = None,
    country: str | list[str] = None,
    sites: list[str] = None,
    wide_fmt: bool = True,
    as_xarray: bool = True,
    lazy: bool = False,
    **kwargs: Any,
) -> Union[pd.DataFrame, xr.Dataset, "dd.DataFrame"]:
    """
    Retrieves OpenAQ data via the REST API.

    Parameters
    ----------
    dates : Union[pd.DatetimeIndex, List[datetime], datetime, str]
        Dates to retrieve.
    parameters : List[str], optional
        Species to retrieve, by default ['pm25', 'o3'].
    country : Union[str, List[str]], optional
        Country code(s).
    sites : List[str], optional
        Site ID(s).
    wide_fmt : bool, optional
        Whether to return data in wide format, by default True.
    as_xarray : bool, optional
        Whether to return an xarray.Dataset, by default True.
    lazy : bool, optional
        Whether to return a dask-backed object, by default False.
    **kwargs : dict
        Additional arguments passed to the API.

    Returns
    -------
    Union[pd.DataFrame, xr.Dataset, dd.DataFrame]
        The loaded data.

    Examples
    --------
    >>> from monetio.readers.openaq_v2 import OpenAQV2Reader
    >>> reader = OpenAQV2Reader()
    >>> df = reader.open_dataset(dates='2023-01-01', as_xarray=False)
    """
    if dates is None:
        raise ValueError("must provide at least one datetime-like via 'dates'")

    dates = pd.to_datetime(dates)
    if pd.api.types.is_scalar(dates):
        dates = pd.DatetimeIndex([dates])
    dates = dates.dropna()
    if dates.empty:
        raise ValueError("must provide at least one datetime-like")

    if parameters is None:
        parameters = ["pm25", "o3"]
    elif isinstance(parameters, str):
        parameters = [parameters]

    if lazy:
        import dask.dataframe as dd
        from dask import delayed

        query_time_split = kwargs.get("query_time_split", "1D")
        query_dt = pd.to_timedelta(query_time_split)
        date_min, date_max = dates.min(), dates.max()

        def iter_time_slices():
            one_sec = pd.Timedelta(seconds=1)
            if date_min < date_max:
                t = date_min
                while t < date_max:
                    t_next = min(t + query_dt, date_max)
                    yield t - one_sec, t_next
                    t = t_next
            else:
                yield date_min - one_sec, date_max

        delayed_dfs = []
        for t_from, t_to in iter_time_slices():
            for p in parameters:
                part_kwargs = kwargs.copy()
                part_kwargs.update(
                    dates=pd.DatetimeIndex([t_from + pd.Timedelta(seconds=1), t_to]),
                    parameters=[p],
                    country=country,
                    sites=sites,
                    query_time_split=None,
                )
                delayed_dfs.append(delayed(self._fetch_data)(**part_kwargs))

        if not delayed_dfs:
            df = dd.from_pandas(pd.DataFrame(), npartitions=1)
        else:
            meta = self._get_meta()
            df = dd.from_delayed(delayed_dfs, meta=meta)
    else:
        df = self._fetch_data(
            dates=dates, parameters=parameters, country=country, sites=sites, **kwargs
        )

    df = self.harmonize(df)

    df = force_object_strings(df)

    if as_xarray:
        return self.to_xarray(df, wide_fmt=wide_fmt, **kwargs)

    if wide_fmt:
        from ..util import long_to_wide

        df = long_to_wide(df)

    return df

to_xarray(df, wide_fmt=True, **kwargs)

Convert to xarray and rename variables for consistency.

Parameters:

Name	Type	Description	Default
df	Union[DataFrame, DataFrame]	Input dataframe.	required
wide_fmt	bool	Whether to expand to wide format, by default True.	True
**kwargs	Any	Additional arguments.	{}

Returns:

Type	Description
Dataset	The loaded dataset.

Source code in monetio/readers/openaq_v2.py

def to_xarray(
    self, df: Union[pd.DataFrame, "dd.DataFrame"], wide_fmt: bool = True, **kwargs: Any
) -> xr.Dataset:
    """
    Convert to xarray and rename variables for consistency.

    Parameters
    ----------
    df : Union[pd.DataFrame, dd.DataFrame]
        Input dataframe.
    wide_fmt : bool, optional
        Whether to expand to wide format, by default True.
    **kwargs : Any
        Additional arguments.

    Returns
    -------
    xr.Dataset
        The loaded dataset.
    """
    ds = super().to_xarray(df, expand2d=wide_fmt, **kwargs)

    if wide_fmt:
        rename_dict = {}
        for v in _PPM_TO_UGM3:
            if v in ds.data_vars:
                ds[v].attrs["units"] = "ppm"
                rename_dict[v] = f"{v}_ppm"
        for v in _NON_MOLEC_PARAMS:
            if v in ds.data_vars:
                ds[v].attrs["units"] = "ug/m3"
                rename_dict[v] = f"{v}_ugm3"

        if rename_dict:
            ds = ds.rename(rename_dict)
            from .base import _format_units

            ds = _format_units(ds)
            ds = update_history(ds, f"Renamed variables: {list(rename_dict.values())}")

    return ds

get_locations(**kwargs)

Get available site info (including site IDs) from OpenAQ v2 API.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Arguments passed to _consume (e.g. limit, npages).	{}

Returns:

Type	Description
DataFrame	Available site info.

Examples:

>>> from monetio.readers.openaq_v2 import get_locations
>>> sites = get_locations(limit=10)

Source code in monetio/readers/openaq_v2.py

@_api_key_warning
def get_locations(**kwargs) -> pd.DataFrame:
    """
    Get available site info (including site IDs) from OpenAQ v2 API.

    Parameters
    ----------
    **kwargs : Any
        Arguments passed to _consume (e.g. limit, npages).

    Returns
    -------
    pd.DataFrame
        Available site info.

    Examples
    --------
    >>> from monetio.readers.openaq_v2 import get_locations
    >>> sites = get_locations(limit=10)
    """
    data = _consume(_ENDPOINTS["locations"], **kwargs)

    some_scalars = [
        "id",
        "name",
        "city",
        "country",
        "isMobile",
        "firstUpdated",
        "lastUpdated",
    ]

    data2 = []
    for d in data:
        lat = d["coordinates"]["latitude"]
        lon = d["coordinates"]["longitude"]
        parameters = [p["parameter"] for p in d["parameters"]]
        mfs = d["manufacturers"]
        manufacturer = mfs[0]["manufacturerName"] if mfs else None
        d2 = {k: d[k] for k in some_scalars}
        d2.update(
            latitude=lat,
            longitude=lon,
            parameters=parameters,
            manufacturer=manufacturer,
        )
        data2.append(d2)

    df = pd.DataFrame(data2)
    if df.empty:
        return df

    df["firstUpdated"] = pd.to_datetime(df.firstUpdated.str.slice(0, 19))
    df["lastUpdated"] = pd.to_datetime(df.lastUpdated.str.slice(0, 19))

    df = df.rename(columns={"id": "siteid"})
    df["siteid"] = df.siteid.astype(str)
    df = df.drop_duplicates("siteid", keep="first").reset_index(drop=True)

    return df

get_parameters(**kwargs)

Get supported parameter info from OpenAQ v2 API.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Arguments passed to _consume (e.g. limit, npages).	{}

Returns:

Type	Description
DataFrame	Supported parameters.

Examples:

>>> from monetio.readers.openaq_v2 import get_parameters
>>> params = get_parameters()

Source code in monetio/readers/openaq_v2.py

def get_parameters(**kwargs) -> pd.DataFrame:
    """
    Get supported parameter info from OpenAQ v2 API.

    Parameters
    ----------
    **kwargs : Any
        Arguments passed to _consume (e.g. limit, npages).

    Returns
    -------
    pd.DataFrame
        Supported parameters.

    Examples
    --------
    >>> from monetio.readers.openaq_v2 import get_parameters
    >>> params = get_parameters()
    """
    data = _consume(_ENDPOINTS["parameters"], **kwargs)
    return pd.DataFrame(data)