LamaH-CE

Overview

LamaH-CE is the Central Europe large-sample hydrological dataset. Large-sample hydrological dataset for Central Europe, covering diverse Alpine and pre-Alpine catchments with high-quality data.

Dataset Information

• Region: Central Europe
• Project: LamaH (Large-sample hydrological data and models)
• Module: hydrodataset.lamah_ce
• Class: LamahCe

About LamaH

LamaH (Large-sample hydrological data and models) provides comprehensive hydrological data for research and modeling:

Key Features

• High-quality, quality-controlled data
• Extensive catchment attributes
• Multiple temporal resolutions
• Detailed metadata
• Suitable for large-sample hydrology studies

Research Applications

• Hydrological model development and testing
• Climate change impact studies
• Regionalization studies
• Machine learning applications
• Comparative hydrology

Features

Static Attributes

Comprehensive static catchment attributes: - Basin geometry and area - Topographic characteristics (elevation, slope) - Land cover information - Soil properties and classes - Geological characteristics - Climate indices - Human influence indicators

Dynamic Variables

Timeseries variables available: - Streamflow (observed) - Precipitation - Temperature (min, max, mean) - Potential evapotranspiration - Snow water equivalent - Solar radiation - Humidity - And more...

Usage

Basic Usage

from hydrodataset.lamah_ce import LamahCe
from hydrodataset import resolve_data_path

# Initialize dataset
data_path = resolve_data_path("lamah_ce")
ds = LamahCe(data_path)

# Get basin IDs
basin_ids = ds.read_object_ids()
print(f"Number of basins: {len(basin_ids)}")

# Check available features
print("Static features:", ds.available_static_features)
print("Dynamic features:", ds.available_dynamic_features)

# Check default time range
print(f"Default time range: {ds.default_t_range}")

# Read timeseries data
timeseries = ds.read_ts_xrdataset(
    gage_id_lst=basin_ids[:5],
    t_range=ds.default_t_range,
    var_lst=["streamflow", "precipitation"]
)
print(timeseries)

# Read attribute data
attributes = ds.read_attr_xrdataset(
    gage_id_lst=basin_ids[:5],
    var_lst=["area", "p_mean"]
)
print(attributes)

Advanced Analysis

# Read multiple variables for detailed analysis
ts_data = ds.read_ts_xrdataset(
    gage_id_lst=basin_ids[:10],
    t_range=["1990-01-01", "2020-12-31"],
    var_lst=[
        "streamflow",
        "precipitation", 
        "temperature_mean",
        "temperature_min",
        "temperature_max",
        "pet",
        "snow_water_equivalent"
    ]
)

# Analyze snow-influenced catchments
import xarray as xr
winter_months = ts_data.sel(time=ts_data.time.dt.month.isin([12, 1, 2]))
mean_swe = winter_months["snow_water_equivalent"].mean(dim="time")
print("Mean winter SWE:", mean_swe)

Reading Specific Variables

# Read with specific time range
ts_data = ds.read_ts_xrdataset(
    gage_id_lst=basin_ids[:10],
    t_range=["2000-01-01", "2010-12-31"],
    var_lst=["streamflow", "precipitation", "temperature_mean"]
)

# Read basin area
areas = ds.read_area(gage_id_lst=basin_ids[:10])

# Read mean precipitation
mean_precip = ds.read_mean_prcp(gage_id_lst=basin_ids[:10])

Station Connectivity Data

LamaH-CE provides stream network connectivity information between gauging stations. This data is essential for hydrological routing models and understanding the river network topology.

Connectivity Variables

The station connectivity data includes the following variables (from Stream_dist.csv):

Variable	Description	Unit
NEXTDOWNID	ID of the next downstream gauge (only one); 0 indicates no downstream gauge	-
dist_hdn	Horizontal stream length from the actual gauge to the next downstream gauge	km
elev_diff	Elevation difference from the actual gauge's zero point to the next downstream gauge's zero point	m
strm_slope	Slope of the actual gauge to the next downstream gauge; fraction of elev_diff and dist_hdn	m km⁻¹

Basic Usage

from hydrodataset.lamah_ce import LamahCe
from hydrodataset import resolve_data_path

# Initialize dataset
data_path = resolve_data_path("lamah_ce")
ds = LamahCe(data_path)

