CAMELS-NZ

Overview

CAMELS-NZ is the New Zealand hydrological dataset implementation. New Zealand CAMELS dataset covering diverse climates from subtropical to alpine.

Dataset Information

• Region: New Zealand
• Module: hydrodataset.camels_nz
• Class: CamelsNz

Features

Static Attributes

Static catchment attributes include: - Basin area - Mean precipitation - Topographic characteristics - Land cover information - Soil properties - Climate indices

Dynamic Variables

Timeseries variables available (varies by dataset): - Streamflow - Precipitation - Temperature (min, max, mean) - Potential evapotranspiration - Solar radiation - And more...

Usage

Basic Usage

from hydrodataset.camels_nz import CamelsNz
from hydrodataset import resolve_data_path

# Initialize dataset
data_path = resolve_data_path("camels_nz")
ds = CamelsNz(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)

# 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)

Reading Specific Variables

# Read with specific time range
ts_data = ds.read_ts_xrdataset(
    gage_id_lst=basin_ids[:10],
    t_range=["1990-01-01", "1995-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])

Data Sources

The dataset supports multiple data sources for certain variables. Check the class documentation for available sources and use tuple notation to specify:

# Request specific data source
ts_data = ds.read_ts_xrdataset(
    gage_id_lst=basin_ids[:5],
    t_range=["1990-01-01", "1995-12-31"],
    var_lst=[
        ("precipitation", "era5land"),  # Specify ERA5-Land source
        "streamflow"  # Use default source
    ]
)

API Reference

hydrodataset.camels_nz.CamelsNz

Bases: HydroDataset

CAMELS_NZ dataset class.

This class uses a custom data reading implementation to support a newer dataset version than the one supported by the underlying aquafetch library. It overrides the download URLs and provides its own parsing and caching logic.

The dataset supports both hourly ('H') and daily ('D') timesteps.

Attributes:

Name	Type	Description
region		Geographic region identifier
download		Whether to download data automatically
timestep		Time step for the data ('H' for hourly, 'D' for daily)

Source code in hydrodataset/camels_nz.py

class CamelsNz(HydroDataset):
    """CAMELS_NZ dataset class.

    This class uses a custom data reading implementation to support a newer
    dataset version than the one supported by the underlying aquafetch library.
    It overrides the download URLs and provides its own parsing and caching logic.

    The dataset supports both hourly ('H') and daily ('D') timesteps.

    Attributes:
        region: Geographic region identifier
        download: Whether to download data automatically
        timestep: Time step for the data ('H' for hourly, 'D' for daily)
    """

    # (subfolder_stem, filename_prefix, data_column, zarr_var_name)
    # subfolder_stem is joined with the timestep to form e.g.
    # "CAMELS_NZ_hourly_Streamflow" / "CAMELS_NZ_daily_Streamflow";
    # daily files are additionally prefixed with "daily_" on disk.
    _VAR_MAP_STEMS = [
        ("CAMELS_NZ_Streamflow",       "flow_station_id_",          "flow",              "q_cms_obs"),
        ("CAMELS_NZ_Precipitation",    "precipitation_station_id_", "precipitation",     "pcp_mm"),
        ("CAMELS_NZ_Temperature",      "temperature_station_id_",   "temperature",       "airtemp_c_mean"),
        ("CAMELS_NZ_PET",              "PET_station_id_",           "PET",               "pet_mm"),
        ("CAMELS_NZ_Relative_Humidity","RH_station_id_",            "Relative_humidity", "rh_"),
    ]
    _ATTR_FILES = [
        "1.CAMELS_NZ_Catchment_information.csv",
        "2.CAMELS_NZ_Climatic_attribute.csv",
        "3.CAMELS_NZ_Landcover_attribute.csv",
        "4.CAMELS_NZ_Geology.csv",
        "5.CAMELS_NZ_Anthropogenic_attribute.csv",
    ]
    _DATA_REL = "CAMELS_NZ/camels_nz"

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

        Args:
            uri: Path to the data directory
            region: Geographic region identifier (optional)
            download: Whether to download data automatically (default: False)
            timestep: Time step for the data ('H' for hourly, 'D' for daily, default: 'H')
        """
        super().__init__(uri)
        self.region = "NZ" if region is None else region
        self.download = download
        self.timestep = timestep
        # Resolve timestep-aware variable map: hourly files live under
        # "CAMELS_NZ_hourly_<Var>" with unprefixed names; daily under
        # "CAMELS_NZ_daily_<Var>" with a "daily_" filename prefix.
        freq = "hourly" if timestep == "H" else "daily"
        file_prefix = "daily_" if timestep == "D" else ""
        # Map each stream/timeseries family to its on-disk folder name
        # (Relative_Humidity must keep its full name, not "Humidity").
        folder_suffix = {
            "CAMELS_NZ_Streamflow": "Streamflow",
            "CAMELS_NZ_Precipitation": "Precipitation",
            "CAMELS_NZ_Temperature": "Temperature",
            "CAMELS_NZ_PET": "PET",
            "CAMELS_NZ_Relative_Humidity": "Relative_Humidity",
        }
        self._VAR_MAP = [
            (
                f"CAMELS_NZ_{freq}_{folder_suffix[stem]}",
                f"{file_prefix}{prefix}",
                col,
                zarr_vn,
            )
            for stem, prefix, col, zarr_vn in self._VAR_MAP_STEMS
        ]
        if not str(uri).startswith("s3://"):
            self.aqua_fetch = CAMELS_NZ(uri, timestep=timestep)

    def read_object_ids(self) -> np.ndarray:
        uri = str(self.data_source_dir).rstrip("/")
        # Streamflow entry of the timestep-aware _VAR_MAP (set in __init__)
        flow_subfolder, flow_prefix, _, _ = self._VAR_MAP[0]
        flow_rel = f"{self._DATA_REL}/{flow_subfolder}"
        if self._is_cloud():
            fs = self._make_s3fs()
            names = [p.split("/")[-1] for p in fs.ls(f"{uri}/{flow_rel}".removeprefix("s3://"))]
        else:
            names = os.listdir(os.path.join(uri, *flow_rel.split("/")))
        ids = sorted(
            n.replace(flow_prefix, "").replace(".csv", "")
            for n in names if n.startswith(flow_prefix)
        )
        return np.array(ids)

    def cache_attributes_xrdataset(self):
        """Build the local attribute cache.

        Mirrors the cloud ``cache_attributes_to_zarr`` path: read the five
        attribute CSVs, clean/rename columns, and derive ``p_mean`` from the
        precipitation timeseries (NZ has no mean-precip attribute).
        """
        if self._is_cloud():
            return super().cache_attributes_xrdataset()
        uri = str(self.data_source_dir).rstrip("/")
        attr_base = os.path.join(uri, *self._DATA_REL.split("/"), "CAMELS_NZ_Catchment_Atrributes")
        dfs = []
        for i, fname in enumerate(self._ATTR_FILES):
            path = os.path.join(attr_base, fname)
            try:
                df = pd.read_csv(path, index_col=0, dtype={0: str}, encoding="utf-8-sig")
                df.index = df.index.astype(str)
                # Every file repeats RID/StationName/latitude/longitude; keep
                # them only from the first file and drop them elsewhere.
                if i > 0:
                    df = df.drop(
                        columns=["RID", "StationName", "latitude", "longitude"],
                        errors="ignore",
                    )
                dfs.append(df)
            except Exception as e:
                print(f"  WARN {fname}: {e}")
        static = pd.concat(dfs, axis=1)
        static = static.loc[~static.index.duplicated(keep="first")]
        stations = self.read_object_ids().tolist()
        static = static.reindex(stations)
        static.columns = self._clean_feature_names(list(static.columns))
        static = static.rename(columns={"uparea": "area_km2"})
        # NZ has no mean-precip attribute; derive p_mean from the timeseries
        static["p_mean"] = self._p_mean_from_precip(static.index)

        ds_attr = static.to_xarray()
        coord_names = list(ds_attr.dims.keys())
        if len(coord_names) > 0 and coord_names[0] != "basin":
            ds_attr = ds_attr.rename({coord_names[0]: "basin"})
        units_map = self._get_attribute_units()
        ds_attr = self._assign_units_to_dataset(ds_attr, units_map)
        cache_file = self.cache_dir.joinpath(self._attributes_cache_filename)
        cache_file.parent.mkdir(parents=True, exist_ok=True)
        ds_attr.to_netcdf(cache_file)
        print(f"Attributes cache written to: {cache_file}")

    def cache_attributes_to_zarr(self) -> None:
        import zarr
        fs = self._make_s3fs()
        uri = str(self.data_source_dir).rstrip("/")
        attr_base = f"{uri}/{self._DATA_REL}/CAMELS_NZ_Catchment_Atrributes"
        dfs = []
        for i, fname in enumerate(self._ATTR_FILES):
            path = f"{attr_base}/{fname}".removeprefix("s3://")
            try:
                with fs.open(path) as fh:
                    raw = fh.read()
                df = pd.read_csv(
                    io.BytesIO(raw),
                    index_col=0, dtype={0: str}, encoding="utf-8-sig",
                )
                df.index = df.index.astype(str)
                # Every file repeats RID/StationName/latitude/longitude; AquaFetch
                # keeps them only from the first file and drops them elsewhere.
                if i > 0:
                    df = df.drop(
                        columns=["RID", "StationName", "latitude", "longitude"],
                        errors="ignore",
                    )
                dfs.append(df)
            except Exception as e:
                print(f"  WARN {fname}: {e}")
        static = pd.concat(dfs, axis=1)
        static = static.loc[~static.index.duplicated(keep="first")]
        stations = self.read_object_ids().tolist()
        static = static.reindex(stations)
        static.columns = self._clean_feature_names(list(static.columns))
        static = static.rename(columns={"uparea": "area_km2"})
        # NZ has no mean-precip attribute; derive p_mean from the timeseries
        static["p_mean"] = self._p_mean_from_precip(static.index)

        zarr_name = self._attributes_cache_filename.replace(".nc", ".zarr")
        out, opts = self._zarr_path_and_opts(zarr_name)
        n = len(stations)
        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[:] = stations
        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("/")
        base = f"{uri}/{self._DATA_REL}"
        freq = "h" if self.timestep == "H" else "D"

        stations = self.read_object_ids().tolist()
        all_times = pd.date_range(self.default_t_range[0], self.default_t_range[1], freq=freq)
        n, nt = len(stations), len(all_times)
        times_ns = all_times.asi8
        all_vars = [row[3] for row in self._VAR_MAP]

        zarr_name = self._timeseries_cache_filename.replace(".nc", ".zarr")
        out, opts = self._zarr_path_and_opts(zarr_name)
        root = zarr.open_group(out, mode="w", storage_options=opts, zarr_format=2)

        # Pre-create coordinate arrays
        time_arr = root.create_array("time", shape=(nt,), chunks=(min(nt, 8760),), 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"]

        # Pre-create all data arrays
        chunk_t = min(nt, 8760)
        for vn in all_vars:
            arr = root.create_array(vn, shape=(n, nt), chunks=(min(n, 50), chunk_t), dtype="float64")
            arr.attrs["_ARRAY_DIMENSIONS"] = ["basin", "time"]

        root.attrs["coordinates"] = "basin time"

        # Fill one variable at a time to limit memory usage
        for subfolder, prefix, col, zarr_vn in self._VAR_MAP:
            print(f"Reading {zarr_vn} ({n} stations)...")
            data = np.full((n, nt), np.nan, dtype="float64")
            ts_base = f"{base}/{subfolder}"
            for i, stn in enumerate(tqdm(stations, desc=zarr_vn)):
                path = f"{ts_base}/{prefix}{stn}.csv".removeprefix("s3://")
                try:
                    with fs.open(path) as fh:
                        df = pd.read_csv(fh, index_col="time", parse_dates=True)
                    if col in df.columns:
                        # some station files carry duplicate timestamps, which
                        # breaks reindex; keep the first occurrence
                        df = df[~df.index.duplicated(keep="first")]
                        df = df[[col]].reindex(all_times)
                        data[i] = pd.to_numeric(df[col], errors="coerce").values
                except Exception as e:
                    print(f"  WARN {stn}: {e}")
            root[zarr_vn][:] = data
            del data
            print(f"  -> written")

        self._write_zarr_units(root, "dynamic")
        print(f"Timeseries zarr written to: {out}")

    @property
    def _attributes_cache_filename(self):
        return f"camels_nz_{self.timestep.lower()}_attributes.nc"

    @property
    def _timeseries_cache_filename(self):
        return f"camels_nz_{self.timestep.lower()}_timeseries.nc"

    @property
    def default_t_range(self):
        return ["1972-01-01", "2024-08-02"]

    # Static variable definitions for CAMELS-NZ
    # Note: specific_name should be the cleaned version (lowercase, no spaces)
    # as stored in the cache file after _clean_feature_names() processing
    _subclass_static_definitions = {
        "area": {"specific_name": "area_km2", "unit": "km^2"},
        "p_mean": {"specific_name": "p_mean", "unit": "mm/day"},
    }

    # Dynamic variable mapping for CAMELS-NZ
    _dynamic_variable_mapping = {
        StandardVariable.STREAMFLOW: {
            "default_source": "obs",
            "sources": {"obs": {"specific_name": "q_cms_obs", "unit": "m^3/s"}},
        },
        StandardVariable.PRECIPITATION: {
            "default_source": "default",
            "sources": {"default": {"specific_name": "pcp_mm", "unit": "mm/day"}},
        },
        StandardVariable.TEMPERATURE_MEAN: {
            "default_source": "default",
            "sources": {"default": {"specific_name": "airtemp_c_mean", "unit": "°C"}},
        },
        StandardVariable.POTENTIAL_EVAPOTRANSPIRATION: {
            "default_source": "default",
            "sources": {"default": {"specific_name": "pet_mm", "unit": "mm/day"}},
        },
        StandardVariable.RELATIVE_HUMIDITY: {
            "default_source": "default",
            "sources": {"default": {"specific_name": "rh_", "unit": "%"}},
        },
    }

default_t_range property

__init__(uri, region=None, download=False, timestep='H')

Initialize CAMELS_NZ 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
timestep	str	Time step for the data ('H' for hourly, 'D' for daily, default: 'H')	'H'

Source code in hydrodataset/camels_nz.py

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

    Args:
        uri: Path to the data directory
        region: Geographic region identifier (optional)
        download: Whether to download data automatically (default: False)
        timestep: Time step for the data ('H' for hourly, 'D' for daily, default: 'H')
    """
    super().__init__(uri)
    self.region = "NZ" if region is None else region
    self.download = download
    self.timestep = timestep
    # Resolve timestep-aware variable map: hourly files live under
    # "CAMELS_NZ_hourly_<Var>" with unprefixed names; daily under
    # "CAMELS_NZ_daily_<Var>" with a "daily_" filename prefix.
    freq = "hourly" if timestep == "H" else "daily"
    file_prefix = "daily_" if timestep == "D" else ""
    # Map each stream/timeseries family to its on-disk folder name
    # (Relative_Humidity must keep its full name, not "Humidity").
    folder_suffix = {
        "CAMELS_NZ_Streamflow": "Streamflow",
        "CAMELS_NZ_Precipitation": "Precipitation",
        "CAMELS_NZ_Temperature": "Temperature",
        "CAMELS_NZ_PET": "PET",
        "CAMELS_NZ_Relative_Humidity": "Relative_Humidity",
    }
    self._VAR_MAP = [
        (
            f"CAMELS_NZ_{freq}_{folder_suffix[stem]}",
            f"{file_prefix}{prefix}",
            col,
            zarr_vn,
        )
        for stem, prefix, col, zarr_vn in self._VAR_MAP_STEMS
    ]
    if not str(uri).startswith("s3://"):
        self.aqua_fetch = CAMELS_NZ(uri, timestep=timestep)

read_object_ids()

Source code in hydrodataset/camels_nz.py

def read_object_ids(self) -> np.ndarray:
    uri = str(self.data_source_dir).rstrip("/")
    # Streamflow entry of the timestep-aware _VAR_MAP (set in __init__)
    flow_subfolder, flow_prefix, _, _ = self._VAR_MAP[0]
    flow_rel = f"{self._DATA_REL}/{flow_subfolder}"
    if self._is_cloud():
        fs = self._make_s3fs()
        names = [p.split("/")[-1] for p in fs.ls(f"{uri}/{flow_rel}".removeprefix("s3://"))]
    else:
        names = os.listdir(os.path.join(uri, *flow_rel.split("/")))
    ids = sorted(
        n.replace(flow_prefix, "").replace(".csv", "")
        for n in names if n.startswith(flow_prefix)
    )
    return np.array(ids)