"""Read, write, and convert gridded setup files.
The module provides shared helpers for NetCDF, ESRI ASCII, and GeoTIFF-style
workflows, including coordinate normalization, header creation, encoding
cleanup, grid metadata handling, and output provenance.
Authors
-------
- Simon Lüdke
"""
import contextlib
import logging
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Dict, Optional, Sequence, Tuple, Union
import numpy as np
import xarray as xr
from mhm_tools.common.constants import NC_ENCODE_DEFAULTS, NO_DATA
from mhm_tools.common.esri_grid import standardize_header, write_grid, write_header
from mhm_tools.common.logger import ErrorLogger, log_arguments
from mhm_tools.common.netcdf import (
apply_cf_baseline_metadata,
generate_bounds_for_all_coords,
get_netcdf_metadata_data_vars,
prepare_dataset_for_netcdf_write,
prepare_time_bounds_encoding,
read_dataset,
)
from mhm_tools.common.provenance import apply_output_provenance
from mhm_tools.common.xarray_utils import (
get_coord_key,
get_dtype,
get_single_data_var,
normalize_lat_lon,
)
logger = logging.getLogger(__name__)
[docs]
@dataclass
class GridDefinition:
"""Container to preserve a dataset's spatial grid metadata."""
template: xr.Dataset
dims: Tuple[str, ...]
[docs]
def get_grid(
ds: xr.Dataset, data_vars: Optional[Union[str, Sequence[str]]] = None
) -> GridDefinition:
"""Extract a grid definition from ``ds``.
Parameters
----------
ds : xr.Dataset
Dataset that defines the spatial/temporal grid.
data_vars : str | sequence[str], optional
Data variables to drop from the returned template. When omitted the
first data variable is used.
"""
if data_vars is None:
var_name = get_single_data_var(ds)
if var_name is None:
msg = "Cannot determine data_var to describe grid."
with ErrorLogger(logger):
raise ValueError(msg)
drop_vars = [var_name]
elif isinstance(data_vars, str):
drop_vars = [data_vars]
var_name = data_vars
else:
drop_vars = list(data_vars)
var_name = drop_vars[0]
if var_name not in ds.data_vars:
msg = f"Grid descriptor variable {var_name} not present in dataset."
with ErrorLogger(logger):
raise ValueError(msg)
template = ds.drop_vars(drop_vars, errors="ignore")
dims = tuple(ds[var_name].dims)
return GridDefinition(template=template, dims=dims)
[docs]
def set_grid(
data: np.ndarray,
grid: GridDefinition,
var_name: str,
data_attrs: Optional[Dict[str, Union[str, float, int]]] = None,
) -> xr.Dataset:
"""Attach ``data`` to a preserved grid definition."""
coords = {}
for name, coord in grid.template.coords.items():
# Only include coords compatible with the target variable dims.
if set(coord.dims).issubset(set(grid.dims)):
coords[name] = coord
da = xr.DataArray(
data,
coords=coords,
dims=grid.dims,
attrs=data_attrs or {},
name=var_name,
)
ds = grid.template.copy(deep=False)
ds[var_name] = da
return ds
[docs]
def crop_file_by_mask(ds, mask_file):
"""Crop file by mask."""
if isinstance(mask_file, xr.Dataset):
mask = mask_file
else:
mask = get_xarray_ds_from_file(mask_file)
lat_key_mask = get_coord_key(mask, lat=True)
lon_key_mask = get_coord_key(mask, lon=True)
lat_key = get_coord_key(ds, lat=True)
lon_key = get_coord_key(ds, lon=True)
return ds.sel(
{
lat_key: slice(mask[lat_key_mask].max(), mask[lat_key_mask].min()),
lon_key: slice(mask[lon_key_mask].min(), mask[lon_key_mask].max()),
}
)
[docs]
def chunk_dataset_space_only(
ds: xr.Dataset, available_mem_gib: float, var_name: Optional[str] = None
) -> Dict[str, int]:
"""Chunk only in space (lat/lon), leaving time whole, sized to available memory.
- Uses 80% of available_mem_gib for a single chunk.
- Computes how many total cells (t * y * x) fit, then allocates all t,
and splits y/x so that t·y·x·bytes_per_cell ≤ work_bytes.
- If no time dimension, behaves similarly with t=1.
"""
logger.info(
f"Chunking spatial dims to fit ≈{available_mem_gib} GiB (time unchunked)"
)
# --- pick one variable to get dtype size ---
if var_name is None:
var_name = get_single_data_var(ds)
dtype_sz = ds[var_name].dtype.itemsize # bytes per element
# --- find coordinate names ---
lat_key = get_coord_key(ds, lat=True)
lon_key = get_coord_key(ds, lon=True)
time_key = None
# with contextlib.suppress(ValueError):
time_key = get_coord_key(ds, time=True, raise_exception=False)
ny = ds.sizes[lat_key]
nx = ds.sizes[lon_key]
nt = ds.sizes.get(time_key, 1)
# --- memory budget in bytes (80%) ---
work_bytes = max(int(0.1 * available_mem_gib * 1024**3), 4 * 1024**2)
# how many total cells fit
max_cells = work_bytes // dtype_sz
# allocate all time steps
cells_per_slice = max_cells // nt
# square-ish spatial block
side = max(1, int(np.sqrt(cells_per_slice)))
y_chunk = min(ny, side)
x_chunk = min(nx, side)
chunks = {lat_key: int(y_chunk), lon_key: int(x_chunk)}
if time_key:
# -1 means “take all” for that dim
chunks[time_key] = -1
logger.debug(
f"Chunk sizes → time: {chunks.get(time_key, '—')}, "
f"{lat_key}: {y_chunk}, {lon_key}: {x_chunk}"
)
return chunks
[docs]
def chunk_dataset_space_and_time(
ds, available_mem_gib, var_name: Optional[str] = None
) -> Dict[str, int]:
"""Chunk dataset adjusting chunk size to avaiable memory.
Simple heuristic:
- try to keep time chunks small (1…4)vi
- make y/x chunks as square as possible
"""
logger.info(
f"Chunking dataset with a max amount of mem of {available_mem_gib:.1f}Gb"
)
# ---------------- metadata only (cheap) --------------------------------
if isinstance(ds, xr.Dataset):
if var_name is None:
var_name = next(iter(ds.data_vars)) # first data variable
var = ds[var_name] # an xarray.Variable wrapper
else:
var = ds
dtype_sz = var.dtype.itemsize # bytes per element
lat_key = get_coord_key(ds, lat=True)
lon_key = get_coord_key(ds, lon=True)
time_key = get_coord_key(ds, time=True, raise_exception=False)
if time_key is None:
return chunk_dataset_space_only(ds, available_mem_gib, var_name)
ny = ds.sizes[lat_key]
nx = ds.sizes[lon_key]
nt = ds.sizes.get(time_key, 1)
# ---------------- convert GiB → bytes and keep 80 % ---------------------
_MIN_BYTES_PER_CHUNK = 4 * 1024**2 # 4MB
work_bytes = max(int(0.8 * available_mem_gib * 1024**3), _MIN_BYTES_PER_CHUNK)
max_cells = work_bytes // dtype_sz # how many array elements fit
# ---------------- choose chunk sizes -----------------------------------
t_chunk = min(nt, 4) if time_key else None # ≤4 along time
cells_per_t = max_cells // (t_chunk or 1)
side = max(1, int(np.sqrt(cells_per_t))) # square-ish y/x chunk
y_chunk = min(ny, side)
x_chunk = min(nx, side)
chunks = {lat_key: int(y_chunk), lon_key: int(x_chunk)}
if time_key:
t_chunk = max(1, max_cells // max(1, y_chunk * x_chunk))
chunks[time_key] = int(t_chunk)
logger.debug(f" The chunks used are {chunks}")
return chunks
[docs]
class ChunkType(Enum):
"""Define Types of chunking.
SPACE: Only chunking in space. Time dimension is conserved.
TIME: Chunking predominately in time. If necessary also in space.
"""
SPACE = 1
TIME = 2
[docs]
@log_arguments()
def chunk_dataset(ds, chunk_type, available_mem_gib, var_name=None):
"""Chunk xarray.DataSet depending on chunk_type and available memory."""
if chunk_type == ChunkType.TIME:
chunks = chunk_dataset_space_and_time(ds, available_mem_gib, var_name)
if chunk_type == ChunkType.SPACE:
chunks = chunk_dataset_space_only(ds, available_mem_gib, var_name)
try:
return ds.chunk(chunks)
except Exception as e:
logger.error(chunks)
logger.error(ds)
with ErrorLogger(logger):
raise e
[docs]
def write_xarray_to_netcdf(
ds,
file_path,
var_name=None,
encoding=None,
engine="netcdf4",
):
"""Write an xarray Dataset or DataArray to a NetCDF file.
Parameters
----------
ds : xr.Dataset or xr.DataArray
Dataset or data array to write.
file_path : str or pathlib.Path
Target NetCDF file path.
var_name : str, optional
Data variable to write or DataArray name override.
encoding : dict, optional
Per-variable NetCDF encoding.
engine : str, default "netcdf4"
Xarray NetCDF backend engine.
Returns
-------
None
"""
ds, data_vars = _get_netcdf_write_dataset_and_data_vars(ds, var_name)
ds = apply_output_provenance(ds)
apply_cf_baseline_metadata(ds, data_vars)
if encoding is None:
encoding = {
var: {
"zlib": True,
"complevel": 4,
"shuffle": True,
**NC_ENCODE_DEFAULTS,
}
for var in data_vars
}
else:
encoding = {key: value for key, value in encoding.items() if key in data_vars}
ds = prepare_time_bounds_encoding(ds)
try:
ds_clean, safe_encoding = prepare_dataset_for_netcdf_write(
ds, data_vars, encoding
)
ds_clean.to_netcdf(
file_path, engine=engine, format="NETCDF4", encoding=safe_encoding
)
except ValueError:
logger.error(f"Error while writing to {file_path}")
logger.error(ds)
logger.info(f"Trying to write without encoding {encoding}")
ds = prepare_time_bounds_encoding(ds, strip_time_attrs=True)
ds.to_netcdf(file_path, engine=engine, format="NETCDF4")
except Exception as e:
with ErrorLogger(logger):
raise e
def _get_netcdf_write_dataset_and_data_vars(ds, var_name=None):
"""Return a Dataset and payload variable names for NetCDF writing.
Parameters
----------
ds : xr.Dataset or xr.DataArray
Object to prepare for NetCDF writing.
var_name : str, optional
Requested data variable name.
Returns
-------
tuple
``(dataset, data_vars)`` where data_vars excludes bounds/grid mappings.
"""
if isinstance(ds, xr.DataArray):
var_name = ds.name if var_name is None else var_name
logger.debug(f"var {var_name}")
ds = ds.to_dataset(name=var_name)
data_vars = [var_name]
logger.debug(f"Creating data vars from dataarray name: {data_vars}")
elif var_name is None or var_name not in ds.data_vars:
if var_name is not None and var_name not in ds.data_vars:
logger.warning(
f"Requested var_name {var_name!r} not found in dataset. "
"Using all data variables instead."
)
logger.info(f"Taking data vars from list of ds.data_vars {ds.data_vars}")
metadata_vars = get_netcdf_metadata_data_vars(ds)
data_vars = [var for var in ds.data_vars if var not in metadata_vars]
else:
data_vars = [var_name]
logger.info(f"Setting data vars as input varname [var_name]: {data_vars}")
if not data_vars:
logger.warning(
"Dataset has no data variables. Writing coordinate-only dataset."
)
logger.debug(f"data vars: {data_vars}")
return ds, data_vars
[docs]
@log_arguments()
def get_xarray_ds_from_file( # noqa: PLR0912
file_path,
var_name=None,
chunking=False,
available_mem_gib=None,
chunk_type=ChunkType.SPACE,
use_mfdataset=False,
engine="netcdf4",
normalize_latlon_coords=False,
force_decending_y=False,
force_ascending_y=False,
landcover=False,
landcover_year_start=None,
create_bounds=False,
):
"""Read file and return xarray dataset."""
file_path = Path(file_path)
logger.debug(f"Reading {file_path} to xarray with chunking = {chunking}")
ds_out = None
if not file_path.is_file():
msg = f"File path does not point to an existing file: {file_path}"
with ErrorLogger(logger):
raise ValueError(msg)
suffix = file_path.suffix.lower()
if suffix == ".asc":
if landcover:
logger.info("Reading ascii landcover file.")
ds_out = read_ascii_to_xarray(
filepath=file_path,
var_name=var_name,
landcover=landcover,
landcover_year_start=landcover_year_start,
)
else:
ds_out = read_ascii_to_xarray(
filepath=file_path,
var_name=var_name,
)
chunk_type = ChunkType.SPACE
elif suffix == ".nc":
ds_out = read_dataset(
file_path=file_path,
use_mfdataset=use_mfdataset,
engine=engine,
)
elif suffix in {".tif", ".tiff"}:
import rioxarray as rxr
da = rxr.open_rasterio(file_path)
if var_name is not None:
da = da.rename(var_name)
else:
da.name = "data"
if "band" in da.dims and da.sizes.get("band", 0) == 1:
da = da.squeeze("band", drop=True)
nodata = None
with contextlib.suppress(Exception):
nodata = da.rio.nodata
if nodata is not None:
da.attrs.setdefault("nodata_value", nodata)
da.attrs.setdefault("_FillValue", nodata)
if "x" in da.coords:
da["x"].attrs.setdefault("axis", "X")
if "y" in da.coords:
da["y"].attrs.setdefault("axis", "Y")
ds_out = da.to_dataset()
if "spatial_ref" in ds_out.coords:
ds_out = ds_out.drop_vars("spatial_ref")
else:
msg = (
"Reading file types other than asci, netcdf, and geotiff is not "
f"implemented. The suffix of the file was: {file_path.suffix}"
)
with ErrorLogger(logger):
raise NotImplementedError(msg)
lat_key = get_coord_key(ds_out, lat=True, raise_exception=False)
lon_key = get_coord_key(ds_out, lon=True, raise_exception=False)
# force correct order of y coordinate
if lat_key is not None and (
(force_decending_y and ds_out[lat_key].data[0] < ds_out[lat_key].data[-1])
or (force_ascending_y and ds_out[lat_key].data[0] > ds_out[lat_key].data[-1])
):
ds_out = ds_out.sel({lat_key: slice(None, None, -1)})
logger.debug(ds_out)
logger.debug(lat_key)
logger.debug(lon_key)
if normalize_latlon_coords:
# re-name input coords to lat and lon
ds_out = normalize_lat_lon(
ds_out, lat_key=lat_key, lon_key=lon_key, raise_exceptions=False
)
if create_bounds:
ds_out = generate_bounds_for_all_coords(ds_out)
if lon_key is None and lat_key is None:
logger.warning("Dataset does not have lon and lat key.")
elif lon_key is None or lat_key is None:
logger.error("Dataset has only one of lon at lat keys.")
if chunking and available_mem_gib is not None:
ds_out = chunk_dataset(ds_out, chunk_type, available_mem_gib, var_name)
else:
# if no chunking remove chunking encoding because this might cause errors while writing
for name in list(ds_out.variables):
enc = ds_out.variables[name].encoding
enc.pop("chunksizes", None)
enc.pop("_ChunkSizes", None)
enc.pop("chunks", None)
enc["contiguous"] = True
if ds_out is None:
msg = f"The dataset read from {file_path} is empty."
with ErrorLogger(logger):
raise NotImplementedError()
logger.debug(f"ds_out: {ds_out}")
return ds_out
[docs]
def write_xarray_to_file(
ds,
file_path,
var_name=None,
create_folder=True,
encoding=None,
engine="netcdf4",
resolution=None,
):
"""Write xarray Datasets to file with file type depending on the file suffix."""
file_path = Path(file_path)
if file_path.suffix not in {".asc", ".nc"}:
msg = (
"Writing to file types other than asci and netcdf is not implemented. "
f"The suffix of the file was: {file_path.suffix}"
)
with ErrorLogger(logger):
raise NotImplementedError(msg)
if create_folder and not file_path.parent.is_dir():
file_path.parent.mkdir(parents=True, exist_ok=True)
if file_path.is_file():
file_path.unlink()
logger.info(f"Writing file to {file_path}")
# ds = chunk_if_too_big(ds)
# ds = ds.chunk({'time': 512, 'lat': 121, 'lon': 131})
if file_path.suffix == ".asc":
write_xarray_to_ascii(ds, file_path, var_name, resolution=resolution)
elif file_path.suffix == ".nc":
write_xarray_to_netcdf(ds, file_path, var_name, encoding, engine)
[docs]
def write_xarray_to_ascii(
dataset, filepath, data_var=None, nodata_value=None, resolution=None
):
"""Write xarray Dataset to an ASCII file that can be read by mHM."""
# check if a data_var can be optained for writing the data
if data_var is None and isinstance(dataset, xr.Dataset):
data_var = get_single_data_var(dataset)
if data_var is None:
logger.error(
f"Data can not be written to {filepath} as the dataset has multiple data_vars, which is incompatible with asci or no datavar exists."
)
return
# get the data from the dataset
data = dataset[data_var] if isinstance(dataset, xr.Dataset) else dataset
# set the nodata value
dtype = get_dtype(data)
if nodata_value is None:
is_int = issubclass(np.dtype(dtype).type, (np.integer, np.unsignedinteger))
typ = int if is_int else float
nodata_value = typ(NO_DATA)
header = create_header(dataset, no_data_value=nodata_value)
if resolution is not None:
header["cellsize"] = resolution
data_to_write = data
if isinstance(data_to_write, xr.DataArray):
data_to_write = data_to_write.data
if data_to_write.dtype.kind in ["i", "u", "f"]: # i=int, u=unsigned, f=float
data_to_write = np.where(np.isnan(data_to_write), nodata_value, data_to_write)
out_header_str = write_grid(
file=filepath, header=header, dtype=dtype, data=data_to_write
)
logger.info(f"Writting file to {filepath}")
logger.debug(f"Header written:\n{out_header_str}")
[docs]
def read_ascii_to_xarray(
filepath,
var_name=None,
landcover=False,
landcover_year_start=None,
):
"""Read an mHM readable asci file to an xarray dataset."""
# Read the header from the file
with filepath.open("r") as f:
header = {}
for i, line in enumerate(f.readlines()):
line_striped = line.strip()
if not line_striped:
continue
logger.debug(f"File {filepath.name} {i}: {line_striped}")
key, value = line_striped.split()
header[key.lower()] = float(value) if "." in value else int(value)
if len(header) == 6:
break
# Extract header information
ncols = header["ncols"]
nrows = header["nrows"]
xllcorner = header["xllcorner"]
yllcorner = header["yllcorner"]
cellsize = header["cellsize"]
nodata_value = header["nodata_value"]
# Load the data values
data_values = np.loadtxt(filepath, skiprows=i + 1)
if isinstance(nodata_value, int):
data_values = data_values.astype(np.int32)
# Calculate latitude and longitude coordinates
lon = np.arange(
xllcorner + cellsize / 2, xllcorner + (ncols + 0.5) * cellsize, cellsize
)
lat = np.arange(
yllcorner + (nrows - 0.5) * cellsize, yllcorner - cellsize / 2, -cellsize
)
logger.debug(lon)
logger.debug(lat)
# Create DataArray with lat/lon dimensions and nodata value
name = "data" if var_name is None else var_name
coords = {"lon": ("lon", lon, {"axis": "X"}), "lat": ("lat", lat, {"axis": "Y"})}
# If this is a landcover file add a 1-element time coordinate
if landcover and landcover_year_start is not None:
start_ts = np.datetime64(f"{landcover_year_start}-01-01", "ns")
time = np.array([start_ts], dtype="datetime64[ns]")
coords["time"] = ("time", time)
# If we added a time coordinate, expand the data to have a leading time dim
if "time" in coords:
data_arr = np.expand_dims(data_values, axis=0) # shape (1, nrows, ncols)
dims = ["time", "lat", "lon"]
else:
data_arr = data_values
dims = ["lat", "lon"]
da = xr.DataArray(
data=data_arr,
dims=dims,
coords=coords,
name=name,
attrs={"nodata_value": nodata_value, "_FillValue": nodata_value},
)
# Convert to Dataset
ds = da.to_dataset()
# Drop spatial_ref if present
if "spatial_ref" in ds.coords:
ds = ds.reset_coords("spatial_ref", drop=True)
return ds
[docs]
def get_coord_values(ds, lat=False, lon=False):
"""Get latitude or longitude values from DataSet."""
key = get_coord_key(ds, lat=lat, lon=lon)
return ds[key].values
[docs]
def get_dataset_from_path(
path,
var_name=None,
chunking=None,
available_mem_gib=None,
chunk_type=ChunkType.SPACE,
use_mfdataset=False,
engine="netcdf4",
normalize_latlon_coords=False,
force_decending_y=False,
force_ascending_y=False,
landcover=False,
landcover_year_start=None,
available_mem=None,
file_name="*.*",
):
"""Load a dataset from a file, directory, or pattern.
This mirrors ``get_xarray_ds_from_file`` for single-file inputs and
extends it to directories (multi-file datasets).
"""
if path is None:
path_is_none_msg = "Input path is None. Please provide a valid file path, directory path, or glob pattern."
with ErrorLogger(logger):
raise ValueError(path_is_none_msg)
if available_mem_gib is None and available_mem is not None:
available_mem_gib = available_mem
if chunking is None:
chunking = True
if chunking is None:
chunking = False
def _postprocess(ds_out):
lat_key = get_coord_key(ds_out, lat=True, raise_exception=False)
lon_key = get_coord_key(ds_out, lon=True, raise_exception=False)
if lat_key is not None and (
(force_decending_y and ds_out[lat_key].data[0] < ds_out[lat_key].data[-1])
or (
force_ascending_y and ds_out[lat_key].data[0] > ds_out[lat_key].data[-1]
)
):
ds_out = ds_out.sel({lat_key: slice(None, None, -1)})
logger.debug(ds_out)
logger.debug(lat_key)
logger.debug(lon_key)
if normalize_latlon_coords:
ds_out = normalize_lat_lon(
ds_out, lat_key=lat_key, lon_key=lon_key, raise_exceptions=False
)
if lon_key is None and lat_key is None:
logger.warning("Dataset does not have lon and lat key.")
elif lon_key is None or lat_key is None:
logger.error("Dataset has only one of lon at lat keys.")
if chunking and available_mem_gib is not None:
ds_out = chunk_dataset(ds_out, chunk_type, available_mem_gib, var_name)
else:
for name in list(ds_out.variables):
enc = ds_out.variables[name].encoding
enc.pop("chunksizes", None)
enc.pop("_ChunkSizes", None)
enc.pop("chunks", None)
enc["contiguous"] = True
return ds_out
def _is_dir_or_list(p):
if isinstance(p, list):
return True
if isinstance(p, str):
p = Path(p)
return p.is_dir()
if _is_dir_or_list(path):
file_list = []
if not isinstance(path, list):
path = Path(path)
file_list = list(path.rglob(file_name))
else:
path = [Path(p) for p in path]
for p in path:
if p.is_file():
file_list.append(p)
elif p.is_dir():
file_list.extend(list(p.rglob(file_name)))
if not file_list:
with ErrorLogger(logger):
msg = f"No files found in {path}."
raise ValueError(msg)
if len(file_list) == 1:
return get_xarray_ds_from_file(
file_list[0],
var_name=var_name,
chunking=chunking,
available_mem_gib=available_mem_gib,
chunk_type=chunk_type,
use_mfdataset=use_mfdataset,
engine=engine,
normalize_latlon_coords=normalize_latlon_coords,
force_decending_y=force_decending_y,
force_ascending_y=force_ascending_y,
landcover=landcover,
landcover_year_start=landcover_year_start,
)
non_nc = [p for p in file_list if Path(p).suffix != ".nc"]
if non_nc:
with ErrorLogger(logger):
msg = "Multi-file loading supports NetCDF files only."
raise ValueError(msg)
ds_out = read_dataset(file_list, use_mfdataset=use_mfdataset, engine=engine)
return _postprocess(ds_out)
path_in = path
path = Path(path_in)
if path.is_file():
return get_xarray_ds_from_file(
path,
var_name=var_name,
chunking=chunking,
available_mem_gib=available_mem_gib,
chunk_type=chunk_type,
use_mfdataset=use_mfdataset,
engine=engine,
normalize_latlon_coords=normalize_latlon_coords,
force_decending_y=force_decending_y,
force_ascending_y=force_ascending_y,
landcover=landcover,
landcover_year_start=landcover_year_start,
)
path_str = str(path_in)
if any(w in path_str for w in ("*", "?", "[", "]")) and path_str.endswith(".nc"):
ds_out = read_dataset(path_str, use_mfdataset=use_mfdataset, engine=engine)
return _postprocess(ds_out)
with ErrorLogger(logger):
msg = f"Path {path} does not exist."
raise ValueError(msg)