# Cache the station connectivity data (only needed once)
# This reads Stream_dist.csv and saves it as NetCDF
ds.cache_stations_xrdataset()

# Read station connectivity for specific stations
stations = ds.read_stations_xrdataset(
    station_id_lst=["3", "4"]
)
print(stations)

# Example output:
# <xarray.Dataset>
# Dimensions:     (ID: 2)
# Coordinates:
#   * ID          (ID) <U3 '3' '4'
# Data variables:
#     NEXTDOWNID  (ID) <U3 '2' '3'
#     dist_hdn    (ID) <U18 '8.9' '12.3'
#     elev_diff   (ID) <U5 '45.0' '32.0'
#     strm_slope  (ID) <U19 '5.06' '2.60'

# Read all station connectivity data (no filter)
all_stations = ds.read_stations_xrdataset()
print(f"Total stations: {all_stations.dims['ID']}")

Finding Downstream Stations

# Find downstream station for a specific gauge
station_id = "114"
station_data = ds.read_stations_xrdataset(station_id_lst=[station_id])
downstream_id = station_data["NEXTDOWNID"].values[0]
print(f"Downstream station of {station_id}: {downstream_id}")

Network Traversal Example

def get_downstream_path(ds, start_station_id, max_depth=10):
    """Get the downstream path from a starting station.

    Args:
        ds: LamahCe dataset instance
        start_station_id: Starting station ID
        max_depth: Maximum number of downstream stations to traverse

    Returns:
        List of station IDs from start to outlet
    """
    path = [start_station_id]
    current_id = start_station_id

    for _ in range(max_depth):
        station_data = ds.read_stations_xrdataset(station_id_lst=[current_id])
        next_id = station_data["NEXTDOWNID"].values[0]

        if next_id == "0":  # No downstream gauge (outlet)
            break

        path.append(next_id)
        current_id = next_id

    return path

# Example: trace downstream from station 114
downstream_path = get_downstream_path(ds, "114")
print(f"Downstream path: {' -> '.join(downstream_path)}")

Data Quality and Completeness

LamaH datasets feature: - Rigorous quality control procedures - Documentation of data gaps - Metadata completeness - Peer-reviewed methodology - Regular updates

Regional Characteristics

LamaH-CE

• Alpine and pre-Alpine catchments
• Snow-influenced hydrology
• Elevation range from lowlands to high mountains
• Mixed land use patterns

LamaH-ICE

• Volcanic landscapes
• Glacial-influenced catchments
• Geothermal activity impact
• Unique geological conditions

API Reference

hydrodataset.lamah_ce.LamahCe

Bases: HydroDataset

LamaHCE dataset class extending HydroDataset.

This class provides access to the LamaHCE dataset, which contains hourly hydrological and meteorological data for various watersheds.

Attributes:

Name	Type	Description
region		Geographic region identifier
download		Whether to download data automatically
ds_description		Dictionary containing dataset file paths

Source code in hydrodataset/lamah_ce.py

