"""
Utilities for parsing a project file.
"""
import shlex
import textwrap
from collections import defaultdict
from dataclasses import asdict, dataclass
from dataclasses import field as data_field
from datetime import datetime
from itertools import chain
from os import PathLike
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
import numpy as np
import pandas as pd
import xarray as xr
import imod
import imod.logging
from imod.logging.loglevel import LogLevel
FilePath = Union[str, "PathLike[str]"]
KEYS = {
"(bnd)": ("ibound",),
"(top)": ("top",),
"(bot)": ("bottom",),
"(thk)": ("thickness",),
"(khv)": ("kh",),
"(kva)": ("vertical_anisotropy",),
"(kdw)": ("transmissivity",),
"(kvv)": ("kv",),
"(vcw)": ("resistance",),
"(shd)": ("head",),
"(sto)": ("storage_coefficient",),
"(spy)": ("specific_yield",),
"(por)": ("porosity",),
"(ani)": ("factor", "angle"),
"(hfb)": ("gen",),
"(ibs)": (None),
"(pwt)": (None),
"(sft)": (None),
"(obs)": (None),
"(cbi)": (None),
"(sco)": (None),
"(dsp)": (None),
"(ics)": (None),
"(fcs)": (None),
"(ssc)": (None),
"(fod)": (None),
"(fos)": (None),
"(rct)": (None),
"(con)": (None),
"(pst)": (None),
}
DATE_KEYS = {
"(uzf)": (None,),
"(rch)": ("rate",),
"(evt)": ("rate", "surface", "depth"),
"(drn)": ("conductance", "elevation"),
"(olf)": ("elevation",),
"(riv)": ("conductance", "stage", "bottom_elevation", "infiltration_factor"),
"(isg)": ("isg",),
"(sfr)": ("isg",),
"(lak)": (None,),
"(wel)": ("ipf",),
"(mnw)": (None,),
"(ghb)": ("conductance", "head"),
"(chd)": ("head",),
"(fhb)": (None,),
"(fde)": (None,),
"(tvc)": (None,),
}
METASWAP_VARS = (
"boundary",
"landuse",
"rootzone_thickness",
"soil_physical_unit",
"meteo_station_number",
"surface_elevation",
"artificial_recharge",
"artificial_recharge_layer",
"artificial_recharge_capacity",
"wetted_area",
"urban_area",
"urban_ponding_depth",
"rural_ponding_depth",
"urban_runoff_resistance",
"rural_runoff_resistance",
"urban_runon_resistance",
"rural_runon_resistance",
"urban_infiltration_capacity",
"rural_infiltration_capacity",
"perched_water_table_level",
"soil_moisture_fraction",
"conductivitiy_factor",
"plot_number",
"steering_location",
"plot_drainage_level",
"plot_drainage_resistance",
)
class _LineIterator:
"""
Like iter(lines), but we can go back and we check if we're finished.
"""
def __init__(self, lines: List[List[str]]):
self.lines = lines
self.count = -1
self.length = len(lines)
def __iter__(self):
return self
def __next__(self) -> List[str]:
if self.finished:
raise StopIteration
self.count += 1
return self.lines[self.count]
def back(self) -> None:
self.count = max(self.count - 1, -1)
@property
def finished(self) -> bool:
return (self.count + 1) >= self.length
def _tokenize(line: str) -> List[str]:
"""
A value separator in Fortran list-directed input is:
* A comma if period decimal edit mode is POINT.
* One or more contiguous spaces (blanks); no tabs.
Other remarks:
* Values, except for character strings, cannot contain blanks.
* Strings may be unquoted if they do not start with a digit and no value
separators.
* Character strings can be quoted strings, using pairs of quotes ("), pairs
of apostrophes (').
* A quote or apostrophe must be preceded by a value separator to initite a
quoted string.
* An empty entry consists of two consecutive commas (or semicolons).
For the use here (parsing IMOD's project files), we ignore:
* A semicolon value separator if period decimal edit mode is COMMA.
* Complex constants given as two real constants separated by a comma and
enclosed in parentheses.
* Repetition counts: 4*(3.,2.) 2*, 4*'hello'
Furthermore, we do not expect commas inside of the project file entries,
since we expect:
* Package names: unquoted character strings.
* File paths: will not contain commas, no single apostrophe, nor a single
quote symbol, may contain whitespace if quoted. * Integers for counts and
settings.
* Floats for addition and multiplication values.
* Simple character strings for period names (summer, winter). These
technically could contain commas if quoted, which is very unlikely.
* No quotes or apostrophes are escaped.
With these assumptions, we can limit complexity considerably (see the
PyLiDiRe link for a complete implementation):
* First we split by comma (we don't expected commas in quote strings).
* Next we split by whitespace, unless quoted.
We can expect both single and double quotes, even within a single line:
shlex.split() handles this. Note that additional entries are likely
allowed, as the Fortran implementation only reads what is necessary,
then stops parsing.
See also:
* https://stackoverflow.com/questions/36165050/python-equivalent-of-fortran-list-directed-input
* https://gitlab.com/everythingfunctional/PyLiDiRe
* https://docs.oracle.com/cd/E19957-01/805-4939/6j4m0vnc5/index.html
* The Fortran 2003 Handbook
Examples
--------
Raise ValueError, due to missing closing quotation. (Can be enabled
shlex.split(s, posix=False)):
>> _tokenize("That's life")
>> _tokenize("That 's life'")
>> ["That", "s life"]
>> _tokenize("That,'s life'")
>> ["That", "s life"]
"""
values = [v.strip().replace("\\", "/") for v in line.split(",")]
tokens = list(chain.from_iterable(shlex.split(v) for v in values))
return tokens
def _wrap_error_message(
exception: Exception, description: str, lines: _LineIterator
) -> None:
lines.back()
content = next(lines)
number = lines.count + 1
raise type(exception)(
f"{exception}\n"
f"Failed to parse {description} for line {number} with content:\n{content}"
)
def _parse_blockheader(lines: _LineIterator) -> Tuple[int, str, str]:
try:
no_result = None, None, None
line = next(lines)
# Skip if it's an empty line.
if len(line) == 0:
return no_result
first = line[0].lower()
if first in ("periods", "species"):
return 1, first, None
# The line must contain atleast nper, key, active.
elif len(line) >= 3:
n = int(first)
key = line[1].lower()
active = line[2]
return n, key, active
# It's a comment or something.
else:
return no_result
except Exception as e:
_wrap_error_message(e, "block header", lines)
def _parse_time(lines: _LineIterator) -> str:
try:
line = next(lines)
date = line[0].lower()
if len(line) > 1:
time = line[1]
return f"{date} {time}"
else:
return date
except Exception as e:
_wrap_error_message(e, "date time", lines)
def _parse_blockline(lines: _LineIterator, time: str = None) -> Dict[str, Any]:
try:
line = next(lines)
content = {
"active": bool(int(line[0])),
"is_constant": int(line[1]),
"layer": int(line[2]),
"factor": float(line[3]),
"addition": float(line[4]),
"constant": float(line[5]),
}
if content["is_constant"] == 2:
content["path"] = Path(line[6]).resolve()
if time is not None:
content["time"] = time
return content
except Exception as e:
_wrap_error_message(e, "entries", lines)
def _parse_nsub_nsystem(lines: _LineIterator) -> Tuple[int, int]:
try:
line = next(lines)
n_entry = int(line[0])
n_system = int(line[1])
return n_entry, n_system
except Exception as e:
_wrap_error_message(e, "number of sub-entries and number of systems", lines)
def _parse_notimeblock(
lines: _LineIterator,
fields: List[str],
) -> Dict[str, Any]:
n_entry, n_system = _parse_nsub_nsystem(lines)
if len(fields) != n_entry:
raise ValueError(
f"Expected NSUB entry of {len(fields)} for {fields}, read: {n_entry}"
)
content = {
field: [_parse_blockline(lines) for _ in range(n_system)] for field in fields
}
content["n_system"] = n_system
return content
def _parse_capblock(
lines: _LineIterator,
) -> Dict[str, Any]:
fields = METASWAP_VARS
n_entry, n_system = _parse_nsub_nsystem(lines)
if n_entry == 21:
# Remove layer entry.
fields = list(fields[:22]).pop(8)
elif n_entry == 22:
fields = fields[:22]
elif n_entry == 26:
pass
else:
raise ValueError(
f"Expected NSUB entry of 21, 22, or 26 for {fields}, read: {n_entry}"
)
content = {
field: [_parse_blockline(lines) for _ in range(n_system)] for field in fields
}
content["n_system"] = n_system
return content
def _parse_extrablock(lines: _LineIterator, n: int) -> Dict[str, List[str]]:
"""Parse the MetaSWAP "extra files" block"""
return {"paths": [next(lines) for _ in range(n)]}
def _parse_timeblock(
lines: List[str],
fields: List[str],
n: int,
) -> Dict[str, Any]:
n_fields = len(fields)
content = defaultdict(list)
for _ in range(n):
time = _parse_time(lines)
content["time"].append(time)
n_entry, n_system = _parse_nsub_nsystem(lines)
if n_fields != n_entry:
raise ValueError(
f"Expected NSUB entry of {n_fields} for {fields}, read: {n_entry}"
)
for field in fields:
content[field].extend(
[_parse_blockline(lines, time) for _ in range(n_system)]
)
content["n_system"] = n_system
return content
def _parse_pcgblock(lines: _LineIterator) -> Dict[str, Any]:
try:
line = next(lines)
# TODO: which are optional? How many to expect?
# Check for an empty line to terminate the block?
types = {
"mxiter": int,
"iter1": int,
"hclose": float,
"rclose": float,
"relax": float,
"npcond": int,
"iprpcg": int,
"mutpcg": int,
"damppcg": float,
"damppcgt": float,
"iqerror": int,
"qerror": float,
}
if len(line) == 12:
line_iterator = iter(line)
content = {
k: valuetype(next(line_iterator)) for k, valuetype in types.items()
}
elif any("=" in s for s in line):
pcglines = [line] + [next(lines) for _ in range(11)]
content = {}
for line in pcglines:
# undo separation, partition on equality sign instead.
line = "".join(line)
key, _, value = line.lower().partition("=")
value = types[key](value)
content[key] = value
else:
raise ValueError(
f"Expected 12 KEY = VALUE pairs, or 12 values. Found {len(line)}"
)
return content
except Exception as e:
_wrap_error_message(e, "PCG entry", lines)
def _parse_periodsblock(lines: _LineIterator) -> Dict[str, str]:
try:
periods = {}
while not lines.finished:
line = next(lines)
# Stop if we encounter an empty line.
if len(line) == 0:
break
# Read the alias
alias = line[0]
# Now read the time associated with it.
start = _parse_time(lines)
periods[alias] = start
return periods
except Exception as e:
_wrap_error_message(e, "periods data block", lines)
def _parse_speciesblock(lines: _LineIterator):
try:
species = {}
while not lines.finished:
line = next(lines)
# Stop if we encounter an empty line.
if len(line) == 0:
break
name, nr = line
nr = int(nr)
species[nr] = name
return species
except Exception as e:
_wrap_error_message(e, "species entry", lines)
def _parse_block(lines: _LineIterator, content: Dict[str, Any]) -> None:
"""
Mutates content dict.
"""
n = key = active = None
# A project file may contain any number of lines outside of a "topic"
# block. _parse_blockheader will return triple None in that case.
while key is None and not lines.finished:
n, key, active = _parse_blockheader(lines)
try:
if key in KEYS:
if n != 1:
raise ValueError(f"Expected N=1 for {key}, read: {n}")
fields = KEYS[key]
blockcontent = _parse_notimeblock(lines, fields)
elif key in DATE_KEYS:
fields = DATE_KEYS[key]
blockcontent = _parse_timeblock(lines, fields, n)
elif key == "(cap)":
blockcontent = _parse_capblock(lines)
elif key == "(pcg)":
blockcontent = _parse_pcgblock(lines)
elif key == "periods":
blockcontent = _parse_periodsblock(lines)
elif key == "species":
blockcontent = _parse_speciesblock(lines)
elif key == "extra":
blockcontent = _parse_extrablock(lines, n)
else:
other = ("(pcg)", "(gcg)", "(vdf)")
options = tuple(KEYS.keys()) + tuple(DATE_KEYS.keys()) + other
lines.back()
line = next(lines)
number = lines.count + 1
raise ValueError(
f"Failed to recognize header keyword: {key}. Expected one of keywords {options}"
f"\nErrored in line {number} with entries:\n{line}"
)
except Exception as e:
raise type(e)(f"{e}\nError occurred for keyword: {key}")
if blockcontent is not None and active is not None:
blockcontent["active"] = active
content[key] = blockcontent
return
def _process_package_entry(entry: Dict):
"""
The iMOD project file supports constants in lieu of IDFs.
"""
coords = {"layer": entry["layer"]}
dims = ("layer",)
if "path" not in entry:
path = None
header = {"coords": coords}
value = entry["constant"]
else:
path = entry["path"]
header = imod.idf.header(path, pattern="{name}")
value = None
header["dims"] = dims
return path, header, value
def _merge_coords(headers: List[Dict[str, Any]]) -> Dict[str, np.ndarray]:
coords = defaultdict(list)
for header in headers:
for key, value in header["coords"].items():
coords[key].append(value)
return {k: np.unique(coords[k]) for k in coords}
def _try_read_with_func(func, path, *args, **kwargs):
try:
return func(path, *args, **kwargs)
except Exception as e:
raise type(e)(f"{e}. Error thrown while opening file: {path}")
def _get_array_transformation_parameters(
headers: List[Dict[str, Any]], key: str, dim: str
) -> Union[xr.DataArray | float]:
"""
In imod5 prj files one can add linear transformation parameters to transform
the data read from an idf file: we can specify a multiplication factor and a
constant that will be added to the values. The factor and addition
parameters can be can be scalar (if applied to 1 idf), or they can be
xr.DataArrays if the factor and addition are for example layer- or
time-dependent (if both we apply the transformations one at a time)
Parameters
----------
headers: List[Dict[str, Any]]
prj-file lines which we want to import, serialized as a dictionary.
key: str
specifies the name of the transformation parameter in the idf file.
Usually "factor" or "addition"
dim: str
the name of the dimension over which transformation parameters are
expected to differ for the current import. Usually "time"or "layer"
"""
if dim in headers[0].keys():
return xr.DataArray(
data=[header[key] for header in headers],
dims=(dim,),
coords={dim: [header[dim] for header in headers]},
)
else:
return headers[0][key]
def _create_dataarray_from_paths(
paths: List[str], headers: List[Dict[str, Any]], dim: str
) -> xr.DataArray:
factor = _get_array_transformation_parameters(headers, "factor", dim)
addition = _get_array_transformation_parameters(headers, "addition", dim)
da = _try_read_with_func(
imod.formats.array_io.reading._load,
paths,
use_cftime=False,
_read=imod.idf._read,
headers=headers,
)
# Ensure factor and addition do not have more dimensions than da
if isinstance(factor, xr.DataArray):
missing_dims = set(factor.dims) - set(da.dims)
if missing_dims:
factor = factor.isel({d: 0 for d in missing_dims}, drop=True)
addition = addition.isel({d: 0 for d in missing_dims}, drop=True)
return da * factor + addition
def _create_dataarray_from_values(
values: List[float], headers: List[Dict[str, Any]], dim: str
):
factor = _get_array_transformation_parameters(headers, "factor", dim)
addition = _get_array_transformation_parameters(headers, "addition", dim)
coords = _merge_coords(headers)
firstdims = headers[0]["dims"]
shape = [len(coord) for coord in coords.values()]
da = xr.DataArray(np.reshape(values, shape), dims=firstdims, coords=coords)
return da * factor + addition
def _create_dataarray(
paths: List[str], headers: List[Dict[str, Any]], values: List[float], dim: str
) -> xr.DataArray:
"""
Create a DataArray from a list of IDF paths, or from constant values.
There are mixed cases possible, where some of the layers or stress periods
contain only a single constant value, and the others are specified as IDFs.
In that case, we cannot do a straightforward concatenation.
"""
values_valid = []
paths_valid = []
headers_paths = []
headers_values = []
for path, header, value in zip(paths, headers, values):
if path is None:
headers_values.append(header)
values_valid.append(value)
else:
headers_paths.append(header)
paths_valid.append(path)
if paths_valid and values_valid:
# Both lists contain entries: mixed case.
dap = _create_dataarray_from_paths(paths_valid, headers_paths, dim=dim)
dav = _create_dataarray_from_values(values_valid, headers_values, dim=dim)
dap.name = "tmp"
dav.name = "tmp"
da = xr.merge((dap, dav), join="outer")["tmp"]
elif paths_valid:
# Only paths provided
da = _create_dataarray_from_paths(paths_valid, headers_paths, dim=dim)
elif values_valid:
# Only scalar values provided
da = _create_dataarray_from_values(values_valid, headers_values, dim=dim)
return da
def _open_package_idf(
block_content: Dict[str, Any], variables: Sequence[str]
) -> list[dict[str, xr.DataArray]]:
das = {}
for variable in variables:
variable_content = block_content[variable]
paths = []
headers = []
values = []
for entry in variable_content:
path, header, value = _process_package_entry(entry)
header["name"] = variable
header["dims"] = ["layer"]
header["layer"] = entry["layer"]
header["addition"] = entry["addition"]
header["factor"] = entry["factor"]
paths.append(path)
headers.append(header)
values.append(value)
das[variable] = _create_dataarray(paths, headers, values, dim="layer")
return [das]
def _process_time(time: str, yearfirst: bool = True):
if time == "steady-state":
time = None
else:
if yearfirst:
if len(time) == 19:
time = datetime.strptime(time, "%Y-%m-%d %H:%M:%S")
elif len(time) == 10:
time = datetime.strptime(time, "%Y-%m-%d")
else:
raise ValueError(
f"time data {time} does not match format "
'"%Y-%m-%d %H:%M:%S" or "%Y-%m-%d"'
)
else:
if len(time) == 19:
time = datetime.strptime(time, "%d-%m-%Y %H:%M:%S")
elif len(time) == 10:
time = datetime.strptime(time, "%d-%m-%Y")
else:
raise ValueError(
f"time data {time} does not match format "
'"%d-%m-%Y %H:%M:%S" or "%d-%m-%Y"'
)
return time
def _process_boundary_condition_entry(entry: Dict, periods: Dict[str, datetime]):
"""
The iMOD project file supports constants in lieu of IDFs.
Also process repeated stress periods (on a yearly basis): substitute the
original date here.
"""
coords = {}
timestring = entry["time"]
# Resolve repeating periods first:
time = periods.get(timestring)
if time is not None:
repeat = time
else:
# this resolves e.g. "steady-state"
time = _process_time(timestring)
repeat = None
if time is None:
dims = ()
else:
dims = ("time",)
coords["time"] = time
# 0 signifies that the layer must be determined on the basis of
# bottom elevation and stage.
layer = entry["layer"]
if layer <= 0:
layer is None
else:
coords["layer"] = layer
dims = dims + ("layer",)
if "path" not in entry:
path = None
header = {"coords": coords}
value = entry["constant"]
else:
path = entry["path"]
header = imod.idf.header(path, pattern="{name}")
value = None
header["addition"] = entry["addition"]
header["factor"] = entry["factor"]
header["dims"] = dims
if layer is not None:
header["layer"] = layer
if time is not None:
header["time"] = time
return path, header, value, repeat
def _open_boundary_condition_idf(
block_content, variables, periods: Dict[str, datetime]
) -> Tuple[List[Dict[str, xr.DataArray]], List[datetime]]:
"""
Read the variables specified from block_content.
"""
n_system = block_content["n_system"]
n_time = len(block_content["time"])
n_total = n_system * n_time
das = [{} for _ in range(n_system)]
for variable in variables:
variable_content = block_content[variable]
n = len(variable_content)
if n != n_total:
raise ValueError(
f"Expected n_time * n_system = {n_time} * {n_system} = "
f"{n_total} entries for variable {variable}. Received: {n}"
)
# Group the paths and headers by system.
system_paths = defaultdict(list)
system_headers = defaultdict(list)
system_values = defaultdict(list)
all_repeats = set()
for i, entry in enumerate(variable_content):
path, header, value, repeat = _process_boundary_condition_entry(
entry, periods
)
header["name"] = variable
key = i % n_system
system_paths[key].append(path)
system_headers[key].append(header)
system_values[key].append(value)
if repeat:
all_repeats.add(repeat)
# Concat one system at a time.
for i, (paths, headers, values) in enumerate(
zip(system_paths.values(), system_headers.values(), system_values.values())
):
das[i][variable] = _create_dataarray(paths, headers, values, dim="time")
repeats = sorted(all_repeats)
return das, repeats
def _read_package_gen(
block_content: Dict[str, Any], has_topbot: bool
) -> List[Dict[str, Any]]:
out = []
for entry in block_content["gen"]:
gdf = imod.gen.read(entry["path"])
if has_topbot:
gdf["resistance"] = entry["factor"] * entry["addition"]
else:
gdf["multiplier"] = entry["factor"] * entry["addition"]
d = {
"geodataframe": gdf,
"layer": entry["layer"],
}
out.append(d)
return out
IPF_LOG_MESSAGE_TEMPLATE = """\
A well with the same x, y, id, filter_top and
filter_bot was already imported. This happened at x =
{row[1]}, y = {row[2]}, id = {path_assoc.stem}. Now the
ID for this new well was appended with the suffix _{suffix})
"""
@dataclass
class IpfResult:
has_associated: bool = data_field(default_factory=bool)
dataframe: list[pd.DataFrame] = data_field(default_factory=list)
layer: list[int] = data_field(default_factory=list)
time: list[str] = data_field(default_factory=list)
factor: list[float] = data_field(default_factory=list)
addition: list[float] = data_field(default_factory=list)
def append(
self,
dataframe: pd.DataFrame,
layer: int,
time: str,
factor: float,
addition: float,
):
self.dataframe.append(dataframe)
self.layer.append(layer)
self.time.append(time)
self.factor.append(factor)
self.addition.append(addition)
def _process_ipf_time(
entry: Dict[str, Any], periods: Dict[str, datetime], repeats: list[str]
) -> tuple[Optional[str], list[str]]:
timestring = entry["time"]
time = periods.get(timestring)
if time is None:
time = _process_time(timestring)
else:
repeats.append(time)
return time, repeats
def _prepare_df_unassociated(ipf_df: pd.DataFrame, layer: int) -> pd.DataFrame:
columns = ("x", "y", "rate")
if layer <= 0:
df = ipf_df.iloc[:, :5]
columns = columns + ("filt_top", "filt_bot")
else:
df = ipf_df.iloc[:, :3]
df.columns = columns
return df
def _prepare_df_associated(
ipf_df: pd.DataFrame,
imported_wells: dict[tuple, int],
indexcol: int,
path: Path,
ext: str,
) -> pd.DataFrame:
dfs = []
for row in ipf_df.itertuples():
filename = row[indexcol]
path_assoc = path.parent.joinpath(f"{filename}.{ext}")
well_characteristics_dict = {
"x": row[1],
"y": row[2],
"id": path_assoc.stem,
"filt_top": row[4],
"filt_bot": row[5],
}
df_assoc = _try_read_with_func(imod.ipf.read_associated, path_assoc).iloc[:, :2]
df_assoc.columns = ["time", "rate"]
df_assoc = df_assoc.assign(**well_characteristics_dict)
well_characteristics = tuple(well_characteristics_dict.values())
if well_characteristics not in imported_wells.keys():
imported_wells[well_characteristics] = 0
else:
suffix = imported_wells[well_characteristics] + 1
imported_wells[well_characteristics] = suffix
df_assoc["id"] = df_assoc["id"] + f"_{suffix}"
log_message = textwrap.dedent(
IPF_LOG_MESSAGE_TEMPLATE.format(
row=row, path_assoc=path_assoc, suffix=suffix
)
)
imod.logging.logger.log(
loglevel=LogLevel.WARNING,
message=log_message,
additional_depth=2,
)
dfs.append(df_assoc)
return pd.concat(dfs, ignore_index=True, sort=False)
def _read_package_ipf(
block_content: Dict[str, Any], periods: Dict[str, datetime]
) -> Tuple[Dict[str, Dict], List[datetime]]:
out = defaultdict(IpfResult)
repeats = []
# we will store in this set the tuples of (x, y, id, well_top, well_bot)
# which should be unique for each well
imported_wells = {}
for entry in block_content["ipf"]:
time, repeats = _process_ipf_time(entry, periods, repeats)
layer = entry["layer"]
factor = entry["factor"]
addition = entry["addition"]
# Ensure the columns are identifiable.
path = Path(entry["path"])
ipf_df, indexcol, ext = _try_read_with_func(imod.ipf._read_ipf, path)
if indexcol == 0:
# No associated files
has_associated = False
df = _prepare_df_unassociated(ipf_df, layer)
else:
has_associated = True
df = _prepare_df_associated(ipf_df, imported_wells, indexcol, path, ext)
df["rate"] = df["rate"] * factor + addition
out[path.stem].has_associated = has_associated
out[path.stem].append(df, layer, time, factor, addition)
out_dict_ls: dict[str, dict] = {key: asdict(o) for key, o in out.items()}
repeats = sorted(repeats)
return out_dict_ls, repeats
[docs]
def read_projectfile(path: FilePath) -> Dict[str, Any]:
"""
Read an iMOD project file into a collection of nested dictionaries.
The top-level keys are the "topic" entries such "bnd" or "riv" in the
project file. An example structure of the dictionaries is visualized below:
.. code-block::
content
├── bnd
│ ├── active: bool
│ └── ibound: list of dictionaries for each layer
├── riv
│ ├── active: bool
│ ├── conductance: list of dictionaries for each time and layer.
│ ├── stage: idem.
│ ├── bottom_elevation: idem.
│ └── infiltration_factor: idem.
etc.
Time and layer are flattened into a single list and time is included in
every dictionary:
.. code-block::
stage
├── 0 # First entry in list
│ ├── active: bool
│ ├── is_constant: bool
│ ├── layer: int
│ ├── factor: float
│ ├── addition: float
│ ├── constant: float
│ ├── path: str
│ └── time: str
│
├── 1 # Second entry in list
│ ├── etc.
etc.
Parameters
----------
path: str or Path
Returns
-------
content: Dict[str, Any]
"""
# Force to Path
path = Path(path)
with open(path) as f:
lines = f.readlines()
tokenized = []
for i, line in enumerate(lines):
try:
tokenized.append(_tokenize(line))
except Exception as e:
raise type(e)(f"{e}\nError occurred in line {i}")
lines = _LineIterator(tokenized)
content = {}
wdir = path.parent
# Change dir temporarely to projectfile dir to resolve relative paths
with imod.util.cd(wdir):
while not lines.finished:
_parse_block(lines, content)
return content
def _is_var_ipf_and_path(block_content: dict[str, Any], var: str):
block_item = block_content[var]
path = block_item[0]["path"]
is_ipf = path.suffix.lower() == ".ipf"
return is_ipf, path
[docs]
def open_projectfile_data(path: FilePath) -> tuple[dict[str, Any], dict[str, Any]]:
"""
Read the contents of an iMOD project file and read/open the data present in
it:
* IDF data is lazily loaded into xarray.DataArrays.
* GEN data is eagerly loaded into geopandas.GeoDataFrames
* IPF data is eagerly loaded into pandas.DataFrames
* Non-file based entries (such as the PCG settings) are kept as a dictionary.
When multiple systems are present, they are numbered starting from one, e.g.:
* drn-1
* drn-2
Xarray requires valid dates for the time coordinate. Aliases such as
"summer" and "winter" that are associated with dates in the project file
Periods block cannot be used in the time coordinate. Hence, this function
will instead insert the dates associated with the aliases, with the year
replaced by 1899; as the iMOD calendar starts at 1900, this ensures that
the repeats are always first and that no date collisions will occur.
Parameters
----------
path: pathlib.Path or str.
Returns
-------
data: Dict[str, Any]
Keys are the iMOD project file "topics", without parentheses.
"""
content = read_projectfile(path)
periods_block = content.pop("periods", None)
if periods_block is None:
periods = {}
else:
# Set the year of a repeat date to 1899: this ensures it falls outside
# of the iMOD calendar. Collisions are then always avoided.
periods = {
key: _process_time(time, yearfirst=False).replace(year=1899)
for key, time in periods_block.items()
}
# Pop species block, at the moment we do not do much with,
# since most regional models are without solute transport
content.pop("species", None)
has_topbot = "(top)" in content and "(bot)" in content
prj_data = {}
repeat_stress = {}
for key, block_content in content.items():
repeats = None
try:
if key == "(hfb)":
data = _read_package_gen(block_content, has_topbot)
elif key == "(wel)":
data, repeats = _read_package_ipf(block_content, periods)
elif key == "(cap)":
maybe_ipf_var = "artificial_recharge_layer"
is_ipf, maybe_ipf_path = _is_var_ipf_and_path(
block_content, maybe_ipf_var
)
# If its an ipf drop it and manually add it by calling the
# ipf.read function. If its not an ipf then its an idf, which
# doesnt need any special treatment.
if is_ipf:
block_content.pop(maybe_ipf_var)
variables = set(METASWAP_VARS).intersection(block_content.keys())
data = _open_package_idf(block_content, variables)
# Read and reattach ipf data to package data.
if is_ipf:
data[0][maybe_ipf_var] = imod.ipf.read(maybe_ipf_path)
elif key in ("extra", "(pcg)"):
data = [block_content]
elif key in KEYS:
variables = KEYS[key]
data = _open_package_idf(block_content, variables)
elif key in DATE_KEYS:
variables = DATE_KEYS[key]
data, repeats = _open_boundary_condition_idf(
block_content, variables, periods
)
else:
raise KeyError(f"Unsupported key: '{key}'")
except Exception as e:
raise type(e)(f"{e}. Errored while opening/reading data entries for: {key}")
strippedkey = key.strip("(").strip(")")
if strippedkey == "wel":
for key, d in data.items():
named_key = f"{strippedkey}-{key}"
prj_data[named_key] = d
repeat_stress[named_key] = repeats
elif len(data) > 1:
for i, da in enumerate(data):
numbered_key = f"{strippedkey}-{i + 1}"
prj_data[numbered_key] = da
repeat_stress[numbered_key] = repeats
else:
prj_data[strippedkey] = data[0]
repeat_stress[strippedkey] = repeats
repeat_stress = {k: v for k, v in repeat_stress.items() if v}
return prj_data, repeat_stress
[docs]
def read_timfile(path: FilePath) -> List[Dict]:
def parsetime(time: str) -> datetime:
# Check for steady-state:
if time == "00000000000000":
return None
return datetime.strptime(time, "%Y%m%d%H%M%S")
with open(path, "r") as f:
lines = f.readlines()
# A line contains 2, 3, or 4 values:
# time, isave, nstp, tmult
casters = {
"time": parsetime,
"save": lambda x: bool(int(x)),
"n_timestep": int,
"timestep_multiplier": float,
}
content = []
for line in lines:
stripped = line.strip()
if stripped == "":
continue
parts = stripped.split(",")
entry = {k: cast(s.strip()) for s, (k, cast) in zip(parts, casters.items())}
content.append(entry)
return content
