From 777e6c25ccb17cc57fc953c053abf59f79ddb042 Mon Sep 17 00:00:00 2001 From: jpalm3r Date: Tue, 11 Aug 2026 10:40:32 +0200 Subject: [PATCH 1/3] Turn the network module into a package The MIKE+ database reader lands next and does not belong in the same file as the Network class. Move network.py to network/__init__.py unchanged so the split that follows is a pure addition. Co-Authored-By: Claude Opus 5 --- src/modelskill/{network.py => network/__init__.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename src/modelskill/{network.py => network/__init__.py} (100%) diff --git a/src/modelskill/network.py b/src/modelskill/network/__init__.py similarity index 100% rename from src/modelskill/network.py rename to src/modelskill/network/__init__.py From dee6be394e8cd1577106f93624e6a0e8f04d809f Mon Sep 17 00:00:00 2001 From: jpalm3r Date: Tue, 11 Aug 2026 10:40:59 +0200 Subject: [PATCH 2/3] Locate network observations with a MIKE+ database A network result file identifies nodes and reaches by ID, but observations are usually recorded against real-world station names held in the MIKE+ setup database. Read that sqlite database to resolve a station to the node or reach it sits on, so observations can be placed without hand-mapping every ID. Dataset variables also gain a long_name attribute, so a quantity keeps its label once it reaches xarray. Co-Authored-By: Claude Opus 5 --- src/modelskill/model/network.py | 7 + src/modelskill/network/__init__.py | 7 +- src/modelskill/network/_mikeplus.py | 275 +++++++++++++++++ src/modelskill/obs.py | 352 ++++++++++++++++++++- tests/test_mikeplus.py | 453 ++++++++++++++++++++++++++++ tests/test_network.py | 111 ++++++- 6 files changed, 1189 insertions(+), 16 deletions(-) create mode 100644 src/modelskill/network/_mikeplus.py create mode 100644 tests/test_mikeplus.py diff --git a/src/modelskill/model/network.py b/src/modelskill/model/network.py index f4d64b4f2..5c752df9b 100644 --- a/src/modelskill/model/network.py +++ b/src/modelskill/model/network.py @@ -144,6 +144,13 @@ def __init__( if quantity is None: da = self.data[sel_items.values] quantity = Quantity.from_cf_attrs(da.attrs) + if quantity == Quantity.undefined(): + # A network knows its quantity by name even when no unit + # travels with the data -- res1d and EPANET files carry the + # name only. Quantity.from_cf_attrs needs both, so fall back to + # the name alone rather than reporting nothing at all. + name = da.attrs.get("long_name") or str(sel_items.values) + quantity = Quantity(name=name, unit="") self.quantity = quantity # Mark data variables as model data diff --git a/src/modelskill/network/__init__.py b/src/modelskill/network/__init__.py index 0bad60450..a185e51ad 100644 --- a/src/modelskill/network/__init__.py +++ b/src/modelskill/network/__init__.py @@ -960,7 +960,12 @@ def to_dataset(self) -> xr.Dataset: df = df_raw.reorder_levels(["quantity", "node"], axis=1) quantities = df.columns.get_level_values("quantity").unique() return xr.Dataset( - {q: xr.DataArray(df[q], dims=["time", "node"]) for q in quantities} + { + q: xr.DataArray( + df[q], dims=["time", "node"], attrs={"long_name": str(q)} + ) + for q in quantities + } ) @property diff --git a/src/modelskill/network/_mikeplus.py b/src/modelskill/network/_mikeplus.py new file mode 100644 index 000000000..bbc341c96 --- /dev/null +++ b/src/modelskill/network/_mikeplus.py @@ -0,0 +1,275 @@ +"""Resolve dfs0 items to network locations using a MIKE+ database. + +A MIKE+ project ships a sqlite database alongside its result files. Two of its +tables describe where the measured timeseries belong in the network: + +* ``m_Measurement`` — one row per measured timeseries, naming the file + (``tsfilename``) and the item within it (``tsitemname``), plus the modelled + quantity (``resitemname``). +* ``m_Station`` — the location, as ``locationid`` plus a ``locationtype`` + saying whether that identifier names a node or a link. + +:func:`resolve_stations` joins the two and returns one row per timeseries item. +Everything MIKE+-specific — table names, the join, the ``locationtype`` codes, +the encoding of ``resitemname`` — is contained in this module. Callers see only +the columns listed in :data:`CONTRACT_COLUMNS`, so a change to the database +layout is a change to this file alone. +""" + +from __future__ import annotations + +import sqlite3 +from contextlib import contextmanager +from pathlib import Path +from typing import Iterable, Iterator, Literal, Sequence + +import pandas as pd + +Kind = Literal["node", "reach"] + +#: Columns returned by :func:`resolve_stations`, one row per timeseries item. +#: +#: ``item_name`` name of the item in the data source +#: ``name`` display name for the observation +#: ``location`` node alias, or ``(reach_id, distance)`` for a breakpoint +#: ``kind`` ``"node"`` or ``"reach"``, i.e. which observation class fits +#: ``quantity`` modelled quantity name +CONTRACT_COLUMNS = ["item_name", "name", "location", "kind", "quantity"] + +# MIKE+ m_Station.locationtype codes. 8 is a junction and 12 a tank or +# reservoir; both are graph nodes. 9 is a link, which becomes a breakpoint when +# the station carries a chainage and a whole reach when it does not. Unknown +# codes raise rather than guess. +_FAMILY_BY_LOCATIONTYPE: dict[int, str] = {8: "node", 9: "link", 12: "node"} + +_REQUIRED_COLUMNS: dict[str, set[str]] = { + "m_Station": {"muid", "locationid", "locationtype", "chainagevalue", "assetname"}, + "m_Measurement": { + "measurementstationid", + "tsfilename", + "tsitemname", + "resitemname", + }, +} + +_JOIN_QUERY = """ + SELECT m.tsitemname AS item_name, + m.tsfilename AS tsfilename, + m.resitemname AS resitemname, + s.assetname AS assetname, + s.locationid AS locationid, + s.locationtype AS locationtype, + s.chainagevalue AS chainagevalue + FROM m_Measurement m + JOIN m_Station s ON s.muid = m.measurementstationid +""" + + +@contextmanager +def _connect(db: str | Path | sqlite3.Connection) -> Iterator[sqlite3.Connection]: + if isinstance(db, sqlite3.Connection): + yield db + else: + conn = sqlite3.connect(str(db)) + try: + yield conn + finally: + conn.close() + + +def _validate_schema(conn: sqlite3.Connection) -> None: + tables = { + row[0] + for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'") + } + if missing := sorted(set(_REQUIRED_COLUMNS) - tables): + raise ValueError( + f"Database is missing table(s) {missing}. " + "A MIKE+ database with 'm_Station' and 'm_Measurement' is required." + ) + for table, required in _REQUIRED_COLUMNS.items(): + columns = {row[1] for row in conn.execute(f"PRAGMA table_info([{table}])")} + if missing_cols := sorted(required - columns): + raise ValueError( + f"Table '{table}' is missing column(s) {missing_cols}. " + "The database layout is not the one modelskill expects." + ) + + +def _read_join(conn: sqlite3.Connection, source: str | None) -> pd.DataFrame: + query = _JOIN_QUERY + params: list[str] = [] + if source is not None: + query += " WHERE m.tsfilename LIKE ?" + params.append(f"%{Path(source).name}%") + return pd.read_sql_query(query, conn, params=params) + + +def _kind_and_location(row: pd.Series) -> tuple[Kind, str | tuple[str, float]]: + # A link station with a chainage names a point along a reach, which is a + # node observation at a breakpoint. Without a chainage it names the reach + # as a whole. + location_type = row["locationtype"] + try: + family = _FAMILY_BY_LOCATIONTYPE[int(location_type)] + except (KeyError, TypeError, ValueError): + raise ValueError( + f"Station '{row['locationid']}' has unsupported locationtype " + f"{location_type!r}. Known codes are {sorted(_FAMILY_BY_LOCATIONTYPE)}." + ) from None + + chainage = row["chainagevalue"] + if family == "node": + return "node", str(row["locationid"]) + if pd.isna(chainage): + return "reach", str(row["locationid"]) + return "node", (str(row["locationid"]), float(chainage)) + + +def _describe_unresolved( + conn: sqlite3.Connection, missing: Sequence[str], source: str | None +) -> str: + assets = set( + pd.read_sql_query("SELECT assetname FROM m_Station", conn)["assetname"] + .dropna() + .tolist() + ) + known = [item for item in missing if item in assets] + unknown = [item for item in missing if item not in assets] + + lines = [] + if known: + where = f" for '{Path(source).name}'" if source else "" + lines.append( + f" Known station, no measurement registered{where} ({len(known)}):\n" + + "\n".join(f" {item}" for item in known) + ) + if unknown: + lines.append( + f" Not found in the database ({len(unknown)}):\n" + + "\n".join(f" {item}" for item in unknown) + ) + return "\n".join(lines) + + +def _choose_names(selection: pd.DataFrame) -> pd.Series: + # assetname is far shorter than the raw item name and is normally unique, + # but it is only safe as a display name when it distinguishes every row. + assets = selection["assetname"] + if assets.notna().all() and assets.nunique() == len(selection): + return assets.astype(str) + return selection["item_name"].astype(str) + + +def resolve_stations( + db: str | Path | sqlite3.Connection, + *, + item_names: Iterable[str], + source: str | None = None, + quantity: str | None = None, + kind: Kind | None = None, + on_missing: Literal["raise", "skip"] = "raise", +) -> pd.DataFrame: + """Resolve data source items to network locations via a MIKE+ database. + + Parameters + ---------- + db : str, Path or sqlite3.Connection + MIKE+ database, as a path or an already-open connection. + item_names : Iterable[str] + Item names to resolve, e.g. the column names of a dfs0. + source : str, optional + File the items come from, matched against ``tsfilename``. Only the file + name is used, so a full path is fine. By default None, in which case + measurements from every file are considered and an item registered + against more than one file raises. + quantity : str, optional + Quantity to select, e.g. ``"Pressure"``. By default None, in which case + the quantity is inferred when the selection holds only one and raises + when it holds several. + kind : {"node", "reach"}, optional + Restrict to locations of this kind, by default None (no restriction). + on_missing : {"raise", "skip"}, optional + What to do with items that resolve to no measurement at all, by default + "raise". + + Returns + ------- + pd.DataFrame + One row per resolved item, with the columns in + :data:`CONTRACT_COLUMNS`. + + Raises + ------ + ValueError + If the database layout is not the expected one, if items cannot be + resolved and ``on_missing="raise"``, if the quantity is ambiguous or + absent, or if no location of the requested kind remains. + """ + requested = list(dict.fromkeys(item_names)) + + with _connect(db) as conn: + _validate_schema(conn) + rows = _read_join(conn, source) + rows["quantity"] = rows["resitemname"].str.split(";").str[0].str.strip() + rows = rows[rows["item_name"].isin(requested)].copy() + + missing = [item for item in requested if item not in set(rows["item_name"])] + if missing and on_missing == "raise": + raise ValueError( + f"{len(missing)} of {len(requested)} items could not be resolved " + f"against the MIKE+ database.\n" + + _describe_unresolved(conn, missing, source) + + '\n Pass on_missing="skip" to ignore these.' + ) + + if ambiguous := sorted( + rows.loc[rows.duplicated("item_name", keep=False), "item_name"].unique() + ): + raise ValueError( + f"Item(s) {ambiguous} are registered against more than one file. " + "Pass 'source' to say which file the data comes from." + ) + + if rows.empty: + raise ValueError("No items could be resolved against the MIKE+ database.") + + resolved = rows.apply(_kind_and_location, axis=1) + rows["kind"] = [k for k, _ in resolved] + rows["location"] = [location for _, location in resolved] + + if quantity is None: + pool = rows if kind is None else rows[rows["kind"] == kind] + available = sorted(pool["quantity"].unique()) + if len(available) == 0: + raise ValueError( + f"No {kind} locations found. Quantities present: " + f"{rows['quantity'].value_counts().to_dict()}." + ) + if len(available) > 1: + raise ValueError( + "Several quantities present, so 'quantity' cannot be inferred: " + f"{pool['quantity'].value_counts().to_dict()}. Pass one of {available}." + ) + quantity = available[0] + + selection = rows[rows["quantity"] == quantity] + if selection.empty: + raise ValueError( + f"Quantity {quantity!r} not found. Available: " + f"{rows['quantity'].value_counts().to_dict()}." + ) + + if kind is not None: + of_kind = selection[selection["kind"] == kind] + if of_kind.empty: + other = sorted(selection["kind"].unique()) + raise ValueError( + f"All {len(selection)} {quantity!r} station(s) are of kind " + f"{other}, not {kind!r}." + ) + selection = of_kind + + selection = selection.copy() + selection["name"] = _choose_names(selection) + return selection[CONTRACT_COLUMNS].reset_index(drop=True) diff --git a/src/modelskill/obs.py b/src/modelskill/obs.py index e06b79858..143b788f9 100644 --- a/src/modelskill/obs.py +++ b/src/modelskill/obs.py @@ -1,18 +1,20 @@ """ # Observations -ModelSkill supports four types of observations: +ModelSkill supports five types of observations: * [`PointObservation`](`modelskill.PointObservation`) - a point timeseries from a dfs0/nc file or a DataFrame * [`TrackObservation`](`modelskill.TrackObservation`) - a track (moving point) timeseries from a dfs0/nc file or a DataFrame * [`VerticalObservation`](`modelskill.VerticalObservation`) - a vertical profile from a dfs0/nc file or a DataFrame * [`NodeObservation`](`modelskill.NodeObservation`) - a network node timeseries for specific node IDs. +* [`ReachObservation`](`modelskill.ReachObservation`) - a network reach timeseries for a quantity uniform along the reach. An observation can be created by explicitly invoking one of the above classes or using the [`observation()`](`modelskill.observation`) function which will return the appropriate type based on the input data (if possible). """ from __future__ import annotations +from pathlib import Path from typing import Literal, Any, Union, overload from typing_extensions import Self import warnings @@ -34,6 +36,10 @@ # NetCDF attributes can only be str, int, float https://unidata.github.io/netcdf4-python/#attributes-in-a-netcdf-file Serializable = Union[str, int, float] +# Where a node observation sits: an internal network ID, an original node alias, +# or a breakpoint given as (reach_id, distance) along a reach. +NodeLocation = Union[int, str, tuple[str, float]] + def observation( data: DataInputType, @@ -113,6 +119,79 @@ def _guess_gtype(**kwargs) -> GeometryType: return GeometryType.POINT +def _item_names(data: Any) -> list[str]: + """Names of the individual timeseries held by an already-opened data source.""" + if isinstance(data, pd.DataFrame): + return [str(c) for c in data.columns] + if isinstance(data, xr.Dataset): + return [str(v) for v in data.data_vars] + if hasattr(data, "names"): # mikeio.Dataset + return [str(n) for n in data.names] + if hasattr(data, "name"): # pd.Series, mikeio.DataArray, xr.DataArray + return [str(data.name)] + raise ValueError( + f"Cannot determine item names from data of type {type(data).__name__}" + ) + + +def _observations_from_mikeplus( + cls: type, + *, + data: PointType, + db: Any, + kind: Literal["node", "reach"], + location_arg: str, + quantity: Quantity | str | None, + source: str | None, + on_missing: Literal["raise", "skip"], + aux_items: list[int | str] | None, + attrs: dict | None, +) -> list[Any]: + """Build observations from a data source and a MIKE+ database.""" + from .network._mikeplus import resolve_stations + from .timeseries._point import _open_and_name + + if source is None and isinstance(data, (str, Path)): + source = str(data) + + # Open once rather than per observation; a path would otherwise be re-read + # for every station in the database. + opened, _ = _open_and_name(data, None) + + given_quantity = quantity if isinstance(quantity, Quantity) else None + wanted = quantity.name if isinstance(quantity, Quantity) else quantity + + stations = resolve_stations( + db, + item_names=_item_names(opened), + source=source, + quantity=wanted, + kind=kind, + on_missing=on_missing, + ) + + observations = [] + for station in stations.itertuples(): + obs = cls( + opened, + item=station.item_name, + name=station.name, + quantity=given_quantity, + aux_items=aux_items, + attrs=attrs, + **{location_arg: station.location}, + ) + if given_quantity is None: + # The database names the quantity; the data source knows its unit. + obs.quantity = Quantity( + name=station.quantity, + unit=obs.quantity.unit, + is_directional=obs.quantity.is_directional, + ) + observations.append(obs) + return observations + + def _validate_attrs(data_attrs: dict, attrs: dict | None) -> None: # See similar method in xarray https://github.com/pydata/xarray/blob/main/xarray/backends/api.py#L165 @@ -595,7 +674,7 @@ def from_multiple( cls, *, data: PointType, - nodes: dict[int, str | int], + nodes: dict[NodeLocation, str | int], quantity: Quantity | None = None, aux_items: list[int | str] | None = None, attrs: dict | None = None, @@ -606,20 +685,37 @@ def from_multiple( def from_multiple( cls, *, - nodes: dict[int, PointType], + nodes: dict[NodeLocation, PointType], quantity: Quantity | None = None, aux_items: list[int | str] | None = None, attrs: dict | None = None, ) -> list[NodeObservation]: pass + @overload + @classmethod + def from_multiple( + cls, + *, + data: PointType, + db: str | Path | Any, + quantity: Quantity | str | None = None, + source: str | None = None, + on_missing: Literal["raise", "skip"] = "raise", + aux_items: list[int | str] | None = None, + attrs: dict | None = None, + ) -> list[NodeObservation]: ... + @classmethod def from_multiple( cls, *, data: PointType | None = None, - nodes: dict[int, Any] | None = None, - quantity: Quantity | None = None, + nodes: dict[NodeLocation, Any] | None = None, + db: str | Path | Any | None = None, + quantity: Quantity | str | None = None, + source: str | None = None, + on_missing: Literal["raise", "skip"] = "raise", aux_items: list[int | str] | None = None, attrs: dict | None = None, ) -> list[NodeObservation]: @@ -638,14 +734,41 @@ def from_multiple( obs = NodeObservation.from_multiple(data=df, nodes={123: "col_a", 456: "col_b"}) + 3. **MIKE+ database** — pass a single ``data`` object together with + ``db``, and the locations are looked up in the database:: + + obs = NodeObservation.from_multiple(data="calib.dfs0", db="model.sqlite") + + One observation is created per item of ``data`` that the database + places on a node, so several sensors at the same node are all kept. + Parameters ---------- data : PointType, optional - Shared data source (required when ``nodes`` values are column selectors). - nodes : dict[int, PointType | str | int] - Mapping of node_id -> data source or column selector. - quantity : Quantity | None, optional - Physical quantity metadata, by default None. + Shared data source (required when ``nodes`` values are column + selectors, and when ``db`` is given). + nodes : dict[int | str | tuple[str, float], PointType | str | int] + Mapping of location -> data source or column selector. A location + takes any of the forms accepted by ``at``: an internal network ID, + a node alias, or a ``(reach_id, distance)`` breakpoint. + + Note that a location can appear only once, so this form cannot + express several observations at the same node. Use ``db`` when the + data has several sensors at one location. + db : str, Path or sqlite3.Connection, optional + MIKE+ database locating the items of ``data`` in the network. + Mutually exclusive with ``nodes``. + quantity : Quantity or str, optional + Physical quantity metadata, by default None. With ``db``, a string + selects which quantity to build observations for and the metadata + comes from the database; omit it and the quantity is inferred when + the data holds only one. + source : str, optional + With ``db``, the file the items come from. Taken from ``data`` when + that is a path, by default None. + on_missing : {"raise", "skip"}, optional + With ``db``, what to do with items the database cannot place, by + default "raise". aux_items : list[int | str] | None, optional Auxiliary items, by default None. attrs : dict | None, optional @@ -655,7 +778,39 @@ def from_multiple( ------- list[NodeObservation] List of NodeObservation objects. + + Raises + ------ + ValueError + If both ``nodes`` and ``db`` are given, if neither is, or if the + database cannot resolve the requested items. """ + if db is not None: + if nodes is not None: + raise ValueError( + "'nodes' and 'db' are mutually exclusive: the database " + "supplies the locations." + ) + if data is None: + raise ValueError("'data' is required when 'db' is given") + return _observations_from_mikeplus( + cls, + data=data, + db=db, + kind="node", + location_arg="at", + quantity=quantity, + source=source, + on_missing=on_missing, + aux_items=aux_items, + attrs=attrs, + ) + + if isinstance(quantity, str): + raise TypeError( + "'quantity' must be a Quantity unless 'db' is given, got str" + ) + if nodes is None: raise ValueError("'nodes' argument is required") if not isinstance(nodes, dict): @@ -764,6 +919,183 @@ def _create_new_instance(self, data: xr.Dataset) -> Self: """Reconstruct instance from a dataset slice.""" return self.__class__(data, reach=str(data.coords["reach"].item())) + @overload + @classmethod + def from_multiple( + cls, + *, + data: PointType, + reaches: dict[str, str | int], + quantity: Quantity | None = None, + aux_items: list[int | str] | None = None, + attrs: dict | None = None, + ) -> list[ReachObservation]: ... + + @overload + @classmethod + def from_multiple( + cls, + *, + reaches: dict[str, PointType], + quantity: Quantity | None = None, + aux_items: list[int | str] | None = None, + attrs: dict | None = None, + ) -> list[ReachObservation]: + pass + + @overload + @classmethod + def from_multiple( + cls, + *, + data: PointType, + db: str | Path | Any, + quantity: Quantity | str | None = None, + source: str | None = None, + on_missing: Literal["raise", "skip"] = "raise", + aux_items: list[int | str] | None = None, + attrs: dict | None = None, + ) -> list[ReachObservation]: ... + + @classmethod + def from_multiple( + cls, + *, + data: PointType | None = None, + reaches: dict[str, Any] | None = None, + db: str | Path | Any | None = None, + quantity: Quantity | str | None = None, + source: str | None = None, + on_missing: Literal["raise", "skip"] = "raise", + aux_items: list[int | str] | None = None, + attrs: dict | None = None, + ) -> list[ReachObservation]: + """Create multiple ReachObservation objects. + + Two calling conventions are supported: + + 1. **Separate data sources** — pass only ``reaches`` as a dict mapping + each reach ID to its own data source (file path, DataFrame, etc.):: + + obs = ReachObservation.from_multiple(reaches={"r1": df1, "r2": "sensor.csv"}) + + 2. **Shared data source** — pass a single ``data`` object together with + ``reaches`` as a dict mapping each reach ID to the column name or + index to select from ``data``:: + + obs = ReachObservation.from_multiple(data=df, reaches={"r1": "col_a", "r2": "col_b"}) + + 3. **MIKE+ database** — pass a single ``data`` object together with + ``db``, and the reaches are looked up in the database:: + + obs = ReachObservation.from_multiple(data="calib.dfs0", db="model.sqlite") + + One observation is created per item of ``data`` that the database + places on a link without a chainage. + + Parameters + ---------- + data : PointType, optional + Shared data source (required when ``reaches`` values are column + selectors, and when ``db`` is given). + reaches : dict[str, PointType | str | int] + Mapping of reach_id -> data source or column selector. + + Note that a reach can appear only once, so this form cannot express + several observations on the same reach. Use ``db`` when the data has + several sensors on one reach. + db : str, Path or sqlite3.Connection, optional + MIKE+ database locating the items of ``data`` in the network. + Mutually exclusive with ``reaches``. + quantity : Quantity or str, optional + Physical quantity metadata, by default None. With ``db``, a string + selects which quantity to build observations for and the metadata + comes from the database; omit it and the quantity is inferred when + the data holds only one. + source : str, optional + With ``db``, the file the items come from. Taken from ``data`` when + that is a path, by default None. + on_missing : {"raise", "skip"}, optional + With ``db``, what to do with items the database cannot place, by + default "raise". + aux_items : list[int | str] | None, optional + Auxiliary items, by default None. + attrs : dict | None, optional + Additional attributes, by default None. + + Returns + ------- + list[ReachObservation] + List of ReachObservation objects. + + Raises + ------ + ValueError + If both ``reaches`` and ``db`` are given, if neither is, or if the + database cannot resolve the requested items. + """ + if db is not None: + if reaches is not None: + raise ValueError( + "'reaches' and 'db' are mutually exclusive: the database " + "supplies the locations." + ) + if data is None: + raise ValueError("'data' is required when 'db' is given") + return _observations_from_mikeplus( + cls, + data=data, + db=db, + kind="reach", + location_arg="reach", + quantity=quantity, + source=source, + on_missing=on_missing, + aux_items=aux_items, + attrs=attrs, + ) + + if isinstance(quantity, str): + raise TypeError( + "'quantity' must be a Quantity unless 'db' is given, got str" + ) + + if reaches is None: + raise ValueError("'reaches' argument is required") + if not isinstance(reaches, dict): + raise TypeError( + f"'reaches' must be a dict mapping reach_id -> data_source, got {type(reaches).__name__}" + ) + + reach_ids = list(reaches.keys()) + + if data is None: + data_sources: list[PointType] = list(reaches.values()) + return [ + cls( + data_i, + reach=reach_i, + item=None, + quantity=quantity, + aux_items=aux_items, + attrs=attrs, + ) + for data_i, reach_i in zip(data_sources, reach_ids) + ] + else: + reach_items: list[int | str | None] = list(reaches.values()) + return [ + cls( + data, + reach=reach_i, + item=item_i, + quantity=quantity, + aux_items=aux_items, + attrs=attrs, + ) + for reach_i, item_i in zip(reach_ids, reach_items) + ] + def unit_display_name(name: str) -> str: """Display name diff --git a/tests/test_mikeplus.py b/tests/test_mikeplus.py new file mode 100644 index 000000000..e04f26076 --- /dev/null +++ b/tests/test_mikeplus.py @@ -0,0 +1,453 @@ +# ruff: noqa: E402 +import sqlite3 + +import numpy as np +import pandas as pd +import pytest + +pytest.importorskip("networkx") + +from modelskill.network._mikeplus import CONTRACT_COLUMNS, resolve_stations +from modelskill.obs import NodeObservation, ReachObservation + +STATION_COLUMNS = [ + "muid", + "locationid", + "locationtype", + "chainagevalue", + "assetname", +] +MEASUREMENT_COLUMNS = [ + "measurementstationid", + "tsfilename", + "tsitemname", + "resitemname", +] + +JUNCTION = 8 +LINK = 9 +TANK = 12 + + +def station(muid, locationid, locationtype, assetname, chainagevalue=None): + return dict( + muid=muid, + locationid=locationid, + locationtype=locationtype, + chainagevalue=chainagevalue, + assetname=assetname, + ) + + +def measurement(station_muid, item, quantity, file="calib.dfs0"): + return dict( + measurementstationid=station_muid, + tsfilename=rf"..\Scripts\{file}", + tsitemname=item, + resitemname=f"{quantity};{quantity};100450", + ) + + +def build_db(path, stations, measurements, *, station_columns=STATION_COLUMNS): + conn = sqlite3.connect(str(path)) + pd.DataFrame(stations, columns=station_columns).to_sql( + "m_Station", conn, index=False + ) + pd.DataFrame(measurements, columns=MEASUREMENT_COLUMNS).to_sql( + "m_Measurement", conn, index=False + ) + conn.commit() + conn.close() + return str(path) + + +@pytest.fixture +def db(tmp_path): + """Two pressure sensors on nodes, one flow meter on a link.""" + stations = [ + station("s1", "wNode_1", JUNCTION, "PT.401"), + station("s2", "Tank_A", TANK, "LT.410"), + station("s3", "Pipe_7", LINK, "FT.403"), + ] + measurements = [ + measurement("s1", "item_pressure_1", "Pressure"), + measurement("s2", "item_pressure_2", "Pressure"), + measurement("s3", "item_flow_1", "Flow"), + ] + return build_db(tmp_path / "mikeplus.sqlite", stations, measurements) + + +def test_returns_contract_columns(db): + df = resolve_stations(db, item_names=["item_pressure_1"], quantity="Pressure") + + assert list(df.columns) == CONTRACT_COLUMNS + assert len(df) == 1 + + +def test_junction_and_tank_are_nodes(db): + df = resolve_stations( + db, item_names=["item_pressure_1", "item_pressure_2"], quantity="Pressure" + ) + + assert set(df["kind"]) == {"node"} + assert set(df["location"]) == {"wNode_1", "Tank_A"} + + +def test_link_without_chainage_is_a_reach(db): + df = resolve_stations(db, item_names=["item_flow_1"], quantity="Flow") + + assert df["kind"].tolist() == ["reach"] + assert df["location"].tolist() == ["Pipe_7"] + + +def test_link_with_chainage_is_a_node_at_a_breakpoint(tmp_path): + path = build_db( + tmp_path / "chainage.sqlite", + [station("s1", "Pipe_7", LINK, "FT.403", chainagevalue=24.5)], + [measurement("s1", "item_flow_1", "Flow")], + ) + + df = resolve_stations(path, item_names=["item_flow_1"], quantity="Flow") + + assert df["kind"].tolist() == ["node"] + assert df["location"].tolist() == [("Pipe_7", 24.5)] + + +def test_several_items_at_one_location_all_survive(tmp_path): + path = build_db( + tmp_path / "shared.sqlite", + [ + station("s1", "wNode_1", JUNCTION, "PT.401"), + station("s2", "wNode_1", JUNCTION, "PT.402"), + ], + [ + measurement("s1", "before_valve", "Pressure"), + measurement("s2", "after_valve", "Pressure"), + ], + ) + + df = resolve_stations( + path, item_names=["before_valve", "after_valve"], quantity="Pressure" + ) + + assert len(df) == 2 + assert df["location"].tolist() == ["wNode_1", "wNode_1"] + assert df["name"].tolist() == ["PT.401", "PT.402"] + + +def test_name_falls_back_to_item_name_when_assetnames_collide(tmp_path): + path = build_db( + tmp_path / "collide.sqlite", + [ + station("s1", "wNode_1", JUNCTION, "same"), + station("s2", "wNode_2", JUNCTION, "same"), + ], + [ + measurement("s1", "item_a", "Pressure"), + measurement("s2", "item_b", "Pressure"), + ], + ) + + df = resolve_stations(path, item_names=["item_a", "item_b"], quantity="Pressure") + + assert df["name"].tolist() == ["item_a", "item_b"] + + +def test_quantity_is_inferred_when_unambiguous(db): + df = resolve_stations(db, item_names=["item_pressure_1", "item_pressure_2"]) + + assert df["quantity"].tolist() == ["Pressure", "Pressure"] + + +def test_ambiguous_quantity_raises_and_lists_options(db): + with pytest.raises(ValueError, match="cannot be inferred") as excinfo: + resolve_stations(db, item_names=["item_pressure_1", "item_flow_1"]) + + assert "Pressure" in str(excinfo.value) + assert "Flow" in str(excinfo.value) + + +def test_kind_narrows_the_pool_used_for_inference(db): + df = resolve_stations( + db, + item_names=["item_pressure_1", "item_pressure_2", "item_flow_1"], + kind="node", + ) + + assert df["quantity"].tolist() == ["Pressure", "Pressure"] + + +def test_unknown_quantity_raises(db): + with pytest.raises(ValueError, match="not found"): + resolve_stations(db, item_names=["item_pressure_1"], quantity="Discharge") + + +def test_wrong_kind_raises_naming_the_actual_kind(db): + with pytest.raises(ValueError, match="not 'node'"): + resolve_stations(db, item_names=["item_flow_1"], quantity="Flow", kind="node") + + +def test_missing_items_raise_and_separate_the_two_causes(db): + with pytest.raises(ValueError) as excinfo: + resolve_stations( + db, + item_names=["item_pressure_1", "PT.401", "never_heard_of_it"], + quantity="Pressure", + ) + + message = str(excinfo.value) + assert "no measurement registered" in message + assert "PT.401" in message + assert "Not found in the database" in message + assert "never_heard_of_it" in message + + +def test_missing_items_can_be_skipped(db): + df = resolve_stations( + db, + item_names=["item_pressure_1", "never_heard_of_it"], + quantity="Pressure", + on_missing="skip", + ) + + assert df["item_name"].tolist() == ["item_pressure_1"] + + +def test_source_selects_between_files(tmp_path): + path = build_db( + tmp_path / "files.sqlite", + [station("s1", "wNode_1", JUNCTION, "PT.401")], + [ + measurement("s1", "item_a", "Pressure", file="main.dfs0"), + measurement("s1", "item_a", "Pressure", file="other.dfs0"), + ], + ) + + df = resolve_stations( + path, item_names=["item_a"], quantity="Pressure", source="main.dfs0" + ) + + assert len(df) == 1 + + +def test_source_accepts_a_full_path(tmp_path): + path = build_db( + tmp_path / "fullpath.sqlite", + [station("s1", "wNode_1", JUNCTION, "PT.401")], + [measurement("s1", "item_a", "Pressure", file="main.dfs0")], + ) + + df = resolve_stations( + path, + item_names=["item_a"], + quantity="Pressure", + source="/some/where/main.dfs0", + ) + + assert len(df) == 1 + + +def test_item_registered_against_several_files_raises_without_source(tmp_path): + path = build_db( + tmp_path / "ambiguous.sqlite", + [station("s1", "wNode_1", JUNCTION, "PT.401")], + [ + measurement("s1", "item_a", "Pressure", file="main.dfs0"), + measurement("s1", "item_a", "Pressure", file="other.dfs0"), + ], + ) + + with pytest.raises(ValueError, match="more than one file"): + resolve_stations(path, item_names=["item_a"], quantity="Pressure") + + +def test_unknown_locationtype_raises(tmp_path): + path = build_db( + tmp_path / "weird.sqlite", + [station("s1", "wNode_1", 99, "PT.401")], + [measurement("s1", "item_a", "Pressure")], + ) + + with pytest.raises(ValueError, match="unsupported locationtype"): + resolve_stations(path, item_names=["item_a"], quantity="Pressure") + + +def test_accepts_an_open_connection(db): + conn = sqlite3.connect(db) + try: + df = resolve_stations(conn, item_names=["item_flow_1"], quantity="Flow") + finally: + conn.close() + + assert len(df) == 1 + + +def test_missing_table_raises(tmp_path): + path = str(tmp_path / "empty.sqlite") + conn = sqlite3.connect(path) + pd.DataFrame({"a": [1]}).to_sql("something_else", conn, index=False) + conn.close() + + with pytest.raises(ValueError, match="missing table"): + resolve_stations(path, item_names=["item_a"]) + + +def test_missing_column_raises(tmp_path): + path = build_db( + tmp_path / "thin.sqlite", + [ + dict(muid="s1", locationid="wNode_1", locationtype=JUNCTION, assetname="a"), + ], + [measurement("s1", "item_a", "Pressure")], + station_columns=["muid", "locationid", "locationtype", "assetname"], + ) + + with pytest.raises(ValueError, match="missing column"): + resolve_stations(path, item_names=["item_a"]) + + +@pytest.fixture +def calibration_data(): + """A data source holding both pressure and flow items.""" + time = pd.date_range("2024-01-01", periods=24, freq="h") + rng = np.random.default_rng(42) + return pd.DataFrame( + { + "item_pressure_1": rng.normal(35.0, 1.0, len(time)), + "item_pressure_2": rng.normal(36.0, 1.0, len(time)), + "item_flow_1": rng.normal(120.0, 5.0, len(time)), + }, + index=time, + ) + + +class TestNodeObservationFromDatabase: + def test_builds_one_observation_per_item(self, db, calibration_data): + obs_list = NodeObservation.from_multiple( + data=calibration_data, db=db, quantity="Pressure" + ) + + assert len(obs_list) == 2 + assert all(isinstance(obs, NodeObservation) for obs in obs_list) + assert [obs.at for obs in obs_list] == ["wNode_1", "Tank_A"] + + def test_names_come_from_the_database(self, db, calibration_data): + obs_list = NodeObservation.from_multiple( + data=calibration_data, db=db, quantity="Pressure" + ) + + assert [obs.name for obs in obs_list] == ["PT.401", "LT.410"] + + def test_quantity_comes_from_the_database(self, db, calibration_data): + obs_list = NodeObservation.from_multiple( + data=calibration_data, db=db, quantity="Pressure" + ) + + assert all(obs.quantity.name == "Pressure" for obs in obs_list) + + def test_data_is_selected_per_item(self, db, calibration_data): + obs_list = NodeObservation.from_multiple( + data=calibration_data, db=db, quantity="Pressure" + ) + + expected = calibration_data["item_pressure_1"].to_numpy() + assert obs_list[0].values == pytest.approx(expected) + + def test_quantity_is_inferred_when_only_nodes_are_wanted( + self, db, calibration_data + ): + obs_list = NodeObservation.from_multiple(data=calibration_data, db=db) + + assert len(obs_list) == 2 + assert all(obs.quantity.name == "Pressure" for obs in obs_list) + + def test_several_sensors_at_one_node_are_all_kept(self, tmp_path): + path = build_db( + tmp_path / "shared.sqlite", + [ + station("s1", "wNode_1", JUNCTION, "PT.401"), + station("s2", "wNode_1", JUNCTION, "PT.402"), + ], + [ + measurement("s1", "before_valve", "Pressure"), + measurement("s2", "after_valve", "Pressure"), + ], + ) + time = pd.date_range("2024-01-01", periods=5, freq="h") + data = pd.DataFrame( + {"before_valve": range(5), "after_valve": range(5, 10)}, index=time + ) + + obs_list = NodeObservation.from_multiple(data=data, db=path) + + assert [obs.at for obs in obs_list] == ["wNode_1", "wNode_1"] + assert [obs.name for obs in obs_list] == ["PT.401", "PT.402"] + + def test_flow_on_a_link_points_at_reach_observation(self, db, calibration_data): + with pytest.raises(ValueError, match="not 'node'"): + NodeObservation.from_multiple(data=calibration_data, db=db, quantity="Flow") + + def test_unresolvable_item_raises(self, db, calibration_data): + data = calibration_data.rename(columns={"item_pressure_1": "mystery_sensor"}) + + with pytest.raises(ValueError, match="could not be resolved"): + NodeObservation.from_multiple(data=data, db=db, quantity="Pressure") + + def test_unresolvable_item_can_be_skipped(self, db, calibration_data): + data = calibration_data.rename(columns={"item_pressure_1": "mystery_sensor"}) + + obs_list = NodeObservation.from_multiple( + data=data, db=db, quantity="Pressure", on_missing="skip" + ) + + assert [obs.name for obs in obs_list] == ["LT.410"] + + def test_db_and_nodes_are_mutually_exclusive(self, db, calibration_data): + with pytest.raises(ValueError, match="mutually exclusive"): + NodeObservation.from_multiple( + data=calibration_data, db=db, nodes={1: "item_pressure_1"} + ) + + def test_db_without_data_raises(self, db): + with pytest.raises(ValueError, match="'data' is required"): + NodeObservation.from_multiple(db=db) + + def test_quantity_string_without_db_raises(self, calibration_data): + with pytest.raises(TypeError, match="must be a Quantity"): + NodeObservation.from_multiple( + data=calibration_data, + nodes={1: "item_pressure_1"}, + quantity="Pressure", + ) + + +class TestReachObservationFromDatabase: + def test_builds_reach_observations(self, db, calibration_data): + obs_list = ReachObservation.from_multiple( + data=calibration_data, db=db, quantity="Flow" + ) + + assert len(obs_list) == 1 + assert isinstance(obs_list[0], ReachObservation) + assert obs_list[0].reach == "Pipe_7" + assert obs_list[0].name == "FT.403" + assert obs_list[0].quantity.name == "Flow" + + def test_quantity_is_inferred_when_only_reaches_are_wanted( + self, db, calibration_data + ): + obs_list = ReachObservation.from_multiple(data=calibration_data, db=db) + + assert [obs.reach for obs in obs_list] == ["Pipe_7"] + + def test_pressure_on_a_node_points_at_node_observation(self, db, calibration_data): + with pytest.raises(ValueError, match="not 'reach'"): + ReachObservation.from_multiple( + data=calibration_data, db=db, quantity="Pressure" + ) + + def test_db_and_reaches_are_mutually_exclusive(self, db, calibration_data): + with pytest.raises(ValueError, match="mutually exclusive"): + ReachObservation.from_multiple( + data=calibration_data, db=db, reaches={"r1": "item_flow_1"} + ) diff --git a/tests/test_network.py b/tests/test_network.py index 2d115b022..4e98a9c56 100644 --- a/tests/test_network.py +++ b/tests/test_network.py @@ -32,7 +32,7 @@ _MIKE_EXTENSIONS, _UNSUPPORTED_EXTENSIONS, ) -from modelskill.obs import NodeObservation +from modelskill.obs import NodeObservation, ReachObservation from modelskill.quantity import Quantity @@ -155,6 +155,36 @@ def test_init_with_network(self, sample_network): assert isinstance(nmr.time, pd.DatetimeIndex) assert len(nmr.nodes) == 3 + def test_quantity_name_survives_to_the_model_result(self, sample_network): + """The network knows its quantity by name even without a unit.""" + nmr = NetworkModelResult(sample_network) + + assert nmr.quantity.name == "WaterLevel" + assert nmr.quantity != Quantity.undefined() + + def test_quantity_carries_into_extracted_node(self, sample_network): + nmr = NetworkModelResult(sample_network) + obs_data = pd.DataFrame({"sensor": np.zeros(len(nmr.time))}, index=nmr.time) + extracted = nmr.extract(NodeObservation(obs_data, at="123")) + + assert extracted.quantity.name == "WaterLevel" + + def test_explicit_quantity_wins(self, sample_network): + given = Quantity(name="Water Level", unit="meter") + nmr = NetworkModelResult(sample_network, quantity=given) + + assert nmr.quantity == given + + def test_unit_is_used_when_the_data_carries_one(self, sample_network): + network = sample_network.copy() + ds = network.to_dataset() + ds["WaterLevel"].attrs["units"] = "meter" + network.to_dataset = lambda: ds # type: ignore[method-assign] + + nmr = NetworkModelResult(network) + + assert nmr.quantity == Quantity(name="WaterLevel", unit="meter") + def test_init_with_name(self, sample_network): """Test initialization with explicit name""" nmr = NetworkModelResult(sample_network, name="Test_Network") @@ -375,6 +405,71 @@ def test_single_node_dict(self, sample_node_data): assert isinstance(obs_list[0], NodeObservation) assert obs_list[0].node == 123 + def test_nodes_keys_accept_aliases(self, multi_data): + obs_list = NodeObservation.from_multiple( + data=multi_data, nodes={"node_A": "station_0", "node_B": "station_1"} + ) + + assert [obs.at for obs in obs_list] == ["node_A", "node_B"] + + def test_nodes_keys_accept_breakpoints(self, multi_data): + obs_list = NodeObservation.from_multiple( + data=multi_data, + nodes={("reach_1", 24.5): "station_0", ("reach_1", 50.0): "station_1"}, + ) + + assert [obs.at for obs in obs_list] == [("reach_1", 24.5), ("reach_1", 50.0)] + + +class TestReachObservationFromMultiple: + @pytest.fixture + def multi_data(self, sample_node_data): + return pd.DataFrame( + { + "station_0": sample_node_data["WaterLevel"].values, + "station_1": sample_node_data["WaterLevel"].values + 0.1, + }, + index=sample_node_data.index, + ) + + def test_returns_list_of_reach_observations(self, multi_data): + obs_list = ReachObservation.from_multiple( + data=multi_data, reaches={"reach_1": "station_0", "reach_2": "station_1"} + ) + + assert len(obs_list) == 2 + assert all(isinstance(obs, ReachObservation) for obs in obs_list) + assert [obs.reach for obs in obs_list] == ["reach_1", "reach_2"] + assert [obs.name for obs in obs_list] == ["station_0", "station_1"] + + def test_separate_data_sources(self): + obs_list = ReachObservation.from_multiple( + reaches={ + "reach_1": "tests/testdata/network_sensor_1.csv", + "reach_2": "tests/testdata/network_sensor_2.csv", + } + ) + + assert [obs.reach for obs in obs_list] == ["reach_1", "reach_2"] + assert all(len(obs.time) > 0 for obs in obs_list) + + def test_attrs_propagated(self, multi_data): + obs_list = ReachObservation.from_multiple( + data=multi_data, + reaches={"reach_1": "station_0"}, + attrs={"source": "sensor_array"}, + ) + + assert obs_list[0].attrs["source"] == "sensor_array" + + def test_reaches_none_raises(self, multi_data): + with pytest.raises(ValueError, match="'reaches' argument is required"): + ReachObservation.from_multiple(data=multi_data, reaches=None) + + def test_reaches_must_be_dict(self, multi_data): + with pytest.raises(TypeError, match="'reaches' must be a dict"): + ReachObservation.from_multiple(data=multi_data, reaches="reach_1") + class TestNodeModelResult: """Test NodeModelResult class""" @@ -856,7 +951,7 @@ def test_edge_with_no_breakpoints_is_tagged_non_boundary(self): network = Network([BasicReach("r1", a, b, length=100.0)]) - (_, _, data), = network.graph.edges(data=True) + ((_, _, data),) = network.graph.edges(data=True) assert data["boundary"] is False def test_breakpoint_at_reach_start_is_tagged_boundary(self): @@ -866,7 +961,9 @@ def test_breakpoint_at_reach_start_is_tagged_boundary(self): network = Network([BasicReach("r1", a, b, 100.0, breakpoints)]) edges = {frozenset((u, v)): d for u, v, d in network.graph.edges(data=True)} - start_bp_key = frozenset((network.find(node="a"), network.find(reach="r1", distance=0.0))) + start_bp_key = frozenset( + (network.find(node="a"), network.find(reach="r1", distance=0.0)) + ) interior_key = frozenset( ( network.find(reach="r1", distance=0.0), @@ -883,7 +980,9 @@ def test_breakpoint_at_reach_end_is_tagged_boundary(self): network = Network([BasicReach("r1", a, b, 100.0, breakpoints)]) - end_bp_key = frozenset((network.find(node="b"), network.find(reach="r1", distance=100.0))) + end_bp_key = frozenset( + (network.find(node="b"), network.find(reach="r1", distance=100.0)) + ) assert network.graph.edges[tuple(end_bp_key)]["boundary"] is True assert network.graph.edges[tuple(end_bp_key)]["length"] == 0.0 @@ -902,7 +1001,9 @@ def test_end_breakpoint_length_is_clamped_despite_floating_point_noise(self): network = Network([BasicReach("r1", a, b, noisy_length, breakpoints)]) - end_bp_key = frozenset((network.find(node="b"), network.find(reach="r1", distance=100.0))) + end_bp_key = frozenset( + (network.find(node="b"), network.find(reach="r1", distance=100.0)) + ) data = network.graph.edges[tuple(end_bp_key)] assert data["boundary"] is True From 0477c4f79ba9bce022305907bba87c3f2036bb52 Mon Sep 17 00:00:00 2001 From: jpalm3r Date: Tue, 11 Aug 2026 10:41:00 +0200 Subject: [PATCH 3/3] Document locating observations from a MIKE+ database Co-Authored-By: Claude Opus 5 --- docs/user-guide/network.qmd | 40 +++++++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/docs/user-guide/network.qmd b/docs/user-guide/network.qmd index 384b9a4e7..e744d7095 100644 --- a/docs/user-guide/network.qmd +++ b/docs/user-guide/network.qmd @@ -470,6 +470,46 @@ cc_q.skill() Use `ReachObservation` when your measured quantity is representative of the whole reach (e.g. discharge, which is constant along a reach in steady flow). If you need to compare a quantity that varies spatially along the reach (e.g. water level at a specific chainage), use a `NodeObservation` with a `(reach, distance)` tuple instead (see [Option B](#option-b-breakpoint-by-reach-distance-tuple) above). ::: +## Locating observations with a MIKE+ database + +The examples above assume you already know where each sensor sits in the network. A MIKE+ project normally records that itself, in the sqlite database shipped alongside the result files: `m_Measurement` says which file and item each measured timeseries lives in, and `m_Station` says where in the network it belongs. + +Pass that database as `db` and modelskill does the lookup for you: + +```python +quantity = "Pressure" + +network = Network.from_epanet("model.res", quantities=quantity) +network_model = ms.NetworkModelResult(network, item=quantity) + +obs = ms.NodeObservation.from_multiple( + data="calibration.dfs0", + db="model.sqlite", + quantity=quantity, +) + +cc = ms.match(obs, network_model) +``` + +One observation is created per item of the data source, named after the station's asset name. Because the mapping runs item by item rather than location by location, several sensors at the same node — a pair either side of a check valve, say — all become separate observations. + +Reach-uniform quantities work the same way through the sibling method: + +```python +obs_q = ms.ReachObservation.from_multiple( + data="calibration.dfs0", db="model.sqlite", quantity="Flow" +) +``` + +Which class to use is decided by the database, not by you: stations recorded on a junction or a tank are node observations, and stations recorded on a link are reach observations. Asking `NodeObservation` for a quantity that the database places on links raises an error naming `ReachObservation`, and the other way around. + +A few details worth knowing: + +* **`quantity` is optional.** Omit it and the quantity is inferred, as long as the data holds only one for the class you asked for. A calibration file mixing pressure and flow raises an error listing what it found. +* **The quantity name comes from the database, the unit from the data.** Calibration files often carry no usable EUM information, so the database is the only reliable source for the name. +* **`source` picks between files.** It defaults to `data` when that is a path. Pass it explicitly if you hand over an already-read `mikeio.Dataset` or a `DataFrame`, since neither remembers where it came from. +* **Items the database cannot place raise by default**, separating the two causes: a station that exists but has no measurement registered for this file, and an item that is not in the database at all. Pass `on_missing="skip"` to build observations from the rest. + ## Development ### Custom network formats