class LamahCe(HydroDataset):
    """LamaHCE dataset class extending HydroDataset.

    This class provides access to the LamaHCE dataset, which contains hourly
    hydrological and meteorological data for various watersheds.

    Attributes:
        region: Geographic region identifier
        download: Whether to download data automatically
        ds_description: Dictionary containing dataset file paths
    """

    def __init__(
        self,
        uri: str,
        region: Optional[str] = None,
        download: bool = False,
        cache_path: Optional[str] = None,
    ) -> None:
        """Initialize LamaHCE dataset.

        Args:
            uri: Path to the data directory
            region: Geographic region identifier (optional)
            download: Whether to download data automatically (default: False)
            cache_path: Path to the cache directory
        """
        super().__init__(uri, cache_path=cache_path)
        self.region = region
        self.download = download
        # cloud path: aqua_fetch cannot read S3, use cache_*_to_zarr instead
        if str(uri).startswith("s3://"):
            return
        # Use the custom LamaHCE class defined at module level
        self.aqua_fetch = LamaHCE(uri)

    # OSS relative paths (timestep=D, data_type=total_upstrm)
    _CATCH_ATTR_REL = "LamaHCE/A_basins_total_upstrm/1_attributes"
    _METEO_REL = "LamaHCE/A_basins_total_upstrm/2_timeseries/daily"
    _GAUGE_ATTR_REL = "LamaHCE/D_gauges/1_attributes"
    _Q_REL = "LamaHCE/D_gauges/2_timeseries/daily"
    # AquaFetch LamaHCE.static_map
    _STATIC_RENAME = {
        "area_calc": "area_km2",
        "slope_mean": "slope_mkm-1",
        "lon": "long",
    }
    # AquaFetch LamaHCE.dyn_map['D'] resolved to cleaned names (q file's qobs
    # is first renamed to q_cms by AquaFetch)
    _DYN_RENAME = {
        "q_cms": "q_cms_obs",
        "2m_temp_min": "airtemp_c_min",
        "2m_temp_max": "airtemp_c_max",
        "2m_temp_mean": "airtemp_c_mean",
        "prec": "pcp_mm",
        "swe": "swe_mm",
        "surf_net_solar_rad_max": "solrad_wm2_max",
        "surf_net_solar_rad_mean": "solrad_wm2",
        "surf_net_therm_rad_max": "thermrad_wm2_max",
        "surf_net_therm_rad_mean": "thermrad_wm2",
        "10m_wind_u": "windspeedu_mps",
        "10m_wind_v": "windspeedv_mps",
        "2m_dp_temp_max": "dptemp_c_max_2m",
        "2m_dp_temp_mean": "dptemp_c_mean_2m",
        "2m_dp_temp_min": "dptemp_c_min_2m",
        "surf_press": "airpres_hpa",
    }

    def read_object_ids(self) -> np.ndarray:
        if self._is_cloud():
            fs = self._make_s3fs()
            uri = str(self.data_source_dir).rstrip("/")
            names = [p.split("/")[-1] for p in fs.ls(f"{uri}/{self._METEO_REL}".removeprefix("s3://"))]
            ids = sorted(
                (n.split("_")[1].split(".csv")[0] for n in names if n.startswith("ID_")),
                key=lambda x: int(x),
            )
            return np.array(ids)
        return super().read_object_ids()

    def cache_attributes_to_zarr(self) -> None:
        import zarr

        fs = self._make_s3fs()
        uri = str(self.data_source_dir).rstrip("/")

        def _read(rel, fname):
            with fs.open(f"{uri}/{rel}/{fname}".removeprefix("s3://")) as fh:
                df = pd.read_csv(fh, sep=";", index_col="ID")
            df.index = df.index.astype(str)
            return df

        cat = _read(self._CATCH_ATTR_REL, "Catchment_attributes.csv")
        gauge = _read(self._GAUGE_ATTR_REL, "Gauge_attributes.csv")
        static = pd.concat([cat, gauge], axis=1)
        static = static.loc[~static.index.duplicated(keep="first")]
        static = static.rename(columns=self._STATIC_RENAME)
        static.columns = self._clean_feature_names(list(static.columns))
        static = static.loc[:, ~static.columns.duplicated(keep="first")]

        zarr_name = self._attributes_cache_filename.replace(".nc", ".zarr")
        out, opts = self._zarr_path_and_opts(zarr_name)
        ids = static.index.tolist()
        n = len(ids)
        root = zarr.open_group(out, mode="w", storage_options=opts, zarr_format=2)
        for col in static.columns:
            vals = static[col].values.astype(str) if static[col].dtype == object else static[col].values
            arr = root.create_array(col, shape=(n,), chunks=(n,), dtype=vals.dtype)
            arr[:] = vals
            arr.attrs["_ARRAY_DIMENSIONS"] = ["basin"]
        basin_arr = root.create_array("basin", shape=(n,), chunks=(n,), dtype=str)
        basin_arr[:] = ids
        basin_arr.attrs["_ARRAY_DIMENSIONS"] = ["basin"]
        root.attrs["coordinates"] = "basin"
        self._write_zarr_units(root, "static")
        print(f"Attributes zarr written to: {out}")

    def cache_timeseries_to_zarr(self) -> None:
        import zarr

        fs = self._make_s3fs()
        uri = str(self.data_source_dir).rstrip("/")
        meteo_base = f"{uri}/{self._METEO_REL}"
        q_base = f"{uri}/{self._Q_REL}"

        stations = self.read_object_ids().tolist()
        all_times = pd.date_range(self.default_t_range[0], self.default_t_range[1], freq="D")
        n, nt = len(stations), len(all_times)
        times_ns = all_times.asi8

        cleaned_var_lst = []
        for info in self._dynamic_variable_mapping.values():
            for s in info["sources"].values():
                if s["specific_name"] not in cleaned_var_lst:
                    cleaned_var_lst.append(s["specific_name"])

        def _read_dated(path, q=False):
            with fs.open(path.removeprefix("s3://")) as fh:
                df = pd.read_csv(fh, sep=";")
            idx = pd.to_datetime(dict(year=df["YYYY"], month=df["MM"], day=df["DD"]))
            df = df.drop(columns=[c for c in ("YYYY", "MM", "DD", "DOY") if c in df.columns])
            df.index = idx
            if q:
                df = df.rename(columns={"qobs": "q_cms"})
            return df

        data = {vn: np.full((n, nt), np.nan) for vn in cleaned_var_lst}
        for i, stn in enumerate(tqdm(stations, desc="lamah_ce")):
            parts = []
            try:
                parts.append(_read_dated(f"{meteo_base}/ID_{stn}.csv"))
            except Exception as e:
                print(f"  WARN meteo {stn}: {e}")
            try:
                parts.append(_read_dated(f"{q_base}/ID_{stn}.csv", q=True))
            except Exception:
                pass
            if not parts:
                continue
            df = pd.concat(parts, axis=1)
            df = df.loc[~df.index.duplicated(keep="first")]
            df.columns = self._clean_feature_names(
                [self._DYN_RENAME.get(c, c) for c in df.columns]
            )
            if "airpres_hpa" in df.columns:  # AquaFetch dyn_factors: Pa -> hPa
                df["airpres_hpa"] = pd.to_numeric(df["airpres_hpa"], errors="coerce") * 0.01
            df = df.reindex(all_times)
            for vn in cleaned_var_lst:
                if vn in df.columns:
                    data[vn][i] = pd.to_numeric(df[vn], errors="coerce").values

        zarr_name = self._timeseries_cache_filename.replace(".nc", ".zarr")
        out, opts = self._zarr_path_and_opts(zarr_name)
        chunk_t = min(nt, 365)
        root = zarr.open_group(out, mode="w", storage_options=opts, zarr_format=2)
        for vn in cleaned_var_lst:
            arr = root.create_array(vn, shape=(n, nt), chunks=(min(n, 100), chunk_t),
                                    dtype="float64", fill_value=np.nan)
            arr[:] = data[vn]
            arr.attrs["_ARRAY_DIMENSIONS"] = ["basin", "time"]
        time_arr = root.create_array("time", shape=(nt,), chunks=(chunk_t,), dtype="int64")
        time_arr[:] = times_ns
        time_arr.attrs["_ARRAY_DIMENSIONS"] = ["time"]
        time_arr.attrs["units"] = "nanoseconds since 1970-01-01"
        time_arr.attrs["calendar"] = "proleptic_gregorian"
        basin_arr = root.create_array("basin", shape=(n,), chunks=(n,), dtype=str)
        basin_arr[:] = stations
        basin_arr.attrs["_ARRAY_DIMENSIONS"] = ["basin"]
        root.attrs["coordinates"] = "basin time"
        self._write_zarr_units(root, "dynamic")
        print(f"Timeseries zarr written to: {out}")

    @property
    def _attributes_cache_filename(self):
        return "lamahce_attributes.nc"

    @property
    def _timeseries_cache_filename(self):
        return "lamahce_timeseries.nc"

    @property
    def default_t_range(self):
        return ["1981-01-01", "2019-12-31"]

    # get the information of features from table 3 in "https://doi.org/10.5194/essd-13-4529-2021"
    # Static variable definitions based on inspected data
    _subclass_static_definitions = {
        "p_mean": {"specific_name": "p_mean", "unit": "mm/day"},
        "area": {"specific_name": "area_km2", "unit": "km^2"},
    }

    # Dynamic variable mapping based on inspected data
    _dynamic_variable_mapping = {
        StandardVariable.STREAMFLOW: {
            "default_source": "observations",
            "sources": {
                "observations": {"specific_name": "q_cms_obs", "unit": "m^3/s"},
            },
        },
        StandardVariable.PRECIPITATION: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "pcp_mm", "unit": "mm"},
            },
        },
        StandardVariable.TEMPERATURE_MAX: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "airtemp_c_max", "unit": "°C"},
                "dp": {"specific_name": "dptemp_c_max_2m", "unit": "°C"},
            },
        },
        StandardVariable.TEMPERATURE_MIN: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "airtemp_c_min", "unit": "°C"},
                "dp": {"specific_name": "dptemp_c_min_2m", "unit": "°C"},
            },
        },
        StandardVariable.TEMPERATURE_MEAN: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "airtemp_c_mean", "unit": "°C"},
                "dp": {"specific_name": "dptemp_c_mean_2m", "unit": "°C"},
            },
        },
        StandardVariable.EVAPOTRANSPIRATION: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "total_et", "unit": "mm"},
            },
        },
        StandardVariable.SNOW_WATER_EQUIVALENT: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "swe_mm", "unit": "mm"},
            },
        },
        StandardVariable.SOLAR_RADIATION: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "solrad_wm2", "unit": "W/m^2"},
            },
        },
        StandardVariable.SOLAR_RADIATION_MAX: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "solrad_wm2_max", "unit": "W/m^2"},
            },
        },
        StandardVariable.THERMAL_RADIATION: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "thermrad_wm2", "unit": "W/m^2"},
            },
        },
        StandardVariable.THERMAL_RADIATION_MAX: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "thermrad_wm2_max", "unit": "W/m^2"},
            },
        },
        StandardVariable.SURFACE_PRESSURE: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "airpres_hpa", "unit": "Pa"},
            },
        },
        StandardVariable.U_WIND_SPEED: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "windspeedu_mps", "unit": "m/s"},
            },
        },
        StandardVariable.V_WIND_SPEED: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "windspeedv_mps", "unit": "m/s"},
            },
        },
        StandardVariable.VOLUMETRIC_SOIL_WATER_LAYER1: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "volsw_123", "unit": "m^3/m^3"},
            },
        },
        StandardVariable.VOLUMETRIC_SOIL_WATER_LAYER4: {
            "default_source": "era5",
            "sources": {
                "era5": {"specific_name": "volsw_4", "unit": "m^3/m^3"},
            },
        },
    }

    @property
    def _stations_cache_filename(self):
        """Cache filename for stations mapping."""
        return "lamahce_stations.nc"

    # Stream-network topology file (static; identical across daily/hourly bundles,
    # so the extracted 1_LamaH-CE_daily_hourly copy is used). Relative to the
    # dataset root -> works for both local (F:/data) and cloud (s3://bucket).
    _STREAM_REL = (
        "LamaHCE/1_LamaH-CE_daily_hourly/B_basins_intermediate_all/"
        "1_attributes/Stream_dist.csv"
    )

    @staticmethod
    def _prep_stream_df(df):
        """Shared processing of Stream_dist.csv (used by local NC and cloud zarr).

        Keeps NEXTDOWNID/dist_hdn/elev_diff/strm_slope, ID as string index,
        numeric columns as float. No renaming/unit conversion (matches the raw
        file), so local and cloud results are identical.
        """
        df["ID"] = df["ID"].astype(str)
        df = df.set_index("ID")[["NEXTDOWNID", "dist_hdn", "elev_diff", "strm_slope"]]
        df["NEXTDOWNID"] = df["NEXTDOWNID"].astype(str)
        for col in ["dist_hdn", "elev_diff", "strm_slope"]:
            df[col] = pd.to_numeric(df[col], errors="coerce")
        return df

    def cache_stations_xrdataset(self):
        """Read Stream_dist.csv (local) and cache it as NetCDF (ID-indexed).

        Columns: NEXTDOWNID (downstream station id), dist_hdn, elev_diff,
        strm_slope. Output dims/coord: ID (string).
        """
        csv_path = os.path.join(str(self.data_source_dir), *self._STREAM_REL.split("/"))
        if not os.path.exists(csv_path):
            raise FileNotFoundError(f"Stream_dist.csv not found at {csv_path}")
        df = self._prep_stream_df(pd.read_csv(csv_path, sep=";"))
        output_path = self.cache_dir.joinpath(self._stations_cache_filename)
        df.to_xarray().to_netcdf(output_path)
        print(f"Stations stream data saved to: {output_path}")

    def cache_stations_to_zarr(self):
        """Read Stream_dist.csv from OSS and write the stations zarr (cloud)."""
        import zarr

        fs = self._make_s3fs()
        uri = str(self.data_source_dir).rstrip("/")
        path = f"{uri}/{self._STREAM_REL}".removeprefix("s3://")
        with fs.open(path) as fh:
            df = pd.read_csv(fh, sep=";")
        df = self._prep_stream_df(df)

        zarr_name = self._stations_cache_filename.replace(".nc", ".zarr")
        out, opts = self._zarr_path_and_opts(zarr_name)
        ids = df.index.tolist()
        n = len(ids)
        root = zarr.open_group(out, mode="w", storage_options=opts, zarr_format=2)
        for col in df.columns:
            vals = df[col].values.astype(str) if df[col].dtype == object else df[col].values
            arr = root.create_array(col, shape=(n,), chunks=(n,), dtype=vals.dtype)
            arr[:] = vals
            arr.attrs["_ARRAY_DIMENSIONS"] = ["ID"]
        id_arr = root.create_array("ID", shape=(n,), chunks=(n,), dtype=str)
        id_arr[:] = ids
        id_arr.attrs["_ARRAY_DIMENSIONS"] = ["ID"]
        root.attrs["coordinates"] = "ID"
        print(f"Stations zarr written to: {out}")

    def read_stations_xrdataset(
        self,
        station_id_lst: Union[str, List[str]] = None,
    ) -> xr.Dataset:
        """Read station stream data from cached NetCDF file.

        This function reads the station stream NetCDF file and returns the
        corresponding stream attributes for the given station IDs.
        If the cache file does not exist, it will be generated first.

        Args:
            station_id_lst: A single station ID or a list of station IDs to query.
                If None, returns all stations.

        Returns:
            An xarray Dataset containing the station stream data with
            variables: NEXTDOWNID, dist_hdn, elev_diff, strm_slope.
            The dimension and coordinate is ID (station ID as string).

        Examples:
            >>> ds = lamah_ce.read_stations_xrdataset(
            ...     station_id_lst=["114", "200"]
            ... )
            >>> print(ds)
        """
        if self._is_cloud():
            import zarr as _zarr

            out, opts = self._zarr_path_and_opts(
                self._stations_cache_filename.replace(".nc", ".zarr")
            )
            try:
                ds = xr.open_zarr(out, storage_options=opts, consolidated=False,
                                  mask_and_scale=False)
            except _zarr.errors.GroupNotFoundError:
                self.cache_stations_to_zarr()
                ds = xr.open_zarr(out, storage_options=opts, consolidated=False,
                                  mask_and_scale=False)
        else:
            # Load the local cache file, generate if not exists
            cache_file = self.cache_dir.joinpath(self._stations_cache_filename)
            if not os.path.isfile(cache_file):
                self.cache_stations_xrdataset()
            ds = xr.open_dataset(cache_file)

        # Filter by station_id if provided
        if station_id_lst is not None:
            # Convert station_id_lst to list of strings
            if isinstance(station_id_lst, (str, int)):
                station_id_lst = [str(station_id_lst)]
            else:
                station_id_lst = [str(sid) for sid in station_id_lst]

            # Select stations using ID coordinate
            ds = ds.sel(ID=station_id_lst)

        return ds

