# -*- coding: utf-8 -*-
"""This module provides features for executing *HydPy* workflows based on XML
configuration files.
.. _HydPy release: https://github.com/hydpy-dev/hydpy/releases
.. _`OpenDA`: https://www.openda.org/
At the heart of module |xmltools| lies function |run_simulation|, thought to be applied
via a command line (see the documentation on script |hyd| for further information).
|run_simulation| expects that the *HydPy* project you want to work with is available in
your current working directory and contains an XML configuration file (as
`single_run.xml` in the example project folder `HydPy-H-Lahn`). This configuration
file must agree with the XML schema `HydPyConfigSingleRun.xsd`, which is available in
the `conf` subpackage and separately downloadable for each `HydPy release`_. If you
did implement new or changed existing models, you have to update this schema file.
*HydPy* does this automatically through its setup mechanism (see the documentation on
class |XSDWriter|).
To show how to apply |run_simulation| via a command line, we first copy the
`HydPy-H-Lahn` project into the `iotesting` folder by calling the function
|prepare_full_example_1|:
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
Running the simulation requires defining the main script (`hyd.py`), the function
specifying the actual workflow (`run_simulation`), the name of the project of interest
(`HydPy-H-Lahn`), and the name of the relevant XML configuration file
(`single_run.xml`). We pass the required text to function |run_subprocess| of module
subprocess to simulate using the command line:
>>> from hydpy import run_subprocess, TestIO
>>> import subprocess
>>> with TestIO(): # doctest: +ELLIPSIS
... result = run_subprocess("hyd.py run_simulation HydPy-H-Lahn single_run.xml")
Start HydPy project `HydPy-H-Lahn` (...).
Read configuration file `single_run.xml` (...).
Interpret the defined options (...).
Interpret the defined period (...).
Read all network files (...).
Activate the selected network (...).
Read the required control files (...).
Read the required condition files (...).
Read the required time series files (...).
Perform the simulation run (...).
Write the desired condition files (...).
Write the desired time series files (...).
As defined by the XML configuration file, the simulation started on the first and
ended on the sixth of January 1996. The following example shows the read initial
conditions and the written final conditions of sequence |hland_states.SM| for the
12 hydrological response units of the subcatchment `land_dill_assl`:
>>> with TestIO():
... filepath = "HydPy-H-Lahn/conditions/init_1996_01_01_00_00_00/land_dill_assl.py"
... with open(filepath) as file_:
... print("".join(file_.readlines()[10:12]))
... filepath = "HydPy-H-Lahn/conditions/init_1996_01_06/land_dill_assl.py"
... with open(filepath) as file_:
... print("".join(file_.readlines()[12:14]))
sm(185.13164, 181.18755, 199.80432, 196.55888, 212.04018, 209.48859,
222.12115, 220.12671, 230.30756, 228.70779, 236.91943, 235.64427)
<BLANKLINE>
sm(184.517818, 180.588472, 199.142925, 195.90995, 212.04018, 209.48859,
222.12115, 220.12671, 230.30756, 228.70779, 236.91943, 235.64427)
<BLANKLINE>
The intermediate soil moisture values have been stored in a NetCDF file called
`hland_96_state_sm.nc`:
>>> import numpy
>>> from hydpy import print_vector
>>> from hydpy.core.netcdftools import netcdf4, chars2str, query_variable
>>> with TestIO():
... ncfile = netcdf4.Dataset("HydPy-H-Lahn/series/soildata/hland_96_state_sm.nc")
... chars2str(query_variable(ncfile, "station_id")[:].data)[:3]
... print_vector(query_variable(ncfile, "hland_96_state_sm")[:, 0])
['land_dill_assl_0', 'land_dill_assl_1', 'land_dill_assl_2']
184.958475, 184.763638, 184.610776, 184.553224, 184.517818
>>> ncfile.close()
Spatially averaged time series values have been stored in files ending with the suffix
`_mean`:
>>> import time
>>> time.sleep(10)
>>> with TestIO(clear_all=True):
... print_vector(
... numpy.load("HydPy-H-Lahn/series/averages/lahn_marb_sim_q_mean.npy")[13:]
... )
9.64767, 8.513649, 7.777628, 7.343314, 7.156591
"""
# import...
# ...from standard library
from __future__ import annotations
import collections
import contextlib
import copy
import itertools
import os
import warnings
from xml.etree import ElementTree
# ...from HydPy
import hydpy
from hydpy import conf
from hydpy import config
from hydpy import models
from hydpy.core import devicetools
from hydpy.core import exceptiontools
from hydpy.core import hydpytools
from hydpy.core import importtools
from hydpy.core import itemtools
from hydpy.core import objecttools
from hydpy.core import selectiontools
from hydpy.core import parametertools
from hydpy.core import sequencetools
from hydpy.exe import commandtools
from hydpy.core.typingtools import *
if TYPE_CHECKING:
import netCDF4 as netcdf4
import xmlschema
from hydpy.core import modeltools
from hydpy.core import variabletools
else:
netcdf4 = exceptiontools.OptionalImport(
"netcdf4", ["netCDF4", "h5netcdf.legacyapi"], locals()
)
xmlschema = exceptiontools.OptionalImport("xmlschema", ["xmlschema"], locals())
_TypeSetOrAddOrMultiplyItem = TypeVar(
"_TypeSetOrAddOrMultiplyItem",
itemtools.SetItem,
itemtools.AddItem,
itemtools.MultiplyItem,
)
_TypeGetOrChangeItem = TypeVar(
"_TypeGetOrChangeItem", itemtools.GetItem, itemtools.ChangeItem, itemtools.SetItem
)
namespace = (
"{https://github.com/hydpy-dev/hydpy/releases/download/your-hydpy-version/"
"HydPyConfigBase.xsd}"
)
_ITEMGROUP2ITEMCLASS = {
"setitems": itemtools.SetItem,
"additems": itemtools.AddItem,
"multiplyitems": itemtools.MultiplyItem,
"getitems": itemtools.GetItem,
}
@overload
def find(
root: ElementTree.Element, name: str, optional: Literal[True] = True
) -> Optional[ElementTree.Element]:
"""Optional version of function |find|."""
@overload
def find(
root: ElementTree.Element, name: str, optional: Literal[False]
) -> ElementTree.Element:
"""Non-optional version of function |find|."""
[docs]
def find(
root: ElementTree.Element, name: str, optional: Literal[True, False] = True
) -> Optional[ElementTree.Element]:
"""Return the first XML element with the given name found in the given XML root.
>>> from hydpy.exe.xmltools import find, XMLInterface
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> find(interface.root, "timegrid").tag.endswith("timegrid")
True
By default, function |find| returns |None| in case the required element is missing:
>>> find(interface.root, "wrong")
Set the argument `optional` to |False| to let function |find| raise errors instead:
>>> find(interface.root, "wrong", optional=False)
Traceback (most recent call last):
...
AttributeError: The actual XML element `config` does not define a XML subelement \
named `wrong`. Please make sure your XML file follows the relevant XML schema.
"""
element = root.find(f"{namespace}{name}")
if element is None and not optional:
raise AttributeError(
f"The actual XML element `{root.tag.rsplit('}')[-1]}` does not define a "
f"XML subelement named `{name}`. Please make sure your XML file follows "
f"the relevant XML schema."
)
return element
def _query_selections(xmlelement: ElementTree.Element) -> selectiontools.Selections:
selections = []
text = xmlelement.text
assert text is not None
for name in text.split():
try:
selections.append(getattr(hydpy.pub.selections, name))
except AttributeError:
raise NameError(
f"The XML configuration file tries to define a selection using the "
f"text `{name}`, but the actual project does not handle such a "
f"`Selection` object."
) from None
return selectiontools.Selections(*selections)
[docs]
def strip(name: str) -> str:
"""Remove the XML namespace from the given string and return it.
>>> from hydpy.exe.xmltools import strip
>>> strip("{https://github.com/something.xsd}something")
'something'
"""
return name.split("}")[-1]
[docs]
class PrepareSeriesArguments(NamedTuple):
"""Helper class that determines and provides the arguments for function
|IOSequence.prepare_series|."""
allocate_ram: bool
read_jit: bool
write_jit: bool
[docs]
@classmethod
def from_xmldata(
cls, is_reader: bool, is_input: bool, prefer_ram: bool
) -> PrepareSeriesArguments:
"""Create a |PrepareSeriesArguments| object based on the (already prepared)
information of an XML file.
Meaning of the arguments:
* is_reader: is the current XML-Element responsible for reading (or writing)?
* is_input: serve the addressed sequences as inputs (or outputs)?
* prefer_ram: prefer to handle time series data in RAM (or read and write it
just in time)?
>>> from hydpy.exe.xmltools import PrepareSeriesArguments
>>> from_xmldata = PrepareSeriesArguments.from_xmldata
Test cases for reading input data:
>>> from_xmldata(is_reader=True, is_input=True, prefer_ram=True)
PrepareSeriesArguments(allocate_ram=True, read_jit=False, write_jit=False)
>>> from_xmldata(is_reader=True, is_input=True, prefer_ram=False)
PrepareSeriesArguments(allocate_ram=False, read_jit=True, write_jit=False)
Attempting to read output data results in an |AssertionError| (disallowed by
all available XML schema files):
>>> from_xmldata(is_reader=True, is_input=False, prefer_ram=True)
Traceback (most recent call last):
...
AssertionError: reading output values is disallowed
>>> from_xmldata(is_reader=True, is_input=False, prefer_ram=False)
Traceback (most recent call last):
...
AssertionError: reading output values is disallowed
Test cases for writing input data (this is for the rare case where an
external tool like `OpenDA`_ provides or modifies the input data in RAM, and we
want to write it to a file for documentation purposes):
>>> from_xmldata(is_reader=False, is_input=True, prefer_ram=True)
PrepareSeriesArguments(allocate_ram=True, read_jit=False, write_jit=False)
>>> from_xmldata(is_reader=False, is_input=True, prefer_ram=False)
PrepareSeriesArguments(allocate_ram=True, read_jit=False, write_jit=True)
Test cases for writing output data:
>>> from_xmldata(is_reader=False, is_input=False, prefer_ram=True)
PrepareSeriesArguments(allocate_ram=True, read_jit=False, write_jit=False)
>>> from_xmldata(is_reader=False, is_input=False, prefer_ram=False)
PrepareSeriesArguments(allocate_ram=False, read_jit=False, write_jit=True)
"""
is_writer = not is_reader
is_output = not is_input
prefer_jit = not prefer_ram
assert not (is_reader and is_output), "reading output values is disallowed"
return PrepareSeriesArguments(
allocate_ram=prefer_ram or (is_input and is_writer),
read_jit=is_reader and prefer_jit,
write_jit=is_writer and prefer_jit,
)
[docs]
def run_simulation(projectname: str, xmlfile: str) -> None:
"""Perform a *HydPy* workflow according to the given XML configuration file
available in the given project's directory.
Function |run_simulation| is a "script function". We explain its normal usage
in the main documentation on module |xmltools|.
"""
write = commandtools.print_textandtime
hydpy.pub.options.printprogress = False
write(f"Start HydPy project `{projectname}`")
hp = hydpytools.HydPy(projectname)
write(f"Read configuration file `{xmlfile}`")
interface = XMLInterface(xmlfile)
write("Interpret the defined options")
interface.update_options()
hydpy.pub.options.printprogress = False
write("Interpret the defined period")
interface.update_timegrids()
write("Read all network files")
interface.network_io.prepare_network()
write("Create the custom selections (if defined)")
interface.update_selections()
write("Activate the selected network")
hp.update_devices(selection=interface.fullselection, silent=True)
write("Read the required control files")
interface.control_io.prepare_models()
write("Read the required condition files")
interface.conditions_io.load_conditions()
write("Read the required time series files")
series_io = interface.series_io
series_io.prepare_series()
series_io.load_series()
write("Perform the simulation run")
with series_io.modify_inputdir(), series_io.modify_outputdir():
hp.simulate()
write("Write the desired condition files")
interface.conditions_io.save_conditions()
write("Write the desired time series files")
series_io.save_series()
[docs]
class XMLBase:
"""Base class for the concrete classes |XMLInterface|, |XMLConditions|,
|XMLSeries|, and |XMLSubseries|.
Subclasses of |XMLBase| support iterating XML subelements while skipping those
named `selections`:
>>> from hydpy.exe.xmltools import XMLInterface
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("multiple_runs.xml", make_filepath("HydPy-H-Lahn"))
>>> itemgroup = interface.exchange.itemgroups[1]
>>> for element in itemgroup:
... print(strip(element.tag))
hland_96
rconc_uh
>>> for element in itemgroup.models[0].subvars[0].vars[0]:
... print(strip(element.tag))
name
level
"""
root: ElementTree.Element
@property
def name(self) -> str:
"""Apply function |strip| to the root of the object of the |XMLBase| subclass.
>>> from hydpy.exe.xmltools import XMLInterface
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> interface.name
'config'
>>> interface.series_io.readers[0].name
'reader'
"""
return strip(self.root.tag)
@overload
def find(
self, name: str, optional: Literal[True] = True
) -> Optional[ElementTree.Element]:
"""Optional version of method |XMLBase.find|."""
@overload
def find(self, name: str, optional: Literal[False]) -> ElementTree.Element:
"""Non-optional version of function |XMLBase.find|."""
[docs]
def find(
self, name: str, optional: Literal[True, False] = True
) -> Optional[ElementTree.Element]:
"""Apply function |find| to the root of the object of the |XMLBase| subclass.
>>> from hydpy.exe.xmltools import XMLInterface
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> interface.find("timegrid").tag.endswith("timegrid")
True
>>> interface.find("wrong")
>>> interface.find("wrong", optional=False)
Traceback (most recent call last):
...
AttributeError: The actual XML element `config` does not define a XML \
subelement named `wrong`. Please make sure your XML file follows the relevant XML \
schema.
"""
return find(self.root, name, optional)
def __iter__(self) -> Iterator[ElementTree.Element]:
for element in self.root:
name = strip(element.tag)
if name != "selections":
yield element
[docs]
class XMLInterface(XMLBase):
"""An interface to XML configuration files that are valid concerning schema file
`HydPyConfigSingleRun.xsd` or `HydPyConfigMultipleRuns.xsd`.
>>> from hydpy.exe.xmltools import XMLInterface
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> interface.root.tag
'{https://github.com/hydpy-dev/hydpy/releases/download/your-hydpy-version/\
HydPyConfigSingleRun.xsd}config'
>>> interface = XMLInterface('multiple_runs.xml', make_filepath('HydPy-H-Lahn'))
>>> interface.root.tag
'{https://github.com/hydpy-dev/hydpy/releases/download/your-hydpy-version/\
HydPyConfigMultipleRuns.xsd}config'
>>> XMLInterface('wrongfilepath.xml', 'wrongdir') # doctest: +ELLIPSIS
Traceback (most recent call last):
...
FileNotFoundError: While trying to parse the XML configuration file \
...wrongfilepath.xml, the following error occurred: \
[Errno 2] No such file or directory: '...wrongfilepath.xml'
"""
def __init__(self, filename: str, directory: Optional[str] = None) -> None:
if directory is None:
directory = hydpy.pub.projectname
self.filepath = os.path.abspath(os.path.join(directory, filename))
try:
self.root = ElementTree.parse(self.filepath).getroot()
except BaseException:
objecttools.augment_excmessage(
f"While trying to parse the XML configuration file " f"{self.filepath}"
)
[docs]
def validate_xml(self) -> None:
"""Raise an error if the actual XML does not agree with one of the available
schema files.
The first example relies on a distorted version of the configuration file
`single_run.xml`:
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import TestIO, xml_replace
>>> from hydpy.exe.xmltools import XMLInterface
>>> import os
>>> with TestIO(): # doctest: +ELLIPSIS
... xml_replace("HydPy-H-Lahn/single_run",
... firstdate="1996-01-32T00:00:00")
template file: HydPy-H-Lahn/single_run.xmlt
target file: HydPy-H-Lahn/single_run.xml
replacements:
firstdate --> 1996-01-32T00:00:00 (given argument)
prefix --> init (default argument)
zip_ --> false (default argument)
>>> with TestIO():
... interface = XMLInterface("single_run.xml", "HydPy-H-Lahn")
>>> interface.validate_xml() # doctest: +ELLIPSIS
Traceback (most recent call last):
...
xmlschema.validators.exceptions.XMLSchemaDecodeError: While trying to \
validate XML file `...single_run.xml`, the following error occurred: failed \
validating '1996-01-32T00:00:00' with XsdAtomicBuiltin(name='xs:dateTime')...
...
Reason: day is out of range for month
...
Schema component:
...
Instance:
...
<firstdate xmlns="https://github.com/hydpy-dev/hydpy/releases/\
download/your-hydpy-version/HydPyConfigBase.xsd">1996-01-32T00:00:00</firstdate>
...
Path: /hpcsr:config/timegrid/firstdate
...
In the second example, we examine a correct configuration file:
>>> with TestIO(): # doctest: +ELLIPSIS
... xml_replace("HydPy-H-Lahn/single_run")
... interface = XMLInterface("single_run.xml", "HydPy-H-Lahn")
template file: HydPy-H-Lahn/single_run.xmlt
target file: HydPy-H-Lahn/single_run.xml
replacements:
firstdate --> 1996-01-01T00:00:00 (default argument)
prefix --> init (default argument)
zip_ --> false (default argument)
>>> interface.validate_xml()
The XML configuration file must correctly refer to the corresponding schema
file:
>>> with TestIO():
... with open("HydPy-H-Lahn/single_run.xml") as xmlfile:
... text = xmlfile.read()
... text = text.replace("HydPyConfigSingleRun.xsd", "wrong.xsd")
... with open("HydPy-H-Lahn/single_run.xml", "w") as xmlfile:
... _ = xmlfile.write(text)
... interface = XMLInterface("single_run.xml", "HydPy-H-Lahn")
>>> interface.validate_xml() # doctest: +ELLIPSIS
Traceback (most recent call last):
...
RuntimeError: While trying to validate XML file `...single_run.xml`, \
the following error occurred: Configuration file `single_run.xml` does not \
correctly refer to one of the available XML schema files \
(HydPyConfigSingleRun.xsd and HydPyConfigMultipleRuns.xsd).
XML files based on `HydPyConfigMultipleRuns.xsd` can be validated as well:
>>> with TestIO():
... interface = XMLInterface("multiple_runs.xml", "HydPy-H-Lahn")
>>> interface.validate_xml()
"""
try:
filenames = ("HydPyConfigSingleRun.xsd", "HydPyConfigMultipleRuns.xsd")
for name in filenames:
if name in self.root.tag:
schemafile = name
break
else:
raise RuntimeError(
f"Configuration file `{os.path.split(self.filepath)[-1]}` does "
f"not correctly refer to one of the available XML schema files "
f"({objecttools.enumeration(filenames)})."
)
confpath: str = conf.__path__[0]
schemapath = os.path.join(confpath, schemafile)
schema = xmlschema.XMLSchema(schemapath)
schema.validate(self.filepath)
except BaseException:
objecttools.augment_excmessage(
f"While trying to validate XML file `{self.filepath}`"
)
[docs]
def update_options(self) -> None:
"""Update the |Options| object available in the |pub| module with the values
defined in the `options` XML element.
.. testsetup::
>>> from hydpy import pub
>>> del pub.timegrids
>>> del pub.options.simulationstep
>>> from hydpy.exe.xmltools import XMLInterface
>>> from hydpy import pub
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> pub.options.ellipsis = 0
>>> pub.options.parameterstep = "1h"
>>> pub.options.printprogress = True
>>> pub.options.reprdigits = -1
>>> pub.options.utcoffset = -60
>>> pub.options.timestampleft = False
>>> pub.options.warnsimulationstep = 0
>>> interface.update_options()
>>> pub.options
Options(
checkseries -> TRUE
ellipsis -> 0
parameterstep -> Period("1d")
printprogress -> FALSE
reprdigits -> 6
simulationstep -> Period()
timestampleft -> TRUE
trimvariables -> TRUE
usecython -> TRUE
usedefaultvalues -> FALSE
utclongitude -> 15
utcoffset -> 60
warnmissingcontrolfile -> FALSE
warnmissingobsfile -> TRUE
warnmissingsimfile -> TRUE
warnsimulationstep -> FALSE
warntrim -> TRUE
)
>>> pub.options.printprogress = False
>>> pub.options.reprdigits = 6
"""
options = hydpy.pub.options
for option in self.find("options", optional=False):
value = option.text
if value in ("true", "false"):
setattr(options, strip(option.tag), value == "true")
else:
setattr(options, strip(option.tag), value)
options.printprogress = False
[docs]
def update_timegrids(self) -> None:
"""Update the |Timegrids| object available in the |pub| module with the values
defined in the `timegrid` XML element.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO
>>> from hydpy.exe.xmltools import XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> with TestIO():
... XMLInterface("single_run.xml").update_timegrids()
>>> pub.timegrids
Timegrids("1996-01-01T00:00:00",
"1996-01-06T00:00:00",
"1d")
"""
timegrid_xml = self.find("timegrid", optional=False)
firstdate = timegrid_xml[0].text
assert firstdate is not None
lastdate = timegrid_xml[1].text
assert lastdate is not None
stepsize = timegrid_xml[2].text
assert stepsize is not None
hydpy.pub.timegrids = (firstdate, lastdate, stepsize)
[docs]
def update_selections(self) -> None:
"""Create |Selection| objects based on the `add_selections` XML element and
add them to the |Selections| object available in module |pub|.
The `Lahn` example project comes with four selections:
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO
>>> from hydpy.exe.xmltools import find, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> with TestIO():
... hp.prepare_network()
... interface = XMLInterface("single_run.xml")
>>> pub.selections
Selections("complete", "headwaters", "nonheadwaters", "streams")
Following the definitions of the `add_selections` element of the configuration
file `single_run.xml`, method |XMLInterface.update_selections| creates three
additional selections based on given device names, keywords, or selection
names (you could also combine these subelements):
>>> interface.update_selections()
>>> pub.selections
Selections("complete", "from_devices", "from_keywords",
"from_selections", "headwaters", "nonheadwaters", "streams")
>>> pub.selections.from_devices
Selection("from_devices",
nodes=(),
elements=("land_lahn_leun", "land_lahn_marb"))
>>> pub.selections.from_keywords
Selection("from_keywords",
nodes=(),
elements=("land_dill_assl", "land_lahn_kalk",
"land_lahn_leun", "land_lahn_marb"))
>>> pub.selections.from_selections
Selection("from_selections",
nodes=("dill_assl", "lahn_kalk", "lahn_leun", "lahn_marb"),
elements=("land_dill_assl", "land_lahn_kalk",
"land_lahn_leun", "land_lahn_marb",
"stream_dill_assl_lahn_leun",
"stream_lahn_leun_lahn_kalk",
"stream_lahn_marb_lahn_leun"))
Defining wrong device names, keywords, or selection names results in error
messages:
>>> add_selections = find(interface.root, "add_selections")
>>> add_selections[2][1].text = "streams no_selection"
>>> interface.update_selections()
Traceback (most recent call last):
...
RuntimeError: The XML configuration file tried to add the devices of a \
selection named `no_selection` to the custom selection `from_selections` but none \
of the available selections has this name.
>>> add_selections[1][1].text = "catchment no_keyword"
>>> interface.update_selections()
Traceback (most recent call last):
...
RuntimeError: The XML configuration file tried to add at least one device \
based on the keyword `no_keyword` to the custom selection `from_keywords` but none \
of the available devices has this keyword.
>>> add_selections[0][1].text = "dill_assl no_device"
>>> interface.update_selections()
Traceback (most recent call last):
...
RuntimeError: The XML configuration file tried to add a device named \
`no_device` to the custom selection `from_devices` but none of the available \
devices has this name.
"""
def _get_texts(root: ElementTree.Element, name: str) -> list[str]:
xmlelement = find(root=root, name=name, optional=True)
if xmlelement is None or xmlelement.text is None:
return []
return xmlelement.text.split()
elements = hydpy.pub.selections.complete.elements
nodes = hydpy.pub.selections.complete.nodes
add_selections = find(self.root, "add_selections", optional=True)
if add_selections is not None:
for add_selection in add_selections:
name_new_selection = find(add_selection, "name", optional=False).text
assert name_new_selection is not None
new_selection = selectiontools.Selection(name_new_selection)
for name_device in _get_texts(add_selection, "devices"):
try:
element = elements[name_device]
new_selection.elements.add_device(element)
except KeyError:
try:
node = nodes[name_device]
new_selection.nodes.add_device(node)
except KeyError:
raise RuntimeError(
f"The XML configuration file tried to add a device "
f"named `{name_device}` to the custom selection "
f"`{name_new_selection}` but none of the available "
f"devices has this name."
) from None
for keyword in _get_texts(add_selection, "keywords"):
nmb_devices = len(new_selection)
new_selection.elements += elements.search_keywords(keyword)
new_selection.nodes += nodes.search_keywords(keyword)
if len(new_selection) == nmb_devices:
raise RuntimeError(
f"The XML configuration file tried to add at least one "
f"device based on the keyword `{keyword}` to the custom "
f"selection `{name_new_selection}` but none of the "
f"available devices has this keyword."
) from None
for name_old_selection in _get_texts(add_selection, "selections"):
try:
new_selection += hydpy.pub.selections[name_old_selection]
except KeyError:
raise RuntimeError(
f"The XML configuration file tried to add the devices of "
f"a selection named `{name_old_selection}` to the custom "
f"selection `{name_new_selection}` but none of the "
f"available selections has this name."
) from None
hydpy.pub.selections += new_selection
@property
def selections(self) -> selectiontools.Selections:
"""The |Selections| object defined on the main level of the actual XML file.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> with TestIO():
... hp.prepare_network()
... interface = XMLInterface("single_run.xml")
>>> interface.update_selections()
>>> interface.find("selections").text = "headwaters streams"
>>> selections = interface.selections
>>> for selection in selections:
... print(selection.name)
headwaters
streams
>>> selections.headwaters
Selection("headwaters",
nodes=("dill_assl", "lahn_marb"),
elements=("land_dill_assl", "land_lahn_marb"))
>>> interface.find("selections").text = "head_waters"
>>> interface.selections
Traceback (most recent call last):
...
NameError: The XML configuration file tries to define a selection using the \
text `head_waters`, but the actual project does not handle such a `Selection` object.
"""
return _query_selections(self.find("selections", optional=False))
@property
def elements(self) -> Iterator[devicetools.Element]:
"""Yield all |Element| objects returned by |XMLInterface.selections| without
duplicates.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> with TestIO():
... hp.prepare_network()
... interface = XMLInterface("single_run.xml")
>>> interface.update_timegrids()
>>> interface.update_selections()
>>> interface.find("selections").text = "headwaters streams"
>>> for element in interface.elements:
... print(element.name)
land_dill_assl
land_lahn_marb
stream_dill_assl_lahn_leun
stream_lahn_leun_lahn_kalk
stream_lahn_marb_lahn_leun
"""
selections = copy.copy(self.selections)
elements: set[devicetools.Element] = set()
for selection in selections:
for element in selection.elements:
if element not in elements:
elements.add(element)
yield element
@property
def fullselection(self) -> selectiontools.Selection:
"""A |Selection| object that contains all |Element| and |Node| objects defined
by |XMLInterface.selections|.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> with TestIO():
... hp.prepare_network()
... interface = XMLInterface("single_run.xml")
>>> interface.update_selections()
>>> interface.find("selections").text = "nonheadwaters"
>>> interface.fullselection
Selection("fullselection",
nodes=("lahn_kalk", "lahn_leun"),
elements=("land_lahn_kalk", "land_lahn_leun"))
>>> interface.find("selections").text = "from_keywords"
>>> interface.fullselection
Selection("fullselection",
nodes=(),
elements=("land_dill_assl", "land_lahn_kalk",
"land_lahn_leun", "land_lahn_marb"))
"""
fullselection = selectiontools.Selection("fullselection")
for selection in self.selections:
fullselection += selection
return fullselection
@property
def network_io(self) -> Union[XMLNetworkDefault, XMLNetworkUserDefined]:
"""The `network_io` element defined in the actual XML file.
>>> from hydpy.exe.xmltools import XMLInterface, strip
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> interface.network_io.text
'default'
>>> interface = XMLInterface("multiple_runs.xml", make_filepath("HydPy-H-Lahn"))
>>> interface.network_io.text
'default'
"""
network_io = self.find("network_io", optional=True)
if network_io is None:
return XMLNetworkDefault(self, text="default")
return XMLNetworkUserDefined(self, network_io, text=network_io.text)
@property
def control_io(self) -> Union[XMLControlDefault, XMLControlUserDefined]:
"""The `control_io` element defined in the actual XML file.
>>> from hydpy.exe.xmltools import XMLInterface, strip
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> interface.control_io.text
'default'
>>> interface = XMLInterface("multiple_runs.xml", make_filepath("HydPy-H-Lahn"))
>>> interface.control_io.text
'default'
"""
control_io = self.find("control_io", optional=True)
if control_io is None:
return XMLControlDefault(self, text="default")
return XMLControlUserDefined(self, control_io, text=control_io.text)
@property
def conditions_io(self) -> XMLConditions:
"""The `condition_io` element defined in the actual XML file.
>>> from hydpy.exe.xmltools import XMLInterface, strip
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> strip(interface.series_io.root.tag)
'series_io'
"""
return XMLConditions(self, self.find("conditions_io", optional=False))
@property
def series_io(self) -> XMLSeries:
"""The `series_io` element defined in the actual XML file.
>>> from hydpy.exe.xmltools import XMLInterface, strip
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> strip(interface.series_io.root.tag)
'series_io'
"""
return XMLSeries(self, self.find("series_io", optional=False))
@property
def exchange(self) -> XMLExchange:
"""The `exchange` element defined in the actual XML file.
>>> from hydpy.exe.xmltools import XMLInterface, strip
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface(
... "multiple_runs.xml", make_filepath("HydPy-H-Lahn"))
>>> strip(interface.exchange.root.tag)
'exchange'
"""
return XMLExchange(self, self.find("exchange", optional=False))
[docs]
class XMLNetworkBase:
"""Base class for |XMLNetworkDefault| and |XMLNetworkUserDefined|."""
master: XMLInterface
text: Optional[str]
[docs]
def prepare_network(self) -> None:
"""Prepare the |Selections| object available in the global |pub| module:
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import attrready, HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... interface = XMLInterface("single_run.xml")
... interface.find("network_io").text = "wrong"
... interface.network_io.prepare_network() # doctest: +ELLIPSIS
Traceback (most recent call last):
...
RuntimeError: The directory `...wrong` does not contain any network files.
>>> with TestIO():
... interface = XMLInterface("single_run.xml")
... interface.find("network_io").text = "default"
... interface.network_io.prepare_network() # doctest: +ELLIPSIS
>>> pub.selections
Selections("complete", "headwaters", "nonheadwaters", "streams")
"""
if self.text:
hydpy.pub.networkmanager.currentdir = str(self.text)
hydpy.pub.selections = hydpy.pub.networkmanager.load_files()
[docs]
class XMLNetworkDefault(XMLNetworkBase):
"""Helper class for |XMLInterface| responsible for loading devices from network
files when the XML file does not specify a network directory."""
def __init__(self, master: XMLInterface, text: Optional[str]) -> None:
self.master: XMLInterface = master
self.text: Optional[str] = text
[docs]
class XMLNetworkUserDefined(XMLBase, XMLNetworkBase):
"""Helper class for |XMLInterface| responsible for loading devices from network
files when the XML file specifies a network directory."""
def __init__(
self, master: XMLInterface, root: ElementTree.Element, text: Optional[str]
) -> None:
self.master: XMLInterface = master
self.root: ElementTree.Element = root
self.text: Optional[str] = text
[docs]
class XMLControlBase:
"""Base class for |XMLControlDefault| and |XMLControlUserDefined|."""
master: XMLInterface
text: Optional[str]
[docs]
def prepare_models(self) -> None:
"""Prepare the |Model| objects of all |Element| objects returned by
|XMLInterface.elements|:
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import attrready, HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_network()
... interface = XMLInterface("single_run.xml")
... interface.update_selections()
... interface.find("selections").text = "headwaters"
... interface.control_io.prepare_models()
>>> interface.update_timegrids()
>>> hp.elements.land_lahn_marb.model.parameters.control.alpha
alpha(1.0)
>>> attrready(hp.elements.land_lahn_leun, "model")
False
"""
if self.text:
hydpy.pub.controlmanager.currentdir = str(self.text)
for element in self.master.elements:
element.prepare_model()
[docs]
class XMLControlDefault(XMLControlBase):
"""Helper class for |XMLInterface| responsible for loading models from control
files when the XML file does not specify a control directory."""
def __init__(self, master: XMLInterface, text: Optional[str]) -> None:
self.master: XMLInterface = master
self.text: Optional[str] = text
[docs]
class XMLControlUserDefined(XMLBase, XMLControlBase):
"""Helper class for |XMLInterface| responsible for loading models from control
files when the XML file specifies a control directory."""
def __init__(
self, master: XMLInterface, root: ElementTree.Element, text: Optional[str]
) -> None:
self.master: XMLInterface = master
self.root: ElementTree.Element = root
self.text: Optional[str] = text
[docs]
class XMLConditions(XMLBase):
"""Helper class for |XMLInterface| responsible for loading and saving initial
conditions."""
def __init__(self, master: XMLInterface, root: ElementTree.Element) -> None:
self.master: XMLInterface = master
self.root: ElementTree.Element = root
def _determine_currentdir(
self, currentdir: Optional[str], type_: Literal["input", "output"], /
) -> str:
if currentdir is not None:
return currentdir
if (inputdir := self.find(f"{type_}dir")) is not None:
return str(inputdir.text)
conditionmanager = hydpy.pub.conditionmanager
prefix = None if (p := self.find("prefix")) is None else str(p.text)
with conditionmanager.prefix(prefix):
return getattr(conditionmanager, f"{type_}path")
[docs]
def load_conditions(self, currentdir: Optional[str] = None) -> None:
"""Load the condition files of the |Model| objects of all |Element| objects
returned by |XMLInterface.elements|:
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_network()
... hp.prepare_models()
... interface = XMLInterface("single_run.xml")
... interface.update_selections()
... interface.find("selections").text = "headwaters"
... interface.conditions_io.load_conditions()
>>> hp.elements.land_lahn_marb.model.sequences.states.lz
lz(8.18711)
>>> hp.elements.land_lahn_leun.model.sequences.states.lz
lz(nan)
>>> from hydpy import xml_replace
>>> with TestIO():
... xml_replace("HydPy-H-Lahn/single_run", printflag=False, prefix="condi")
... interface = XMLInterface("single_run.xml")
... interface.update_selections()
... interface.conditions_io.load_conditions() # doctest: +ELLIPSIS
Traceback (most recent call last):
...
FileNotFoundError: While trying to load the initial conditions of element \
`land_dill_assl`, the following error occurred: [Errno 2] No such file or directory: \
'...condi_1996_01_01_00_00_00...land_dill_assl.py'
"""
cm = hydpy.pub.conditionmanager
try:
cm.currentdir = self._determine_currentdir(currentdir, "input")
for element in self.master.elements:
element.model.load_conditions()
finally:
cm.currentdir = None # type: ignore[assignment]
[docs]
def save_conditions(self, currentdir: Optional[str] = None) -> None:
"""Save the condition files of the |Model| objects of all |Element| objects
returned by |XMLInterface.elements|:
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> import os
>>> from hydpy import HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_network()
... hp.prepare_models()
... hp.elements.land_dill_assl.model.sequences.states.lz = 999.0
... interface = XMLInterface("single_run.xml")
... interface.update_selections()
... interface.find("selections").text = "headwaters"
... interface.conditions_io.save_conditions()
... dirpath = "HydPy-H-Lahn/conditions/init_1996_01_06"
... with open(os.path.join(dirpath, "land_dill_assl.py")) as file_:
... print(file_.readlines()[12].strip())
... os.path.exists(os.path.join(dirpath, "land_lahn_leun.py"))
lz(999.0)
False
>>> from hydpy import xml_replace
>>> with TestIO():
... xml_replace("HydPy-H-Lahn/single_run", printflag=False, zip_="true")
... interface = XMLInterface("single_run.xml")
... interface.find("selections").text = "headwaters"
... os.path.exists("HydPy-H-Lahn/conditions/init_1996_01_06.zip")
... interface.conditions_io.save_conditions()
... os.path.exists("HydPy-H-Lahn/conditions/init_1996_01_06.zip")
False
True
"""
cm = hydpy.pub.conditionmanager
try:
cm.currentdir = self._determine_currentdir(currentdir, "output")
for element in self.master.elements:
element.model.save_conditions()
if (zip_ := self.find("zip")) is not None:
zip__ = str(zip_.text)
if objecttools.value2bool(zip__, zip__):
cm.zip_currentdir()
finally:
cm.currentdir = None # type: ignore[assignment]
[docs]
class XMLSeries(XMLBase):
"""Helper class for |XMLInterface| responsible for loading and saving time series
data, which is further delegated to suitable instances of class |XMLSubseries|."""
def __init__(self, master: XMLInterface, root: ElementTree.Element) -> None:
self.master: XMLInterface = master
self.root: ElementTree.Element = root
@property
def readers(self) -> list[XMLSubseries]:
"""The reader XML elements defined in the actual XML file.
>>> from hydpy.exe.xmltools import XMLInterface
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> for reader in interface.series_io.readers:
... print(reader.info)
all input data
"""
return [XMLSubseries(self, _) for _ in self.find("readers", optional=False)]
@property
def writers(self) -> list[XMLSubseries]:
"""The writer XML elements defined in the actual XML file.
>>> from hydpy.exe.xmltools import XMLInterface
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> for writer in interface.series_io.writers:
... print(writer.info)
precipitation
soilmoisture
averaged
"""
return [XMLSubseries(self, _) for _ in self.find("writers", optional=False)]
[docs]
def prepare_series(self) -> None:
"""Call |XMLSubseries.prepare_series| of all |XMLSubseries| objects."""
for ioseries in itertools.chain(self.readers, self.writers):
ioseries.prepare_series()
[docs]
def load_series(self, currentdir: Optional[str] = None) -> None:
"""Call |XMLSubseries.load_series| of all |XMLSubseries| objects handled as
"readers"."""
for reader in self.readers:
reader.load_series(currentdir)
[docs]
def save_series(self, currentdir: Optional[str] = None) -> None:
"""Call |XMLSubseries.load_series| of all |XMLSubseries| objects handled as
"writers"."""
for writer in self.writers:
writer.save_series(currentdir)
[docs]
@contextlib.contextmanager
def modify_outputdir(self, currentdir: Optional[str] = None) -> Iterator[None]:
"""Temporarily modify the |IOSequence.dirpath| of all |IOSequence| objects
registered for writing time series data "just in time" during a simulation
run."""
try:
for writer in self.writers:
writer.change_dirpath(currentdir)
yield
finally:
for writer in self.writers:
writer.reset_dirpath()
[docs]
class XMLSelector(XMLBase):
"""Base class for |XMLSubseries| and |XMLVar| responsible for querying the relevant
|Node| and |Element| objects."""
master: XMLBase
root: ElementTree.Element
@property
def selections(self) -> selectiontools.Selections:
"""The |Selections| object defined for the respective respective IO series or
exchange item elements of the actual XML file.
Property |XMLSelector.selections| of class |XMLSelector| falls back to the
general property |XMLInterface.selections| of |XMLInterface| if the relevant IO
series or exchange item element does not define a unique selection:
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_network()
... hp.prepare_models()
... interface = XMLInterface("single_run.xml")
>>> interface.update_selections()
>>> series_io = interface.series_io
>>> for seq in (series_io.readers + series_io.writers):
... print(seq.info, seq.selections.names)
all input data ('from_keywords',)
precipitation ('headwaters', 'from_devices')
soilmoisture ('complete',)
averaged ('from_selections',)
If property |XMLSelector.selections| does not find any definitions, it raises
the following error:
>>> interface.root.remove(interface.find("selections"))
>>> series_io = interface.series_io
>>> for seq in (series_io.readers + series_io.writers):
... print(seq.info, seq.selections.names)
Traceback (most recent call last):
...
AttributeError: Unable to find a XML element named "selections". Please make \
sure your XML file follows the relevant XML schema.
"""
selections = self.find("selections")
master: Optional[XMLBase] = self
while selections is None:
master = getattr(master, "master", None)
if master is None:
raise AttributeError(
'Unable to find a XML element named "selections". Please make '
"sure your XML file follows the relevant XML schema."
)
selections = master.find("selections")
return _query_selections(selections)
@overload
def _get_devices(self, attr: Literal["nodes"]) -> Iterator[devicetools.Node]:
"""Extract all nodes."""
@overload
def _get_devices(self, attr: Literal["elements"]) -> Iterator[devicetools.Element]:
"""Extract all elements."""
def _get_devices(
self, attr: Literal["nodes", "elements"]
) -> Union[Iterator[devicetools.Node], Iterator[devicetools.Element]]:
"""Extract all nodes or elements."""
selections = copy.copy(self.selections)
devices = set()
for selection in selections:
for device in getattr(selection, attr):
if device not in devices:
devices.add(device)
yield device
@property
def elements(self) -> Iterator[devicetools.Element]:
"""Return the |Element| objects selected by the actual IO series or exchange
item element.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> with TestIO():
... hp.prepare_network()
... interface = XMLInterface("single_run.xml")
>>> interface.update_selections()
>>> for element in interface.series_io.writers[0].elements:
... print(element.name)
land_dill_assl
land_lahn_marb
land_lahn_leun
"""
return self._get_devices("elements")
@property
def nodes(self) -> Iterator[devicetools.Node]:
"""Return the |Node| objects selected by the actual IO series or exchange item
element.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> with TestIO():
... hp.prepare_network()
... interface = XMLInterface("single_run.xml")
>>> interface.update_selections()
>>> for node in interface.series_io.writers[0].nodes:
... print(node.name)
dill_assl
lahn_marb
"""
return self._get_devices("nodes")
[docs]
class XMLSubseries(XMLSelector):
"""Helper class for |XMLSeries| responsible for loading and saving time series
data."""
def __init__(self, master: XMLSeries, root: ElementTree.Element) -> None:
self.master: XMLSeries = master
self.root: ElementTree.Element = root
@property
def info(self) -> str:
"""Info attribute of the actual XML `reader` or `writer` element."""
return self.root.attrib["info"]
@property
def _is_reader(self) -> bool:
if self.name == "reader":
return True
if self.name == "writer":
return False
assert False
@property
def _is_writer(self) -> bool:
return not self._is_reader
[docs]
def prepare_sequencemanager(self, currentdir: Optional[str] = None) -> None:
"""Configure the |SequenceManager| object available in module |pub| following
the definitions of the actual XML `reader` or `writer` element when available;
if not, use those of the XML `series_io` element or fall back to the default.
Compare the following results with `single_run.xml` to see that the first
`writer` element re-defines the default file type (`asc`), that the second
`writer` element defines an alternative file type (`npy`), and that the third
`writer` relies on the general file type. The base mechanism is the same
for other options, e.g. the aggregation mode.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, TestIO, XMLInterface, pub
>>> hp = HydPy("HydPy-H-Lahn")
>>> with TestIO():
... hp.prepare_network()
... interface = XMLInterface("single_run.xml")
>>> series_io = interface.series_io
>>> with TestIO():
... series_io.writers[0].prepare_sequencemanager()
... pub.sequencemanager.currentdir
'default'
>>> pub.sequencemanager.filetype
'asc'
>>> pub.sequencemanager.overwrite
TRUE
>>> pub.sequencemanager.aggregation
'none'
>>> pub.sequencemanager.convention
'model-specific'
>>> with TestIO():
... series_io.writers[1].prepare_sequencemanager()
... pub.sequencemanager.currentdir
'soildata'
>>> pub.sequencemanager.filetype
'nc'
>>> pub.sequencemanager.overwrite
FALSE
>>> pub.sequencemanager.aggregation
'none'
>>> pub.sequencemanager.convention
'model-specific'
>>> with TestIO():
... series_io.writers[2].prepare_sequencemanager()
... pub.sequencemanager.currentdir
'averages'
>>> pub.sequencemanager.filetype
'npy'
>>> pub.sequencemanager.aggregation
'mean'
>>> pub.sequencemanager.overwrite
TRUE
>>> pub.sequencemanager.aggregation
'mean'
>>> pub.sequencemanager.convention
'model-specific'
"""
sm = hydpy.pub.sequencemanager
if currentdir is None:
element = self.find("directory")
if element is None:
element = self.master.find("directory")
if element is None:
sm.currentdir = "default"
else:
assert element.text is not None
sm.currentdir = element.text
else:
sm.currentdir = currentdir
for option in ("filetype", "aggregation", "convention", "overwrite"):
delattr(sm, option)
for element in (self.find(option), self.master.find(option)):
if element is not None:
assert element.text is not None
if option == "overwrite":
sm.overwrite = objecttools.value2bool(option, element.text)
else:
setattr(sm, option, element.text)
break
@property
def _ramflag(self) -> bool:
"""A flag analoge to the |IOSequence.ramflag| of class |IOSequence|."""
mode = self.find("mode")
if mode is None:
mode = self.master.find("mode")
if mode is None:
return True
assert mode.text is not None
return mode.text == "ram"
@property
def model2subs2seqs(
self,
) -> collections.defaultdict[str, collections.defaultdict[str, list[str]]]:
"""A nested |collections.defaultdict| containing the model-specific information
provided by the XML `sequences` element.
>>> from hydpy.exe.xmltools import XMLInterface
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> series_io = interface.series_io
>>> model2subs2seqs = series_io.writers[2].model2subs2seqs
>>> for model, subs2seqs in sorted(model2subs2seqs.items()):
... for subs, seq in sorted(subs2seqs.items()):
... print(model, subs, seq)
hland_96 factors ['tc']
hland_96 fluxes ['tf']
hland_96 states ['sm']
musk_classic states ['discharge']
"""
model2subs2seqs: collections.defaultdict[
str, collections.defaultdict[str, list[str]]
] = collections.defaultdict(lambda: collections.defaultdict(list))
for model in self.find("sequences", optional=False):
model_name = strip(model.tag)
if model_name == "node":
continue
for group in model:
group_name = strip(group.tag)
for sequence in group:
seq_name = strip(sequence.tag)
model2subs2seqs[model_name][group_name].append(seq_name)
return model2subs2seqs
@property
def subs2seqs(self) -> dict[str, list[str]]:
"""A |collections.defaultdict| containing the node-specific information
provided by XML `sequences` element.
>>> from hydpy.exe.xmltools import XMLInterface
>>> from hydpy.data import make_filepath
>>> interface = XMLInterface("single_run.xml", make_filepath("HydPy-H-Lahn"))
>>> series_io = interface.series_io
>>> subs2seqs = series_io.writers[2].subs2seqs
>>> for subs, seq in sorted(subs2seqs.items()):
... print(subs, seq)
node ['sim', 'obs']
"""
subs2seqs: collections.defaultdict[str, list[str]]
subs2seqs = collections.defaultdict(list)
nodes = find(self.find("sequences", optional=False), "node")
if nodes is not None:
for seq in nodes:
subs2seqs["node"].append(strip(seq.tag))
return subs2seqs
def _iterate_sequences(self) -> Iterator[sequencetools.IOSequence]:
return itertools.chain(
self._iterate_model_sequences(), self._iterate_node_sequences()
)
def _iterate_model_sequences(self) -> Iterator[sequencetools.IOSequence]:
m2s2s = self.model2subs2seqs
for element in self.elements:
for model in element.model.find_submodels(include_mainmodel=True).values():
for subseqs_name, seq_names in m2s2s.get(model.name, {}).items():
subseqs = getattr(model.sequences, subseqs_name)
for seq_name in seq_names:
yield getattr(subseqs, seq_name)
def _iterate_node_sequences(self) -> Iterator[sequencetools.IOSequence]:
s2s = self.subs2seqs
for node in self.nodes:
for seq_names in s2s.values():
for seq_name in seq_names:
yield getattr(node.sequences, seq_name)
[docs]
def prepare_series(self) -> None:
"""Call method |IOSequence.prepare_series| of class |IOSequence| for all
sequences selected by the given element of the actual XML file.
Method |IOSequence.prepare_series| solves a complex task, as it needs to
determine the correct arguments for method |IOSequence.prepare_series| of class
|IOSequence|. Those arguments depend on whether the respective |XMLSubseries|
element is for reading or writing data, addresses input or output sequences,
and if one prefers to handle time series data in RAM or read or write it "just
in time" during model simulations. Method |XMLSubseries.prepare_series|
delegates some of the related logic to the |PrepareSeriesArguments.from_xmldata|
method of class |PrepareSeriesArguments|. The following examples demonstrate
that method |XMLSubseries.prepare_series| implements the remaining logic
correctly.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_network()
... hp.prepare_models()
... interface = XMLInterface("single_run.xml")
>>> interface.update_timegrids()
>>> interface.update_selections()
First, we check and discuss the proper setting of the properties
|IOSequence.ramflag|, |IOSequence.diskflag_reading|, and
|IOSequence.diskflag_writing| for the sequences defined in `single_run.xml`.
First, we call |XMLSubseries.prepare_series| for available |XMLSubseries|
objects:
>>> series_io = interface.series_io
>>> for reader in series_io.readers:
... reader.prepare_series()
>>> for writer in series_io.writers:
... writer.prepare_series()
The following test function prints the options of the specified sequence of
element "Dill_assl":
>>> def print_io_options(groupname, sequencename):
... sequences = hp.elements.land_dill_assl.model.sequences
... sequence = sequences[groupname][sequencename]
... print(f"ramflag={sequence.ramflag}")
... print(f"diskflag_reading={sequence.diskflag_reading}")
... print(f"diskflag_writing={sequence.diskflag_writing}")
The XML file uses the `jit` mode for all non-aggregated time series. Reader
elements handle input sequences and writer elements handle output sequences.
Hence, |IOSequence.ramflag| is generally |False| while
|IOSequence.diskflag_reading| is |True| for the input sequences and
|IOSequence.diskflag_writing| is |True| for the output sequences:
>>> print_io_options("inputs", "p")
ramflag=False
diskflag_reading=True
diskflag_writing=False
>>> print_io_options("fluxes", "pc")
ramflag=False
diskflag_reading=False
diskflag_writing=True
Currently, aggregation only works in combination with mode `ram`:
>>> print_io_options("factors", "tc")
ramflag=True
diskflag_reading=False
diskflag_writing=False
For sequence |hland_states.SM|, two writers apply. The writer "soil moisture"
triggers writing the complete time series "just in time" during the simulation
run. In contrast, the writer "averaged" initiates writing averaged time series
after the simulation run. The configuration of sequence |hland_states.SM|
reflects this, with both "ram flag" and "disk flag writing" being "True":
>>> print_io_options("states", "sm")
ramflag=True
diskflag_reading=False
diskflag_writing=True
We temporarily convert the single reader element to a writer element and apply
method |XMLSubseries.prepare_series| again. At first, this does not work, as
the affected input sequences have previously been configured for "just in time"
reading, which does not work in combination with "just in time" writing:
>>> reader = series_io.readers[0]
>>> reader.root.tag = reader.root.tag.replace("reader", "writer")
>>> reader.prepare_series()
Traceback (most recent call last):
...
ValueError: Reading from and writing into the same NetCDF file "just in time" \
during a simulation run is not supported but tried for sequence `p` of element \
`land_dill_assl`.
After resetting all |InputSequence| objects and re-applying
|XMLSubseries.prepare_series|, both |IOSequence.ramflag| and
|IOSequence.diskflag_writing| are |True|. This case is the only one where a
single reader or writer enables two flags (see the documentation on method
|PrepareSeriesArguments.from_xmldata| for further information):
>>> hp.prepare_allseries(allocate_ram=False, jit=False)
>>> reader.prepare_series()
>>> reader.root.tag = reader.root.tag.replace("writer", "reader")
>>> print_io_options("inputs", "p")
ramflag=True
diskflag_reading=False
diskflag_writing=True
If we prefer the `ram` mode, things are more straightforward. Then, option
|IOSequence.ramflag| is |True|, and options |IOSequence.diskflag_reading| and
|IOSequence.diskflag_writing| are |False| regardless of all other options:
>>> hp.prepare_allseries(allocate_ram=False, jit=False)
>>> series_io.find("mode").text = "ram"
>>> for reader in series_io.readers:
... reader.find("mode").text = "ram"
... reader.prepare_series()
>>> for writer in series_io.writers:
... mode = writer.find("mode")
... if mode is not None:
... mode.text = "ram"
... writer.prepare_series()
>>> print_io_options("inputs", "p")
ramflag=True
diskflag_reading=False
diskflag_writing=False
>>> print_io_options("fluxes", "pc")
ramflag=True
diskflag_reading=False
diskflag_writing=False
>>> print_io_options("states", "sm")
ramflag=True
diskflag_reading=False
diskflag_writing=False
>>> reader = series_io.readers[0]
>>> reader.root.tag = reader.root.tag.replace("reader", "writer")
>>> reader.prepare_series()
>>> reader.root.tag = reader.root.tag.replace("writer", "reader")
>>> print_io_options("inputs", "p")
ramflag=True
diskflag_reading=False
diskflag_writing=False
"""
input_args = PrepareSeriesArguments.from_xmldata(
is_reader=self._is_reader, is_input=True, prefer_ram=self._ramflag
)
output_args = None
if self._is_writer:
output_args = PrepareSeriesArguments.from_xmldata(
is_reader=False, is_input=False, prefer_ram=self._ramflag
)
input_types = (sequencetools.InputSequence, sequencetools.NodeSequence)
for sequence in self._iterate_sequences():
args = (
input_args
if (output_args is None) or isinstance(sequence, input_types)
else output_args
)
sequence.prepare_series(
allocate_ram=sequence.ramflag or args.allocate_ram,
read_jit=sequence.diskflag_reading or args.read_jit,
write_jit=sequence.diskflag_writing or args.write_jit,
)
[docs]
def load_series(self, currentdir: Optional[str]) -> None:
"""Load time series data as defined by the actual XML `reader` element.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO, xml_replace, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_network()
... hp.prepare_models()
... xml_replace("HydPy-H-Lahn/single_run", printflag=False)
... interface = XMLInterface("single_run.xml")
... interface.update_options()
... interface.update_timegrids()
... interface.update_selections()
... series_io = interface.series_io
... series_io.prepare_series()
... series_io.load_series()
>>> from hydpy import print_vector
>>> print_vector(hp.elements.land_dill_assl.model.sequences.inputs.t.series[:3])
0.0, -0.5, -2.4
"""
if self._is_reader:
hydpy.pub.sequencemanager.open_netcdfreader()
self.prepare_sequencemanager(currentdir)
for sequence in self._iterate_sequences():
if sequence.ramflag:
sequence.load_series()
hydpy.pub.sequencemanager.close_netcdfreader()
[docs]
def save_series(self, currentdir: Optional[str]) -> None:
"""Save time series data as defined by the actual XML `writer` element.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, round_, TestIO, xml_replace, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_network()
... hp.prepare_models()
... xml_replace("HydPy-H-Lahn/single_run", printflag=False)
... interface = XMLInterface("single_run.xml")
>>> interface.update_options()
>>> interface.update_timegrids()
>>> interface.update_selections()
>>> series_io = interface.series_io
>>> series_io.prepare_series()
>>> hp.elements.land_dill_assl.model.sequences.fluxes.pc.series[2, 3] = 9.0
>>> hp.nodes.lahn_leun.sequences.sim.series[4] = 7.0
>>> with TestIO():
... series_io.save_series()
>>> import numpy
>>> with TestIO():
... dirpath = "HydPy-H-Lahn/series/default/"
... os.path.exists(f"{dirpath}land_lahn_leun_hland_96_flux_pc.npy")
... os.path.exists(f"{dirpath}land_lahn_kalk_hland_96_flux_pc.npy")
... round_(numpy.load(f"{dirpath}land_dill_assl_hland_96"
... f"_flux_pc.npy")[13+2, 3])
... round_(numpy.load(f"{dirpath}lahn_leun_sim_q_mean.npy")[13+4])
True
False
9.0
7.0
"""
if self.name == "writer":
hydpy.pub.sequencemanager.open_netcdfwriter()
self.prepare_sequencemanager(currentdir)
for sequence in self._iterate_sequences():
if not sequence.diskflag_writing:
sequence.save_series()
hydpy.pub.sequencemanager.close_netcdfwriter()
[docs]
def change_dirpath(self, currentdir: Optional[str]) -> None:
"""Set the |IOSequence.dirpath| of all relevant |IOSequence| objects to the
|FileManager.currentpath| of the |SequenceManager| object available in the
|pub| module.
This "information freezing" is required for those sequences selected for
reading data from or writing data to different directories "just in time"
during a simulation run.
"""
if not self._ramflag:
self.prepare_sequencemanager(currentdir)
currentpath = hydpy.pub.sequencemanager.currentpath
for sequence in self._iterate_sequences():
sequence.dirpath = currentpath
[docs]
def reset_dirpath(self) -> None:
"""Revert |XMLSubseries.change_dirpath|."""
if not self._ramflag:
for sequence in self._iterate_sequences():
del sequence.dirpath
[docs]
class XMLExchange(XMLBase):
"""Helper class for |XMLInterface| responsible for interpreting exchange items,
accessible via different |XMLItemgroup| instances."""
def __init__(self, master: XMLInterface, root: ElementTree.Element) -> None:
self.master: XMLInterface = master
self.root: ElementTree.Element = root
def _get_items_of_certain_item_types(
self, itemgroups: Iterable[str], itemtype: type[_TypeGetOrChangeItem]
) -> list[_TypeGetOrChangeItem]:
"""Return either all |GetItem| or all |ChangeItem| objects."""
items: list[_TypeGetOrChangeItem] = []
for itemgroup in self.itemgroups:
if (
issubclass(itemtype, itemtools.GetItem)
and (itemgroup.name == "getitems")
) or (
issubclass(itemtype, itemtools.ChangeItem)
and (itemgroup.name != "getitems")
):
for var in (
var
for model in itemgroup.models
for subvars in model.subvars
if subvars.name in itemgroups
for var in subvars.vars
):
item = var.item
assert isinstance(item, itemtype)
items.append(item)
if "nodes" in itemgroups:
for var in (var for node in itemgroup.nodes for var in node.vars):
item = var.item
assert isinstance(item, itemtype)
items.append(item)
return items
@property
def parameteritems(self) -> list[itemtools.ChangeItem]:
"""Create and return all items for changing control parameter values.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_everything()
... interface = XMLInterface("multiple_runs.xml")
>>> interface.update_selections()
>>> for item in interface.exchange.parameteritems:
... print(item.name)
alpha
beta
lag
damp
sfcf_1
sfcf_2
sfcf_3
k4
"""
return self._get_items_of_certain_item_types(
itemgroups=("control",), itemtype=itemtools.ChangeItem
)
@property
def inputitems(self) -> list[itemtools.SetItem]:
"""Return all items for changing input sequence values.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_everything()
... interface = XMLInterface("multiple_runs.xml")
>>> interface.update_selections()
>>> for item in interface.exchange.inputitems:
... print(item.name)
t_headwaters
"""
return self._get_items_of_certain_item_types(
itemgroups=("inputs",), itemtype=itemtools.SetItem
)
@property
def conditionitems(self) -> list[itemtools.SetItem]:
"""Return all items for changing condition sequence values.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_everything()
... interface = XMLInterface("multiple_runs.xml")
>>> interface.update_selections()
>>> for item in interface.exchange.conditionitems:
... print(item.name)
ic_lahn_leun
ic_lahn_marb
sm_lahn_leun
sm_lahn_marb
quh
"""
return self._get_items_of_certain_item_types(
itemgroups=("states", "logs"), itemtype=itemtools.SetItem
)
@property
def outputitems(self) -> list[itemtools.SetItem]:
"""Return all items for querying the current values or the complete time
series of sequences in the "setitem" style.
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_everything()
... interface = XMLInterface("multiple_runs.xml")
>>> interface.update_selections()
>>> for item in interface.exchange.outputitems:
... print(item.name)
swe_headwaters
"""
return self._get_items_of_certain_item_types(
itemgroups=("factors", "fluxes"), itemtype=itemtools.SetItem
)
@property
def getitems(self) -> list[itemtools.GetItem]:
"""Return all items for querying the current values or the complete time
series of sequences in the "getitem style".
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import HydPy, pub, TestIO, XMLInterface
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_everything()
... interface = XMLInterface("multiple_runs.xml")
>>> interface.update_selections()
>>> for item in interface.exchange.getitems:
... print(item.target)
factors_contriarea
fluxes_qt
fluxes_qt_series
states_sm
states_sm_series
nodes_sim_series
"""
return self._get_items_of_certain_item_types(
itemgroups=(
"control",
"inputs",
"factors",
"fluxes",
"states",
"logs",
"nodes",
),
itemtype=itemtools.GetItem,
)
[docs]
def prepare_series(self) -> None:
"""Prepare all required |IOSequence.series| arrays via the
|IOSequence.prepare_series| method.
"""
with warnings.catch_warnings():
warnings.filterwarnings(
"ignore", category=exceptiontools.AttributeNotReadyWarning
)
for item in itertools.chain(
self.inputitems, self.conditionitems, self.outputitems, self.getitems
):
for target in item.device2target.values():
if item.targetspecs.series:
assert isinstance(target, sequencetools.IOSequence)
target.prepare_series(
allocate_ram=True, read_jit=None, write_jit=None
)
# for base in getattr(item, "device2base", {}).values():
# if item.basespecs.series and not base.ramflag:
# base.prepare_series() ToDo
@property
def itemgroups(self) -> list[XMLItemgroup]:
"""The relevant |XMLItemgroup| objects."""
return [XMLItemgroup(self, element) for element in self]
[docs]
class XMLItemgroup(XMLBase):
"""Helper class for |XMLExchange| responsible for handling the exchange items
related to model parameters and sequences separately from the exchange items of
node sequences."""
def __init__(self, master: XMLExchange, root: ElementTree.Element) -> None:
self.master: XMLExchange = master
self.root: ElementTree.Element = root
@property
def models(self) -> list[XMLModel]:
"""The required |XMLModel| objects."""
return [
XMLModel(self, element) for element in self if strip(element.tag) != "nodes"
]
@property
def nodes(self) -> list[XMLNode]:
"""The required |XMLNode| objects."""
return [
XMLNode(self, element) for element in self if strip(element.tag) == "nodes"
]
[docs]
class XMLModel(XMLBase):
"""Helper class for |XMLItemgroup| responsible for handling the exchange items
related to different parameter or sequence groups of |Model| objects."""
def __init__(self, master: XMLItemgroup, root: ElementTree.Element) -> None:
self.master: XMLItemgroup = master
self.root: ElementTree.Element = root
@property
def subvars(self) -> list[XMLSubvars]:
"""The required |XMLSubVars| objects."""
return [XMLSubvars(self, element) for element in self]
[docs]
class XMLSubvars(XMLBase):
"""Helper class for |XMLModel| responsible for handling the exchange items
related to individual parameters or sequences of |Model| objects."""
def __init__(self, master: XMLModel, root: ElementTree.Element):
self.master: XMLModel = master
self.root: ElementTree.Element = root
@property
def vars(self) -> list[XMLVar]:
"""The required |XMLVar| objects."""
return [XMLVar(self, element) for element in self]
[docs]
class XMLNode(XMLBase):
"""Helper class for |XMLItemgroup| responsible for handling the exchange items
related to individual parameters or sequences of |Node| objects."""
def __init__(self, master: XMLItemgroup, root: ElementTree.Element) -> None:
self.master: XMLItemgroup = master
self.root: ElementTree.Element = root
@property
def vars(self) -> list[XMLVar]:
"""The required |XMLVar| objects."""
return [XMLVar(self, element) for element in self]
[docs]
class XMLVar(XMLSelector):
"""Helper class for |XMLSubvars| and |XMLNode| responsible for creating a defined
exchange item."""
def __init__(
self, master: Union[XMLSubvars, XMLNode], root: ElementTree.Element
) -> None:
self.master: Union[XMLSubvars, XMLNode] = master
self.root: ElementTree.Element = root
@property
def item(self) -> itemtools.ExchangeItem:
"""The defined |ExchangeItem| object.
We first prepare the `HydPy-H-Lahn` example project and then create the related
|XMLInterface| object defined by the XML configuration file `multiple_runs`:
>>> from hydpy.core.testtools import prepare_full_example_1
>>> prepare_full_example_1()
>>> from hydpy import (HydPy, round_, print_matrix, print_vector, pub, TestIO,
... XMLInterface)
>>> hp = HydPy("HydPy-H-Lahn")
>>> pub.timegrids = "1996-01-01", "1996-01-06", "1d"
>>> with TestIO():
... hp.prepare_everything()
... interface = XMLInterface("multiple_runs.xml")
>>> interface.update_selections()
One of the defined |SetItem| objects modifies the values of all
|hland_control.Alpha| objects of application model |hland_96|. We demonstrate
this for the control parameter object handled by the `land_dill_assl` element:
>>> var = interface.exchange.itemgroups[0].models[0].subvars[0].vars[0]
>>> item = var.item
>>> round_(item.value)
2.0
>>> hp.elements.land_dill_assl.model.parameters.control.alpha
alpha(1.0)
>>> item.update_variables()
>>> hp.elements.land_dill_assl.model.parameters.control.alpha
alpha(2.0)
The second example is comparable but focuses on a |SetItem| modifying control
parameter |musk_control.NmbSegments| of application model |musk_classic| via
its keyword argument `lag`:
>>> var = interface.exchange.itemgroups[0].models[2].subvars[0].vars[0]
>>> item = var.item
>>> round_(item.value)
5.0
>>> hp.elements.stream_dill_assl_lahn_leun.model.parameters.control.nmbsegments
nmbsegments(lag=0.0)
>>> item.update_variables()
>>> hp.elements.stream_dill_assl_lahn_leun.model.parameters.control.nmbsegments
nmbsegments(lag=5.0)
The third discussed |SetItem| assigns the same value to all entries of state
sequence |hland_states.SM|, resulting in the same soil moisture for all
individual hydrological response units of element `land_lahn_leun`:
>>> var = interface.exchange.itemgroups[1].models[0].subvars[0].vars[2]
>>> item = var.item
>>> item.name
'sm_lahn_leun'
>>> print_vector(item.value)
123.0
>>> hp.elements.land_lahn_leun.model.sequences.states.sm
sm(138.31396, 135.71124, 147.54968, 145.47142, 154.96405, 153.32805,
160.91917, 159.62434, 165.65575, 164.63255)
>>> item.update_variables()
>>> hp.elements.land_lahn_leun.model.sequences.states.sm
sm(123.0, 123.0, 123.0, 123.0, 123.0, 123.0, 123.0, 123.0, 123.0, 123.0)
In contrast to the last example, the fourth |SetItem| is 1-dimensional and thus
allows to assign different values to the individual hydrological response units
of element `land_lahn_marb`:
>>> var = interface.exchange.itemgroups[1].models[0].subvars[0].vars[3]
>>> item = var.item
>>> item.name
'sm_lahn_marb'
>>> print_vector(item.value)
110.0, 120.0, 130.0, 140.0, 150.0, 160.0, 170.0, 180.0, 190.0, 200.0,
210.0, 220.0, 230.0
>>> hp.elements.land_lahn_marb.model.sequences.states.sm
sm(99.27505, 96.17726, 109.16576, 106.39745, 117.97304, 115.56252,
125.81523, 123.73198, 132.80035, 130.91684, 138.95523, 137.25983,
142.84148)
>>> with pub.options.warntrim(False):
... item.update_variables()
>>> hp.elements.land_lahn_marb.model.sequences.states.sm
sm(110.0, 120.0, 130.0, 140.0, 150.0, 160.0, 170.0, 180.0, 190.0, 200.0,
206.0, 206.0, 206.0)
Without defining initial values in the XML file, the |ChangeItem.value|
property of each |SetItem| starts with the averaged (see item `ic_lahn_leun`) or
original (see item `ic_lahn_marb`) values of the corresponding sequences:
>>> var = interface.exchange.itemgroups[1].models[0].subvars[0].vars[0]
>>> item = var.item
>>> item.name
'ic_lahn_leun'
>>> round_(item.value)
1.184948
>>> ic_states = hp.elements.land_lahn_leun.model.sequences.states.ic
>>> round_(ic_states.average_values())
1.184948
>>> var = interface.exchange.itemgroups[1].models[0].subvars[0].vars[1]
>>> item = var.item
>>> item.name
'ic_lahn_marb'
>>> print_vector(item.value)
0.96404, 1.36332, 0.96458, 1.46458, 0.96512, 1.46512, 0.96565,
1.46569, 0.96617, 1.46617, 0.96668, 1.46668, 1.46719
>>> hp.elements.land_lahn_marb.model.sequences.states.ic
ic(0.96404, 1.36332, 0.96458, 1.46458, 0.96512, 1.46512, 0.96565,
1.46569, 0.96617, 1.46617, 0.96668, 1.46668, 1.46719)
Finally, one |SetItem| addresses the time series if the input sequence
|hland_inputs.T| of both headwater catchments. Similar to the example above,
its initial values stem from its target sequences' initial (time series)
values:
>>> var = interface.exchange.itemgroups[2].models[0].subvars[0].vars[0]
>>> item = var.item
>>> print_matrix(item.value)
| 0.0, -0.5, -2.4, -6.8, -7.8 |
| -0.7, -1.5, -4.6, -8.2, -8.7 |
>>> print_vector(hp.elements.land_dill_assl.model.sequences.inputs.t.series)
0.0, -0.5, -2.4, -6.8, -7.8
>>> print_vector(hp.elements.land_lahn_marb.model.sequences.inputs.t.series)
-0.7, -1.5, -4.6, -8.2, -8.7
>>> item.value = [0.0, 1.0, 2.0, 3.0, 4.0], [5.0, 6.0, 7.0, 8.0, 9.0]
>>> item.update_variables()
>>> print_vector(hp.elements.land_dill_assl.model.sequences.inputs.t.series)
0.0, 1.0, 2.0, 3.0, 4.0
>>> print_vector(hp.elements.land_lahn_marb.model.sequences.inputs.t.series)
5.0, 6.0, 7.0, 8.0, 9.0
|AddItem| `sfcf_1`, `sfcf_2`, and `sfcf_3` serve to demonstrate how a scalar
value (`sfcf_1` and `sfcf_2`) or a vector of values can be used to change the
value of a "target" parameter (|hland_control.SfCF|) in relation to a "base"
parameter (|hland_control.RfCF|):
>>> for element in pub.selections.headwaters.elements:
... element.model.parameters.control.rfcf(1.1)
>>> for element in pub.selections.nonheadwaters.elements:
... element.model.parameters.control.rfcf(1.0)
>>> for subvars in interface.exchange.itemgroups[3].models[0].subvars:
... for var in subvars.vars:
... var.item.update_variables()
>>> for element in hp.elements.catchment:
... print(element, repr(element.model.parameters.control.sfcf))
land_dill_assl sfcf(1.4)
land_lahn_kalk sfcf(field=1.1, forest=1.2)
land_lahn_leun sfcf(1.2)
land_lahn_marb sfcf(1.4)
|MultiplyItem| `k4` works similar to the described add items but multiplies
the current values of the base parameter objects of type |hland_control.K|
with 10 to gain new values for the target parameter objects of type
|hland_control.K4|:
>>> for subvars in interface.exchange.itemgroups[4].models[0].subvars:
... for var in subvars.vars:
... var.item.update_variables()
>>> for element in hp.elements.catchment:
... control = element.model.parameters.control
... print(element, repr(control.k), repr(control.k4))
land_dill_assl k(0.005618) k4(0.056177)
land_lahn_kalk k(0.002571) k4(0.025712)
land_lahn_leun k(0.005948) k4(0.059481)
land_lahn_marb k(0.005325) k4(0.053247)
The final three examples focus on |GetItem| objects. One |GetItem| object
queries the actual values of the |hland_states.SM| states of all relevant
elements:
>>> var = interface.exchange.itemgroups[5].models[0].subvars[2].vars[0]
>>> hp.elements.land_dill_assl.model.sequences.states.sm = 1.0
>>> for name, target in var.item.yield_name2value():
... print(name, target) # doctest: +ELLIPSIS
land_dill_assl_states_sm [1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, \
1.0, 1.0]
land_lahn_kalk_states_sm [101.3124...]
land_lahn_leun_states_sm [123.0, 123.0, 123.0, 123.0, 123.0, 123.0, 123.0, \
123.0, 123.0, 123.0]
land_lahn_marb_states_sm [110.0, 120.0, 130.0, 140.0, 150.0, 160.0, 170.0, \
180.0, 190.0, 200.0, 206.0, 206.0, 206.0]
Another |GetItem| object queries the actual value of the
|hland_factors.ContriArea| factor sequence of element `land_dill_assl`:
>>> hp.elements.land_dill_assl.model.sequences.factors.contriarea(1.0)
>>> for var in interface.exchange.itemgroups[5].models[0].subvars[0].vars:
... for name, target in var.item.yield_name2value():
... print(name, target)
land_dill_assl_factors_contriarea 1.0
Another |GetItem| object queries both the actual and the time series values of
the |hland_fluxes.QT| flux sequence of element `land_dill_assl`:
>>> qt = hp.elements.land_dill_assl.model.sequences.fluxes.qt
>>> qt(1.0)
>>> qt.series = 2.0
>>> for var in interface.exchange.itemgroups[5].models[0].subvars[1].vars:
... for name, target in var.item.yield_name2value():
... print(name, target)
land_dill_assl_fluxes_qt 1.0
land_dill_assl_fluxes_qt_series [2.0, 2.0, 2.0, 2.0, 2.0]
Last but not least, one |GetItem| queries the simulated time series values
available through node `dill_assl`:
>>> var = interface.exchange.itemgroups[5].nodes[0].vars[0]
>>> hp.nodes.dill_assl.sequences.sim.series = range(5)
>>> for name, target in var.item.yield_name2value():
... print(name, target)
dill_assl_nodes_sim_series [0.0, 1.0, 2.0, 3.0, 4.0]
>>> for name, target in var.item.yield_name2value(2, 4):
... print(name, target)
dill_assl_nodes_sim_series [2.0, 3.0]
"""
target = f"{self.master.name}.{self.name}"
if self.master.name == "nodes":
master = self.master.name
itemgroup = self.master.master.name
else:
master = self.master.master.name
itemgroup = self.master.master.master.name
itemtype = _ITEMGROUP2ITEMCLASS[itemgroup]
if itemgroup == "getitems":
return self._get_getitem(target, master, itemtype)
return self._get_changeitem(target, master, itemtype)
def _get_getitem(
self, target: str, master: str, itemtype: type[itemtools.GetItem]
) -> itemtools.GetItem:
xmlelement = self.find("name", optional=True)
if xmlelement is None or xmlelement.text is None:
name = cast(Name, "?")
else:
name = cast(Name, xmlelement.text)
item = itemtype(name, master, target)
self._collect_variables(item)
return item
def _get_changeitem(
self, target: str, master: str, itemtype: type[_TypeSetOrAddOrMultiplyItem]
) -> _TypeSetOrAddOrMultiplyItem:
name = cast(Name, self.find("name", optional=False).text)
assert name is not None
level = self.find("level", optional=False).text
assert level is not None
item: _TypeSetOrAddOrMultiplyItem
# Simplify the following if-clauses after Mypy issue 10989 is fixed?
if not issubclass(itemtype, itemtools.SetItem):
item = itemtype(
name=name,
master=master,
target=target,
base=strip(list(self)[-1].tag),
level=cast(itemtools.LevelType, level),
)
elif not issubclass(itemtype, (itemtools.AddItem, itemtools.MultiplyItem)):
keyword = self.find("keyword", optional=True)
item = itemtype(
name=name,
master=master,
target=target,
keyword=None if keyword is None else keyword.text,
level=cast(itemtools.LevelType, level),
)
self._collect_variables(item)
element = self.find("init", optional=True)
if element is not None:
init = element.text
assert init is not None
item.value = eval(",".join(init.split()))
else:
assert isinstance(item, itemtools.SetItem)
item.extract_values()
return item
def _collect_variables(self, item: itemtools.ExchangeItem) -> None:
selections = self.selections
item.collect_variables(selections)
[docs]
class XSDWriter:
"""A pure |classmethod| class for writing the actual XML schema file
`HydPyConfigBase.xsd`, which makes sure that an XML configuration file is
readable by class |XMLInterface|.
Unless you are interested in enhancing HydPy's XML functionalities, you should,
if any, be interested in method |XSDWriter.write_xsd| only.
"""
confpath: str = conf.__path__[0]
filepath_source: str = os.path.join(confpath, "HydPyConfigBase" + ".xsdt")
filepath_target: str = filepath_source[:-1]
[docs]
@classmethod
def write_xsd(cls) -> None:
"""Write the complete base schema file `HydPyConfigBase.xsd` based on the
template file `HydPyConfigBase.xsdt`.
Method |XSDWriter.write_xsd| adds model-specific information to the general
information of template file `HydPyConfigBase.xsdt` regarding reading and
writing of time series data and exchanging parameter and sequence values, for
example, during calibration.
The following example shows that after writing a new schema file, method
|XMLInterface.validate_xml| does not raise an error when either applied to the
XML configuration files `single_run.xml` or `multiple_runs.xml` of the
`HydPy-H-Lahn` example project:
>>> import os
>>> from hydpy.exe.xmltools import XSDWriter, XMLInterface
>>> if os.path.exists(XSDWriter.filepath_target):
... os.remove(XSDWriter.filepath_target)
>>> os.path.exists(XSDWriter.filepath_target)
False
>>> XSDWriter.write_xsd()
>>> os.path.exists(XSDWriter.filepath_target)
True
>>> from hydpy.data import make_filepath
>>> for configfile in ("single_run.xml", "multiple_runs.xml"):
... XMLInterface(configfile, make_filepath("HydPy-H-Lahn")).validate_xml()
"""
with open(cls.filepath_source, encoding=config.ENCODING) as file_:
template = file_.read()
template = template.replace(
"<!--include model sequence groups-->", cls.get_insertion()
)
template = template.replace(
"<!--include exchange items-->", cls.get_exchangeinsertion()
)
with open(cls.filepath_target, "w", encoding=config.ENCODING) as file_:
file_.write(template)
[docs]
@staticmethod
def get_basemodelnames() -> list[str]:
"""Return a sorted |list| containing all base model names.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_basemodelnames()) # doctest: +ELLIPSIS
['arma', 'conv', ..., 'wland', 'wq']
"""
modelspath: str = models.__path__[0]
def _is_basemodel(dirname: str) -> bool:
pathname = os.path.join(modelspath, dirname)
return os.path.isdir(pathname) and ("__init__.py" in os.listdir(pathname))
return sorted(dn for dn in os.listdir(modelspath) if _is_basemodel(dn))
[docs]
@staticmethod
def get_applicationmodelnames() -> list[str]:
"""Return a sorted |list| containing all application model names.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_applicationmodelnames()) # doctest: +ELLIPSIS
[...'dam_v001', 'dam_v002', 'dam_v003', 'dam_v004', 'dam_v005',...]
"""
modelspath: str = models.__path__[0]
return sorted(
str(fn.split(".")[0])
for fn in sorted(os.listdir(modelspath))
if (fn.endswith(".py") and (fn != "__init__.py"))
)
[docs]
@classmethod
def get_insertion(cls) -> str:
"""Return the complete string to be inserted into the string of the template
file.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_insertion()) # doctest: +ELLIPSIS
<complexType name="dummy_interceptedwater_readerType">
<sequence>
<element name="inputs"
minOccurs="0">
<complexType>
<sequence>
<element
name="interceptedwater"
minOccurs="0"/>
</sequence>
</complexType>
</element>
</sequence>
</complexType>
...
<complexType name="dummy_snowalbedo_readerType">
<sequence>
<element name="inputs"
minOccurs="0">
<complexType>
<sequence>
<element
name="snowalbedo"
minOccurs="0"/>
...
<element name="wland_wag"
type="hpcb:wland_wag_readerType"
minOccurs="0"/>
</sequence>
</complexType>
...
<complexType name="arma_rimorido_writerType">
<sequence>
<element name="fluxes"
minOccurs="0">
<complexType>
<sequence>
<element
name="qin"
...
<element
name="qout"
minOccurs="0"/>
</sequence>
</complexType>
</element>
</sequence>
</complexType>
...
<complexType name="writerType">
<sequence>
<element name="node"
type="hpcb:node_writerType"
minOccurs="0"/>
<element name="arma_rimorido"
type="hpcb:arma_rimorido_writerType"
minOccurs="0"/>
...
<element name="wq_trapeze_strickler"
type="hpcb:wq_trapeze_strickler_writerType"
minOccurs="0"/>
</sequence>
</complexType>
<BLANKLINE>
"""
indent = 1
blanks = " " * (indent * 4)
subs = []
types_: tuple[Literal["reader", "writer"], ...] = ("reader", "writer")
for type_ in types_:
for name in cls.get_applicationmodelnames():
model = importtools.prepare_model(name)
modelinsertion = cls.get_modelinsertion(
model=model, type_=type_, indent=indent + 2
)
if modelinsertion:
subs.extend(
[
f'{blanks}<complexType name="{name}_{type_}Type">',
f"{blanks} <sequence>",
modelinsertion,
f"{blanks} </sequence>",
f"{blanks}</complexType>",
"",
]
)
subs.append(cls.get_readerwriterinsertion(type_=type_, indent=indent))
return "\n".join(subs)
[docs]
@classmethod
def get_modelinsertion(
cls, model: modeltools.Model, type_: str, indent: int
) -> Optional[str]:
"""Return the insertion string required for the given application model.
>>> from hydpy.exe.xmltools import XSDWriter
>>> from hydpy import prepare_model
>>> model = prepare_model("hland_96")
>>> print(XSDWriter.get_modelinsertion(
... model=model, type_="reader", indent=1)) # doctest: +ELLIPSIS
<element name="inputs"
minOccurs="0">
<complexType>
<sequence>
<element
name="p"
minOccurs="0"/>
<element
name="t"
minOccurs="0"/>
</sequence>
</complexType>
</element>
>>> print(XSDWriter.get_modelinsertion(
... model=model, type_="writer", indent=1)) # doctest: +ELLIPSIS
<element name="inputs"
minOccurs="0">
<complexType>
<sequence>
<element
name="p"
minOccurs="0"/>
...
</element>
<element name="fluxes"
minOccurs="0">
...
</element>
<element name="states"
minOccurs="0">
...
</element>
>>> model = prepare_model("arma_rimorido")
>>> XSDWriter.get_modelinsertion(
... model=model, type_="reader", indent=1) # doctest: +ELLIPSIS
>>> print(XSDWriter.get_modelinsertion(
... model=model, type_="writer", indent=1)) # doctest: +ELLIPSIS
<element name="fluxes"
minOccurs="0">
<complexType>
<sequence>
<element
name="qin"
minOccurs="0"/>
...
<element
name="qout"
minOccurs="0"/>
</sequence>
</complexType>
</element>
"""
names: tuple[str, ...] = ("inputs",)
if type_ == "writer":
names += "factors", "fluxes", "states"
texts = []
return_none = True
for name in names:
subsequences = getattr(model.sequences, name, None)
if subsequences:
return_none = False
texts.append(cls.get_subsequencesinsertion(subsequences, indent))
return None if return_none else "\n".join(texts)
[docs]
@classmethod
def get_subsequencesinsertion(
cls, subsequences: sequencetools.SubSequences[Any, Any, Any], indent: int
) -> str:
"""Return the insertion string required for the given group of sequences.
>>> from hydpy.exe.xmltools import XSDWriter
>>> from hydpy import prepare_model
>>> model = prepare_model("hland_96")
>>> print(XSDWriter.get_subsequencesinsertion(
... model.sequences.factors, 1)) # doctest: +ELLIPSIS
<element name="factors"
minOccurs="0">
<complexType>
<sequence>
<element
name="tc"
minOccurs="0"/>
<element
name="fracrain"
minOccurs="0"/>
...
<element
name="contriarea"
minOccurs="0"/>
</sequence>
</complexType>
</element>
"""
blanks = " " * (indent * 4)
lines = [
f'{blanks}<element name="{subsequences.name}"',
f'{blanks} minOccurs="0">',
f"{blanks} <complexType>",
f"{blanks} <sequence>",
]
for sequence in subsequences:
lines.append(cls.get_sequenceinsertion(sequence, indent + 3))
lines.extend(
[
f"{blanks} </sequence>",
f"{blanks} </complexType>",
f"{blanks}</element>",
]
)
return "\n".join(lines)
[docs]
@staticmethod
def get_sequenceinsertion(sequence: sequencetools.Sequence_, indent: int) -> str:
"""Return the insertion string required for the given sequence.
>>> from hydpy.exe.xmltools import XSDWriter
>>> from hydpy import prepare_model
>>> model = prepare_model("hland_96")
>>> print(XSDWriter.get_sequenceinsertion(model.sequences.fluxes.pc, 1))
<element
name="pc"
minOccurs="0"/>
"""
blanks = " " * (indent * 4)
return (
f"{blanks}<element\n"
f'{blanks} name="{sequence.name}"\n'
f'{blanks} minOccurs="0"/>'
)
[docs]
@classmethod
def get_readerwriterinsertion(
cls, type_: Literal["reader", "writer"], indent: int
) -> str:
"""Return the insertion all sequences relevant for reading or writing
time series data.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_readerwriterinsertion("reader", 1)) # doctest: +ELLIPSIS
<complexType name="readerType">
<sequence>
<element name="node"
type="hpcb:node_readerType"
minOccurs="0"/>
<element name="dummy_interceptedwater"
type="hpcb:dummy_interceptedwater_readerType"
minOccurs="0"/>
...
<element name="wland_wag"
type="hpcb:wland_wag_readerType"
minOccurs="0"/>
</sequence>
</complexType>
<BLANKLINE>
>>> print(XSDWriter.get_readerwriterinsertion("writer", 1)) # doctest: +ELLIPSIS
<complexType name="writerType">
<sequence>
<element name="node"
type="hpcb:node_writerType"
minOccurs="0"/>
<element name="arma_rimorido"
type="hpcb:arma_rimorido_writerType"
minOccurs="0"/>
...
<element name="wq_trapeze_strickler"
type="hpcb:wq_trapeze_strickler_writerType"
minOccurs="0"/>
</sequence>
</complexType>
<BLANKLINE>
"""
blanks = " " * (indent * 4)
subs = [
f'{blanks}<complexType name="{type_}Type">',
f"{blanks} <sequence>",
f'{blanks} <element name="node"',
f'{blanks} type="hpcb:node_{type_}Type"',
f'{blanks} minOccurs="0"/>',
]
for name in cls.get_applicationmodelnames():
seqs = importtools.prepare_model(name).sequences
if seqs.inputs or (
((type_ == "writer") and (seqs.factors or seqs.fluxes or seqs.states))
):
subs.extend(
[
f'{blanks} <element name="{name}"',
f'{blanks} type="hpcb:{name}_{type_}Type"',
f'{blanks} minOccurs="0"/>',
]
)
subs.extend([f"{blanks} </sequence>", f"{blanks}</complexType>", ""])
return "\n".join(subs)
[docs]
@classmethod
def get_exchangeinsertion(cls) -> str:
"""Return the complete string related to the definition of exchange items to
be inserted into the string of the template file.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_exchangeinsertion()) # doctest: +ELLIPSIS
<complexType name="arma_rimorido_mathitemType">
...
<element name="setitems">
...
<complexType name="arma_rimorido_setitemsType">
...
<element name="additems">
...
<element name="multiplyitems">
...
<element name="getitems">
...
"""
indent = 1
subs = [
cls.get_mathitemsinsertion(indent),
cls.get_keyworditemsinsertion(indent),
]
for groupname in ("setitems", "additems", "multiplyitems", "getitems"):
subs.append(cls.get_itemsinsertion(groupname, indent))
subs.append(cls.get_itemtypesinsertion(groupname, indent))
return "\n".join(subs)
[docs]
@classmethod
def get_mathitemsinsertion(cls, indent: int) -> str:
"""Return a string defining a model-specific XML type extending `ItemType`.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_mathitemsinsertion(1)) # doctest: +ELLIPSIS
<complexType name="arma_rimorido_mathitemType">
<complexContent>
<extension base="hpcb:mathitemType">
<choice>
<element name="control.responses"/>
...
<element name="fluxes.qout"/>
</choice>
</extension>
</complexContent>
</complexType>
<BLANKLINE>
<complexType name="conv_idw_mathitemType">
...
"""
blanks = " " * (indent * 4)
subs = []
for modelname in cls.get_applicationmodelnames():
model = importtools.prepare_model(modelname)
subs.extend(
[
f'{blanks}<complexType name="{modelname}_mathitemType">',
f"{blanks} <complexContent>",
f'{blanks} <extension base="hpcb:mathitemType">',
f"{blanks} <choice>",
]
)
for subvars in cls._get_subvars(model, conditions=False):
for var in subvars:
subs.append(
f"{blanks} "
f'<element name="{subvars.name}.{var.name}"/>'
)
subs.extend(
[
f"{blanks} </choice>",
f"{blanks} </extension>",
f"{blanks} </complexContent>",
f"{blanks}</complexType>",
"",
]
)
return "\n".join(subs)
[docs]
@classmethod
def get_keyworditemsinsertion(cls, indent: int) -> str:
"""Return a string defining additional types that support modifying parameter
values by specific keyword arguments.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_keyworditemsinsertion(1)) # doctest: +ELLIPSIS
<simpleType name="lland_control_kapgrenz_keywordType">
<restriction base="string">
<enumeration value="option"/>
</restriction>
</simpleType>
<BLANKLINE>
<complexType name="lland_control_kapgrenz_setitemType">
<complexContent>
<extension base="hpcb:setitemType">
<sequence>
<element name="keyword"
type="hpcb:lland_control_kapgrenz_keywordType"
minOccurs = "0"/>
</sequence>
</extension>
</complexContent>
</complexType>
...
"""
blanks = " " * (indent * 4)
subs = []
for modelname in cls.get_basemodelnames():
model = importtools.prepare_model(modelname)
for subvars in cls._get_subvars(model, conditions=False):
for var in subvars:
if isinstance(var, parametertools.Parameter) and var.KEYWORDS:
prefix = f"{modelname.split('_')[0]}_{subvars.name}_{var.name}_"
subs.extend(
[
f'{blanks}<simpleType name="{prefix}keywordType">',
f'{blanks} <restriction base="string">',
]
)
for keyword in var.KEYWORDS:
subs.append(
f'{blanks} <enumeration value="{keyword}"/>'
)
subs.extend(
[
f"{blanks} </restriction>",
f"{blanks}</simpleType>",
"",
f'{blanks}<complexType name="{prefix}setitemType">',
f"{blanks} <complexContent>",
f'{blanks} <extension base="hpcb:setitemType">',
f"{blanks} <sequence>",
f'{blanks} <element name="keyword"',
f"{blanks} "
f'type="hpcb:{prefix}keywordType"',
f'{blanks} minOccurs = "0"/>',
f"{blanks} </sequence>",
f"{blanks} </extension>",
f"{blanks} </complexContent>",
f"{blanks}</complexType>",
"",
]
)
return "\n".join(subs)
@staticmethod
def _get_itemstype(modelname: str, itemgroup: str) -> str:
return f"{modelname}_{itemgroup}Type"
[docs]
@classmethod
def get_itemsinsertion(cls, itemgroup: str, indent: int) -> str:
"""Return a string defining the XML element for the given exchange item group.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_itemsinsertion("setitems", 1)) # doctest: +ELLIPSIS
<element name="setitems">
<complexType>
<sequence>
<element ref="hpcb:selections"
minOccurs="0"/>
...
<element name="hland_96"
type="hpcb:hland_96_setitemsType"
minOccurs="0"
maxOccurs="unbounded"/>
...
<element name="nodes"
type="hpcb:nodes_setitemsType"
minOccurs="0"
maxOccurs="unbounded"/>
</sequence>
<attribute name="info" type="string"/>
</complexType>
</element>
<BLANKLINE>
"""
blanks = " " * (indent * 4)
subs = []
subs.extend(
[
f'{blanks}<element name="{itemgroup}">',
f"{blanks} <complexType>",
f"{blanks} <sequence>",
f'{blanks} <element ref="hpcb:selections"',
f'{blanks} minOccurs="0"/>',
]
)
for modelname in cls.get_applicationmodelnames():
type_ = cls._get_itemstype(modelname, itemgroup)
subs.append(f'{blanks} <element name="{modelname}"')
subs.append(f'{blanks} type="hpcb:{type_}"')
subs.append(f'{blanks} minOccurs="0"')
subs.append(f'{blanks} maxOccurs="unbounded"/>')
if itemgroup in ("setitems", "getitems"):
type_ = f"nodes_{itemgroup}Type"
subs.append(f'{blanks} <element name="nodes"')
subs.append(f'{blanks} type="hpcb:{type_}"')
subs.append(f'{blanks} minOccurs="0"')
subs.append(f'{blanks} maxOccurs="unbounded"/>')
subs.extend(
[
f"{blanks} </sequence>",
f'{blanks} <attribute name="info" type="string"/>',
f"{blanks} </complexType>",
f"{blanks}</element>",
"",
]
)
return "\n".join(subs)
[docs]
@classmethod
def get_itemtypesinsertion(cls, itemgroup: str, indent: int) -> str:
"""Return a string defining the required types for the given exchange item
group.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_itemtypesinsertion(
... "setitems", 1)) # doctest: +ELLIPSIS
<complexType name="arma_rimorido_setitemsType">
...
</complexType>
<BLANKLINE>
<complexType name="dam_v001_setitemsType">
...
<complexType name="nodes_setitemsType">
...
"""
subs = []
for modelname in cls.get_applicationmodelnames():
subs.append(cls.get_itemtypeinsertion(itemgroup, modelname, indent))
subs.append(cls.get_nodesitemtypeinsertion(itemgroup, indent))
return "\n".join(subs)
[docs]
@classmethod
def get_itemtypeinsertion(cls, itemgroup: str, modelname: str, indent: int) -> str:
"""Return a string defining the required types for the given combination of
an exchange item group and an application model.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_itemtypeinsertion(
... "setitems", "hland_96", 1)) # doctest: +ELLIPSIS
<complexType name="hland_96_setitemsType">
<sequence>
<element ref="hpcb:selections"
minOccurs="0"/>
<element name="control"
minOccurs="0"
maxOccurs="unbounded">
...
</sequence>
</complexType>
<BLANKLINE>
"""
blanks = " " * (indent * 4)
type_ = cls._get_itemstype(modelname, itemgroup)
subs = [
f'{blanks}<complexType name="{type_}">',
f"{blanks} <sequence>",
f'{blanks} <element ref="hpcb:selections"',
f'{blanks} minOccurs="0"/>',
cls.get_subgroupsiteminsertion(itemgroup, modelname, indent + 2),
f"{blanks} </sequence>",
f"{blanks}</complexType>",
"",
]
return "\n".join(subs)
[docs]
@classmethod
def get_nodesitemtypeinsertion(cls, itemgroup: str, indent: int) -> str:
"""Return a string defining the required types for the given combination of
an exchange item group and |Node| objects.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_nodesitemtypeinsertion(
... "setitems", 1)) # doctest: +ELLIPSIS
<complexType name="nodes_setitemsType">
<sequence>
<element ref="hpcb:selections"
minOccurs="0"/>
<element name="sim"
type="hpcb:setitemType"
minOccurs="0"
maxOccurs="unbounded"/>
<element name="obs"
type="hpcb:setitemType"
minOccurs="0"
maxOccurs="unbounded"/>
<element name="sim.series"
type="hpcb:setitemType"
minOccurs="0"
maxOccurs="unbounded"/>
<element name="obs.series"
type="hpcb:setitemType"
minOccurs="0"
maxOccurs="unbounded"/>
</sequence>
</complexType>
<BLANKLINE>
"""
blanks = " " * (indent * 4)
subs = [
f'{blanks}<complexType name="nodes_{itemgroup}Type">',
f"{blanks} <sequence>",
f'{blanks} <element ref="hpcb:selections"',
f'{blanks} minOccurs="0"/>',
]
type_ = "getitemType" if itemgroup == "getitems" else "setitemType"
for name in ("sim", "obs", "sim.series", "obs.series"):
subs.extend(
[
f'{blanks} <element name="{name}"',
f'{blanks} type="hpcb:{type_}"',
f'{blanks} minOccurs="0"',
f'{blanks} maxOccurs="unbounded"/>',
]
)
subs.extend([f"{blanks} </sequence>", f"{blanks}</complexType>", ""])
return "\n".join(subs)
[docs]
@classmethod
def get_subgroupsiteminsertion(
cls, itemgroup: str, modelname: str, indent: int
) -> str:
"""Return a string defining the required types for the given combination of an
exchange item group and an application model.
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_subgroupsiteminsertion(
... "setitems", "hland_96", 1)) # doctest: +ELLIPSIS
<element name="control"
minOccurs="0"
maxOccurs="unbounded">
...
</element>
<element name="inputs"
...
<element name="factors"
...
<element name="fluxes"
...
<element name="states"
...
"""
subs = []
model = importtools.prepare_model(modelname)
conditions = itemgroup in ("getitems", "setitems")
for subvars in cls._get_subvars(model, conditions=conditions):
subs.append(
cls.get_subgroupiteminsertion(itemgroup, model, subvars, indent)
)
return "\n".join(subs)
@classmethod
def _get_subvars(
cls, model: modeltools.Model, conditions: bool
) -> Iterator[variabletools.SubVariables[Any, Any, Any]]:
yield model.parameters.control
names = ["inputs", "factors", "fluxes"]
if conditions:
names.extend(("states", "logs"))
for name in names:
subseqs = getattr(model.sequences, name, None)
if subseqs:
yield subseqs
[docs]
@classmethod
def get_subgroupiteminsertion(
cls,
itemgroup: str,
model: modeltools.Model,
subgroup: variabletools.SubVariables[Any, Any, Any],
indent: int,
) -> str:
"""Return a string defining the required types for the given combination of an
exchange item group and a specific variable subgroup of an application model or
class |Node|.
Note that for `setitems` and `getitems` `setitemType` and `getitemType` are
referenced, respectively, and for all others, the model-specific `mathitemType`:
>>> from hydpy import prepare_model
>>> model = prepare_model("hland_96")
>>> from hydpy.exe.xmltools import XSDWriter
>>> print(XSDWriter.get_subgroupiteminsertion( # doctest: +ELLIPSIS
... "setitems", model, model.parameters.control, 1))
<element name="control"
minOccurs="0"
maxOccurs="unbounded">
<complexType>
<sequence>
<element ref="hpcb:selections"
minOccurs="0"/>
<element name="area"
type="hpcb:setitemType"
minOccurs="0"
maxOccurs="unbounded"/>
<element name="nmbzones"
...
</sequence>
</complexType>
</element>
>>> print(XSDWriter.get_subgroupiteminsertion( # doctest: +ELLIPSIS
... "getitems", model, model.parameters.control, 1))
<element name="control"
...
<element name="area"
type="hpcb:getitemType"
minOccurs="0"
maxOccurs="unbounded"/>
...
>>> print(XSDWriter.get_subgroupiteminsertion( # doctest: +ELLIPSIS
... "additems", model, model.parameters.control, 1))
<element name="control"
...
<element name="area"
type="hpcb:hland_96_mathitemType"
minOccurs="0"
maxOccurs="unbounded"/>
...
>>> print(XSDWriter.get_subgroupiteminsertion( # doctest: +ELLIPSIS
... "multiplyitems", model, model.parameters.control, 1))
<element name="control"
...
<element name="area"
type="hpcb:hland_96_mathitemType"
minOccurs="0"
maxOccurs="unbounded"/>
...
For sequence classes, additional "series" elements are added:
>>> print(XSDWriter.get_subgroupiteminsertion( # doctest: +ELLIPSIS
... "setitems", model, model.sequences.factors, 1))
<element name="factors"
...
<element name="tc"
type="hpcb:setitemType"
minOccurs="0"
maxOccurs="unbounded"/>
<element name="tc.series"
type="hpcb:setitemType"
minOccurs="0"
maxOccurs="unbounded"/>
<element name="fracrain"
...
</sequence>
</complexType>
</element>
"""
blanks1 = " " * (indent * 4)
blanks2 = " " * ((indent + 5) * 4 + 1)
subs = [
f'{blanks1}<element name="{subgroup.name}"',
f'{blanks1} minOccurs="0"',
f'{blanks1} maxOccurs="unbounded">',
f"{blanks1} <complexType>",
f"{blanks1} <sequence>",
f'{blanks1} <element ref="hpcb:selections"',
f'{blanks1} minOccurs="0"/>',
]
seriesflags = [False] if subgroup.name == "control" else [False, True]
for var in subgroup:
for series in seriesflags:
name = f"{var.name}.series" if series else var.name
subs.append(f'{blanks1} <element name="{name}"')
if itemgroup == "setitems":
if isinstance(var, parametertools.Parameter) and var.KEYWORDS:
type_ = (
f"{model.name.split('_')[0]}_{subgroup.name}_"
f"{var.name}_setitemType"
)
else:
type_ = "setitemType"
elif itemgroup == "getitems":
type_ = "getitemType"
else:
type_ = f"{model.name}_mathitemType"
subs.append(f'{blanks2}type="hpcb:{type_}"')
subs.append(f'{blanks2}minOccurs="0"')
subs.append(f'{blanks2}maxOccurs="unbounded"/>')
subs.extend(
[
f"{blanks1} </sequence>",
f"{blanks1} </complexType>",
f"{blanks1}</element>",
]
)
return "\n".join(subs)
[docs]
def xml_validate(xmlpath: str) -> int:
"""Check if an XML file complies with its XSD Schema file.
|xml_validate| relies on method |XMLInterface.validate_xml| of class
|XMLInterface|. It is primarily designed for command line usage, as explained in
the :ref:`User Guide'a <user_guide>` :ref:`Simulation > XML <simulation_xml>`
section.
"""
dirpath, filename = os.path.split(xmlpath)
interface = XMLInterface(filename=filename, directory=dirpath)
try:
interface.validate_xml()
except BaseException as exc:
print(str(exc).rpartition("the following error occurred:")[2].strip())
return 999
print(f"{xmlpath} successfully validated")
return 0
