CordexClient reference

Source code in c3s_event_attribution_tools/data/cordex_client.py

class CordexClient:
    def __init__(self, cordex_token: str):
        """
        Placeholder for CordexClient
        """
        self.cordex_token = cordex_token

    def fetch_cordex_xr(
        self,
        variable: str,
        model_url: str,
        bbox: tuple[float, float, float, float],
        time_range: tuple[datetime, datetime],
    ) -> xr.Dataset:
        """
        Fetch CORDEX data as an xarray Dataset for a given variable, model, bounding box, and time range.

        Parameters:
            variable (str): The variable to fetch (e.g., 'tasmax').
            model_url (str): The model URL segment to access the specific dataset. (eg. eur11-hist-day-cccma_canesm2-clmcom_clm_cclm4_8_17-r1i1p1)
            bbox (tuple): A tuple defining the bounding box (min_lon, min_lat, max_lon, max_lat).
            time_range (tuple): A tuple defining the time range (start_time, end_time).
        """
        headers = {
            "Authorization": f"Bearer {self.cordex_token}",
        }

        ds = xr.open_zarr(model_url, consolidated=True, storage_options={"headers": headers})
        variable_ds = ds[variable]
        bbox_filtered_ds = variable_ds.sel(
            longitude=slice(bbox[0], bbox[2]),
            latitude=slice(bbox[1], bbox[3]),
            time=slice(time_range[0], time_range[1]),
        )

        out_ds = bbox_filtered_ds.to_dataset()
        return out_ds

    def fetch_cordex_gpd(
        self,
        variable: str,
        model_url: str,
        bbox: tuple[float, float, float, float],
        time_range: tuple[datetime, datetime],
    ) -> gpd.GeoDataFrame:
        """
        Fetch CORDEX data as a GeoDataFrame for a given variable, model, GeoDataFrame, and time range.

        Parameters:
            variable (str): The variable to fetch (e.g., 'tasmax').
            model_url (str): The model URL segment to access the specific dataset. (eg. eur11-hist-day-cccma_canesm2-clmcom_clm_cclm4_8_17-r1i1p1)
            bbox (tuple): A tuple defining the bounding box (min_lat, min_lon, max_lat, max_lon).
            time_range (tuple): A tuple defining the time range (start_time, end_time).
        """
        ds = self.fetch_cordex_xr(variable, model_url, bbox, time_range)

        df = ds.to_dataframe().reset_index()
        # create geometry
        df['geometry'] = gpd.points_from_xy(df['longitude'], df['latitude'])
        gdf = gpd.GeoDataFrame(df, geometry='geometry', crs='EPSG:4326')

        return gdf

    def list_available_models(self) -> list[str]:
        """
        Placeholder method to list available CORDEX models.
        In a real implementation, this would query the CORDEX data store.
        """
        # This is a placeholder implementation.
        return [
            "eur11-hist-day-cccma_canesm2-clmcom_clm_cclm4_8_17-r1i1p1",
            # Add more models as needed
        ]

__init__(cordex_token)

Placeholder for CordexClient

Source code in c3s_event_attribution_tools/data/cordex_client.py

def __init__(self, cordex_token: str):
    """
    Placeholder for CordexClient
    """
    self.cordex_token = cordex_token

fetch_cordex_gpd(variable, model_url, bbox, time_range)

Fetch CORDEX data as a GeoDataFrame for a given variable, model, GeoDataFrame, and time range.

Parameters:

Name	Type	Description	Default
variable	str	The variable to fetch (e.g., 'tasmax').	required
model_url	str	The model URL segment to access the specific dataset. (eg. eur11-hist-day-cccma_canesm2-clmcom_clm_cclm4_8_17-r1i1p1)	required
bbox	tuple	A tuple defining the bounding box (min_lat, min_lon, max_lat, max_lon).	required
time_range	tuple	A tuple defining the time range (start_time, end_time).	required

Source code in c3s_event_attribution_tools/data/cordex_client.py

def fetch_cordex_gpd(
    self,
    variable: str,
    model_url: str,
    bbox: tuple[float, float, float, float],
    time_range: tuple[datetime, datetime],
) -> gpd.GeoDataFrame:
    """
    Fetch CORDEX data as a GeoDataFrame for a given variable, model, GeoDataFrame, and time range.

    Parameters:
        variable (str): The variable to fetch (e.g., 'tasmax').
        model_url (str): The model URL segment to access the specific dataset. (eg. eur11-hist-day-cccma_canesm2-clmcom_clm_cclm4_8_17-r1i1p1)
        bbox (tuple): A tuple defining the bounding box (min_lat, min_lon, max_lat, max_lon).
        time_range (tuple): A tuple defining the time range (start_time, end_time).
    """
    ds = self.fetch_cordex_xr(variable, model_url, bbox, time_range)

    df = ds.to_dataframe().reset_index()
    # create geometry
    df['geometry'] = gpd.points_from_xy(df['longitude'], df['latitude'])
    gdf = gpd.GeoDataFrame(df, geometry='geometry', crs='EPSG:4326')

    return gdf

fetch_cordex_xr(variable, model_url, bbox, time_range)

Fetch CORDEX data as an xarray Dataset for a given variable, model, bounding box, and time range.

Parameters:

Name	Type	Description	Default
variable	str	The variable to fetch (e.g., 'tasmax').	required
model_url	str	The model URL segment to access the specific dataset. (eg. eur11-hist-day-cccma_canesm2-clmcom_clm_cclm4_8_17-r1i1p1)	required
bbox	tuple	A tuple defining the bounding box (min_lon, min_lat, max_lon, max_lat).	required
time_range	tuple	A tuple defining the time range (start_time, end_time).	required

Source code in c3s_event_attribution_tools/data/cordex_client.py

def fetch_cordex_xr(
    self,
    variable: str,
    model_url: str,
    bbox: tuple[float, float, float, float],
    time_range: tuple[datetime, datetime],
) -> xr.Dataset:
    """
    Fetch CORDEX data as an xarray Dataset for a given variable, model, bounding box, and time range.

    Parameters:
        variable (str): The variable to fetch (e.g., 'tasmax').
        model_url (str): The model URL segment to access the specific dataset. (eg. eur11-hist-day-cccma_canesm2-clmcom_clm_cclm4_8_17-r1i1p1)
        bbox (tuple): A tuple defining the bounding box (min_lon, min_lat, max_lon, max_lat).
        time_range (tuple): A tuple defining the time range (start_time, end_time).
    """
    headers = {
        "Authorization": f"Bearer {self.cordex_token}",
    }

    ds = xr.open_zarr(model_url, consolidated=True, storage_options={"headers": headers})
    variable_ds = ds[variable]
    bbox_filtered_ds = variable_ds.sel(
        longitude=slice(bbox[0], bbox[2]),
        latitude=slice(bbox[1], bbox[3]),
        time=slice(time_range[0], time_range[1]),
    )

    out_ds = bbox_filtered_ds.to_dataset()
    return out_ds

list_available_models()

Placeholder method to list available CORDEX models. In a real implementation, this would query the CORDEX data store.

Source code in c3s_event_attribution_tools/data/cordex_client.py

def list_available_models(self) -> list[str]:
    """
    Placeholder method to list available CORDEX models.
    In a real implementation, this would query the CORDEX data store.
    """
    # This is a placeholder implementation.
    return [
        "eur11-hist-day-cccma_canesm2-clmcom_clm_cclm4_8_17-r1i1p1",
        # Add more models as needed
    ]