default_t_range property

__init__(uri, region=None, download=False, cache_path=None)

Initialize LamaHCE dataset.

Parameters:

Name	Type	Description	Default
uri	str	Path to the data directory	required
region	Optional[str]	Geographic region identifier (optional)	None
download	bool	Whether to download data automatically (default: False)	False
cache_path	Optional[str]	Path to the cache directory	None

Source code in hydrodataset/lamah_ce.py

def __init__(
    self,
    uri: str,
    region: Optional[str] = None,
    download: bool = False,
    cache_path: Optional[str] = None,
) -> None:
    """Initialize LamaHCE dataset.

    Args:
        uri: Path to the data directory
        region: Geographic region identifier (optional)
        download: Whether to download data automatically (default: False)
        cache_path: Path to the cache directory
    """
    super().__init__(uri, cache_path=cache_path)
    self.region = region
    self.download = download
    # cloud path: aqua_fetch cannot read S3, use cache_*_to_zarr instead
    if str(uri).startswith("s3://"):
        return
    # Use the custom LamaHCE class defined at module level
    self.aqua_fetch = LamaHCE(uri)

read_object_ids()

Source code in hydrodataset/lamah_ce.py

def read_object_ids(self) -> np.ndarray:
    if self._is_cloud():
        fs = self._make_s3fs()
        uri = str(self.data_source_dir).rstrip("/")
        names = [p.split("/")[-1] for p in fs.ls(f"{uri}/{self._METEO_REL}".removeprefix("s3://"))]
        ids = sorted(
            (n.split("_")[1].split(".csv")[0] for n in names if n.startswith("ID_")),
            key=lambda x: int(x),
        )
        return np.array(ids)
    return super().read_object_ids()

cache_stations_xrdataset()

Read Stream_dist.csv (local) and cache it as NetCDF (ID-indexed).

Columns: NEXTDOWNID (downstream station id), dist_hdn, elev_diff, strm_slope. Output dims/coord: ID (string).

Source code in hydrodataset/lamah_ce.py

def cache_stations_xrdataset(self):
    """Read Stream_dist.csv (local) and cache it as NetCDF (ID-indexed).

    Columns: NEXTDOWNID (downstream station id), dist_hdn, elev_diff,
    strm_slope. Output dims/coord: ID (string).
    """
    csv_path = os.path.join(str(self.data_source_dir), *self._STREAM_REL.split("/"))
    if not os.path.exists(csv_path):
        raise FileNotFoundError(f"Stream_dist.csv not found at {csv_path}")
    df = self._prep_stream_df(pd.read_csv(csv_path, sep=";"))
    output_path = self.cache_dir.joinpath(self._stations_cache_filename)
    df.to_xarray().to_netcdf(output_path)
    print(f"Stations stream data saved to: {output_path}")

read_stations_xrdataset(station_id_lst=None)

Read station stream data from cached NetCDF file.

This function reads the station stream NetCDF file and returns the corresponding stream attributes for the given station IDs. If the cache file does not exist, it will be generated first.

Parameters:

Name	Type	Description	Default
station_id_lst	Union[str, List[str]]	A single station ID or a list of station IDs to query. If None, returns all stations.	None

Returns:

Name	Type	Description
	Dataset	An xarray Dataset containing the station stream data with
variables	Dataset	NEXTDOWNID, dist_hdn, elev_diff, strm_slope.
	Dataset	The dimension and coordinate is ID (station ID as string).

Examples:

>>> ds = lamah_ce.read_stations_xrdataset(
...     station_id_lst=["114", "200"]
... )
>>> print(ds)

Source code in hydrodataset/lamah_ce.py

def read_stations_xrdataset(
    self,
    station_id_lst: Union[str, List[str]] = None,
) -> xr.Dataset:
    """Read station stream data from cached NetCDF file.

    This function reads the station stream NetCDF file and returns the
    corresponding stream attributes for the given station IDs.
    If the cache file does not exist, it will be generated first.

    Args:
        station_id_lst: A single station ID or a list of station IDs to query.
            If None, returns all stations.

    Returns:
        An xarray Dataset containing the station stream data with
        variables: NEXTDOWNID, dist_hdn, elev_diff, strm_slope.
        The dimension and coordinate is ID (station ID as string).

    Examples:
        >>> ds = lamah_ce.read_stations_xrdataset(
        ...     station_id_lst=["114", "200"]
        ... )
        >>> print(ds)
    """
    if self._is_cloud():
        import zarr as _zarr

        out, opts = self._zarr_path_and_opts(
            self._stations_cache_filename.replace(".nc", ".zarr")
        )
        try:
            ds = xr.open_zarr(out, storage_options=opts, consolidated=False,
                              mask_and_scale=False)
        except _zarr.errors.GroupNotFoundError:
            self.cache_stations_to_zarr()
            ds = xr.open_zarr(out, storage_options=opts, consolidated=False,
                              mask_and_scale=False)
    else:
        # Load the local cache file, generate if not exists
        cache_file = self.cache_dir.joinpath(self._stations_cache_filename)
        if not os.path.isfile(cache_file):
            self.cache_stations_xrdataset()
        ds = xr.open_dataset(cache_file)

    # Filter by station_id if provided
    if station_id_lst is not None:
        # Convert station_id_lst to list of strings
        if isinstance(station_id_lst, (str, int)):
            station_id_lst = [str(station_id_lst)]
        else:
            station_id_lst = [str(sid) for sid in station_id_lst]

        # Select stations using ID coordinate
        ds = ds.sel(ID=station_id_lst)

    return ds