Skip to content

DIMR xml files

The input to the Deltares Integrated Model Runner (DIMR) is a single XML file, represented by the classes below.

Model

Component

Bases: BaseModel, ABC

Specification of a BMI-compliant model component instance that will be executed by DIMR.

Attributes:

Name Type Description
library str

The library name of the compoment.

name str

The component name.

workingDir Path

The working directory.

inputFile Path

The name of the input file.

process Optional[int]

Number of subprocesses in the component.

setting Optional[List[KeyValuePair]]

A list of variables that are provided to the BMI model before initialization.

parameter Optional[List[KeyValuePair]]

A list of variables that are provided to the BMI model after initialization.

mpiCommunicator Optional[str]

The MPI communicator value.

model Optional[FileModel]

The model represented by this component.

Source code in hydrolib/core/dimr/models.py
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class Component(BaseModel, ABC):
    """
    Specification of a BMI-compliant model component instance that will be executed by DIMR.

    Attributes:
        library: The library name of the compoment.
        name: The component name.
        workingDir: The working directory.
        inputFile: The name of the input file.
        process: Number of subprocesses in the component.
        setting: A list of variables that are provided to the BMI model before initialization.
        parameter: A list of variables that are provided to the BMI model after initialization.
        mpiCommunicator: The MPI communicator value.
        model: The model represented by this component.
    """

    library: str
    name: str
    workingDir: Path
    inputFile: Path
    process: Optional[int]
    setting: Optional[List[KeyValuePair]] = []
    parameter: Optional[List[KeyValuePair]] = []
    mpiCommunicator: Optional[str]

    model: Optional[FileModel]

    @property
    def filepath(self):
        return self.workingDir / self.inputFile

    @abstractclassmethod
    def get_model(cls) -> Type[FileModel]:
        raise NotImplementedError("Model not implemented yet.")

    @validator("setting", "parameter", pre=True, allow_reuse=True)
    def validate_setting(cls, v):
        return to_list(v)

    def is_intermediate_link(self) -> bool:
        return True

    def _get_identifier(self, data: dict) -> Optional[str]:
        return data.get("name")

    def dict(self, *args, **kwargs):
        # Exclude the FileModel from any DIMR serialization.
        kwargs["exclude"] = {"model"}
        return super().dict(*args, **kwargs)

ComponentOrCouplerRef

Bases: BaseModel

Reference to a BMI-compliant model component instance.

Attributes:

Name Type Description
name str

Name of the reference to a BMI-compliant model component instance.

Source code in hydrolib/core/dimr/models.py
169
170
171
172
173
174
175
176
177
178
179
180
class ComponentOrCouplerRef(BaseModel):
    """
    Reference to a BMI-compliant model component instance.

    Attributes:
        name: Name of the reference to a BMI-compliant model component instance.
    """

    name: str

    def _get_identifier(self, data: dict) -> Optional[str]:
        return data.get("name")

ControlModel

Bases: BaseModel

Overrides to make sure that the control elements in the DIMR are parsed and serialized correctly.

Source code in hydrolib/core/dimr/models.py
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
class ControlModel(BaseModel):
    """
    Overrides to make sure that the control elements in the DIMR
    are parsed and serialized correctly.
    """

    _type: str

    def dict(self, *args, **kwargs):
        """Add control element prefixes for serialized data."""
        return {
            str(self._type): super().dict(*args, **kwargs),
        }

    @classmethod
    def validate(cls, v):
        """Remove control element prefixes from parsed data."""

        # should be replaced by discriminated unions once merged
        # https://github.com/samuelcolvin/pydantic/pull/2336
        if isinstance(v, dict) and len(v.keys()) == 1:
            key = list(v.keys())[0]
            v = v[key]
        return super().validate(v)

dict(*args, **kwargs)

Add control element prefixes for serialized data.

Source code in hydrolib/core/dimr/models.py
271
272
273
274
275
def dict(self, *args, **kwargs):
    """Add control element prefixes for serialized data."""
    return {
        str(self._type): super().dict(*args, **kwargs),
    }

validate(v) classmethod

Remove control element prefixes from parsed data.

Source code in hydrolib/core/dimr/models.py
277
278
279
280
281
282
283
284
285
286
@classmethod
def validate(cls, v):
    """Remove control element prefixes from parsed data."""

    # should be replaced by discriminated unions once merged
    # https://github.com/samuelcolvin/pydantic/pull/2336
    if isinstance(v, dict) and len(v.keys()) == 1:
        key = list(v.keys())[0]
        v = v[key]
    return super().validate(v)

CoupledItem

Bases: BaseModel

Specification of an item that has to be exchanged.

Attributes:

Name Type Description
sourceName str

Name of the item at the source component.

targetName str

Name of the item at the target component.

Source code in hydrolib/core/dimr/models.py
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
class CoupledItem(BaseModel):
    """
    Specification of an item that has to be exchanged.

    Attributes:
        sourceName: Name of the item at the source component.
        targetName: Name of the item at the target component.
    """

    sourceName: str
    targetName: str

    def is_intermediate_link(self) -> bool:
        # TODO set to True once we replace Paths with FileModels
        return False

Coupler

Bases: BaseModel

Specification of the coupling actions to be performed between two BMI-compliant model components.

Attributes:

Name Type Description
name str

The name of the coupler.

sourceComponent str

The component that provides the data to has to be exchanged.

targetComponent str

The component that consumes the data to has to be exchanged.

item List[CoupledItem]

A list of items that have to be exchanged.

logger Optional[Logger]

Logger for logging the values that get exchanged.

Source code in hydrolib/core/dimr/models.py
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
class Coupler(BaseModel):
    """
    Specification of the coupling actions to be performed between two BMI-compliant model components.

    Attributes:
        name: The name of the coupler.
        sourceComponent: The component that provides the data to has to be exchanged.
        targetComponent: The component that consumes the data to has to be exchanged.
        item: A list of items that have to be exchanged.
        logger: Logger for logging the values that get exchanged.
    """

    name: str
    sourceComponent: str
    targetComponent: str
    item: List[CoupledItem] = []
    logger: Optional[Logger]

    @validator("item", pre=True)
    def validate_item(cls, v):
        return to_list(v)

    def is_intermediate_link(self) -> bool:
        # TODO set to True once we replace Paths with FileModels
        return False

    def _get_identifier(self, data: dict) -> Optional[str]:
        return data.get("name")

DIMR

Bases: ParsableFileModel

DIMR model representation.

Attributes:

Name Type Description
documentation Documentation

File metadata.

control List[Union[Start, Parallel]]

The <control> element with a list of Start and Parallel sub-elements, which defines the (sequence of) program(s) to be run. May be empty while constructing, but must be non-empty when saving! Also, all referenced components must be present in component when saving. Similarly, all referenced couplers must be present in coupler.

component List[Union[RRComponent, FMComponent, Component]]

List of <component> elements that defines which programs can be used inside the <control> subelements. Must be non-empty when saving!

coupler Optional[List[Coupler]]

optional list of <coupler> elements that defines which couplers can be used inside the <parallel> elements under <control>.

waitFile Optional[str]

Optional waitfile name for debugging.

global_settings Optional[GlobalSettings]

Optional global DIMR settings.

Source code in hydrolib/core/dimr/models.py
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
class DIMR(ParsableFileModel):
    """DIMR model representation.

    Attributes:
        documentation (Documentation): File metadata.
        control (List[Union[Start, Parallel]]): The `<control>` element with a list
            of [Start][hydrolib.core.dimr.models.Start]
            and [Parallel][hydrolib.core.dimr.models.Parallel] sub-elements,
            which defines the (sequence of) program(s) to be run.
            May be empty while constructing, but must be non-empty when saving!
            Also, all referenced components must be present in `component` when
            saving. Similarly, all referenced couplers must be present in `coupler`.
        component (List[Union[RRComponent, FMComponent, Component]]): List of
            `<component>` elements that defines which programs can be used inside
            the `<control>` subelements. Must be non-empty when saving!
        coupler (Optional[List[Coupler]]): optional list of `<coupler>` elements
            that defines which couplers can be used inside the `<parallel>`
            elements under `<control>`.
        waitFile (Optional[str]): Optional waitfile name for debugging.
        global_settings (Optional[GlobalSettings]): Optional global DIMR settings.
    """

    documentation: Documentation = Documentation()
    control: List[Union[Start, Parallel]] = Field([])
    component: List[Union[RRComponent, FMComponent, Component]] = []
    coupler: Optional[List[Coupler]] = []
    waitFile: Optional[str]
    global_settings: Optional[GlobalSettings]

    @validator("component", "coupler", "control", pre=True)
    def validate_component(cls, v):
        return to_list(v)

    def dict(self, *args, **kwargs):
        kwargs["exclude_none"] = True
        return super().dict(*args, **kwargs)

    def _post_init_load(self) -> None:
        """
        Load the component models of this DIMR model.
        """
        super()._post_init_load()

        for comp in self.component:
            try:
                comp.model = comp.get_model()(filepath=comp.filepath)
            except NotImplementedError:
                pass

    def _serialize(self, data: dict, save_settings: ModelSaveSettings) -> None:
        dimr_as_dict = self._update_dimr_dictonary_with_adjusted_fmcomponent_values(
            data
        )
        super()._serialize(dimr_as_dict, save_settings)

    def _update_dimr_dictonary_with_adjusted_fmcomponent_values(
        self, dimr_as_dict: Dict
    ):
        fmcomponents = [
            item for item in self.component if isinstance(item, FMComponent)
        ]

        list_of_fmcomponents_as_dict = self._get_list_of_updated_fm_components(
            fmcomponents
        )
        dimr_as_dict = self._update_dimr_dictionary(
            dimr_as_dict, list_of_fmcomponents_as_dict
        )
        return dimr_as_dict

    def _update_dimr_dictionary(
        self, dimr_as_dict: Dict, list_of_fm_components_as_dict: List[Dict]
    ) -> Dict:
        if len(list_of_fm_components_as_dict) > 0:
            dimr_as_dict.update({"component": list_of_fm_components_as_dict})

        return dimr_as_dict

    def _get_list_of_updated_fm_components(
        self, fmcomponents: List[FMComponent]
    ) -> List[Dict]:
        list_of_fm_components_as_dict = []
        for fmcomponent in fmcomponents:
            if fmcomponent is None or fmcomponent.process is None:
                continue

            if fmcomponent.process == 1:
                fmcomponent_as_dict = fmcomponent.dict()
                fmcomponent_as_dict.pop("process", None)
            else:
                fmcomponent_process_value = " ".join(
                    str(i) for i in range(fmcomponent.process)
                )
                fmcomponent_as_dict = self._update_component_dictonary(
                    fmcomponent, fmcomponent_process_value
                )

            list_of_fm_components_as_dict.append(fmcomponent_as_dict)

        return list_of_fm_components_as_dict

    def _update_component_dictonary(
        self, fmcomponent: FMComponent, fmcomponent_process_value: str
    ) -> Dict:
        fmcomponent_as_dict = fmcomponent.dict()
        fmcomponent_as_dict.update({"process": fmcomponent_process_value})
        return fmcomponent_as_dict

    @classmethod
    def _ext(cls) -> str:
        return ".xml"

    @classmethod
    def _filename(cls) -> str:
        return "dimr_config"

    @classmethod
    def _get_serializer(
        cls,
    ) -> Callable[[Path, Dict, SerializerConfig, ModelSaveSettings], None]:
        return DIMRSerializer.serialize

    @classmethod
    def _get_parser(cls) -> Callable:
        return DIMRParser.parse

    @classmethod
    def _parse(cls, path: Path) -> Dict:
        data = super()._parse(path)
        return cls._update_component(data)

    @classmethod
    def _update_component(cls, data: Dict) -> Dict:
        component = data.get("component", None)

        if not isinstance(component, Dict):
            return data

        process_value = component.get("process", None)

        if not isinstance(process_value, str):
            return data

        if cls._is_valid_process_string(process_value):
            value_as_int = cls._parse_process(process_value)
            component.update({"process": value_as_int})
            data.update({"component": component})

        return data

    @classmethod
    def _parse_process(cls, process_value: str) -> int:
        if ":" in process_value:
            semicolon_split_values = process_value.split(":")
            start_value = int(semicolon_split_values[0])
            end_value = int(semicolon_split_values[-1])
            return end_value - start_value + 1

        return len(process_value.split())

    @classmethod
    def _is_valid_process_string(cls, process_value: str) -> bool:
        if ":" in process_value:
            return cls._is_valid_process_with_semicolon_string(process_value)

        return cls._is_valid_process_list_string(process_value)

    @classmethod
    def _is_valid_process_with_semicolon_string(cls, process_value: str) -> bool:
        semicolon_split_values = process_value.split(":")

        if len(semicolon_split_values) != 2:
            return False

        last_value: str = semicolon_split_values[-1]
        if last_value.isdigit():
            return True

        return False

    @classmethod
    def _is_valid_process_list_string(cls, process_value: str) -> bool:
        split_values = process_value.split()

        if len(split_values) < 1:
            return False

        if split_values[0] != "0":
            return False

        for value in split_values:
            if not value.isdigit():
                return False

        return True

Documentation

Bases: BaseModel

Information on the present DIMR configuration file.

Attributes:

Name Type Description
fileVersion str

The DIMR file version.

createdBy str

Creators of the DIMR file.

creationDate datetime

The creation date of the DIMR file.

Source code in hydrolib/core/dimr/models.py
143
144
145
146
147
148
149
150
151
152
153
154
155
class Documentation(BaseModel):
    """
    Information on the present DIMR configuration file.

    Attributes:
        fileVersion: The DIMR file version.
        createdBy: Creators of the DIMR file.
        creationDate: The creation date of the DIMR file.
    """

    fileVersion: str = "1.3"
    createdBy: str = f"hydrolib-core {__version__}"
    creationDate: datetime = Field(default_factory=datetime.utcnow)

FMComponent

Bases: Component

Component to include the D-Flow FM program in a DIMR control flow.

Source code in hydrolib/core/dimr/models.py
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class FMComponent(Component):
    """Component to include the D-Flow FM program in a DIMR control flow."""

    library: Literal["dflowfm"] = "dflowfm"

    @validator("process", pre=True)
    def validate_process(cls, value, values: dict) -> Union[None, int]:
        """
        Validation for the process Attribute.

        args:
            value : The value which is to be validated for process.
            values : FMComponent used to retrieve the name of the component.

        Returns:
            int : The process as int, when given value is None, None is returned.

        Raises:
            ValueError : When value is set to 0 or negative.
            ValueError : When value is not int or None.
        """
        if value is None:
            return value

        if isinstance(value, int) and cls._is_valid_process_int(
            value, values.get("name")
        ):
            return value

        raise ValueError(
            f"In component '{values.get('name')}', the keyword process '{value}', is incorrect."
        )

    @classmethod
    def _is_valid_process_int(cls, value: int, name: str) -> bool:
        if value > 0:
            return True

        raise ValueError(
            f"In component '{name}', the keyword process can not be 0 or negative, please specify value of 1 or greater."
        )

    @classmethod
    def get_model(cls):
        return FMModel

validate_process(value, values)

Validation for the process Attribute.

Parameters:

Name Type Description Default
value

The value which is to be validated for process.

required
values

FMComponent used to retrieve the name of the component.

required

Returns:

Name Type Description
int Union[None, int]

The process as int, when given value is None, None is returned.

Raises:

Type Description
ValueError

When value is set to 0 or negative.

ValueError

When value is not int or None.

Source code in hydrolib/core/dimr/models.py
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@validator("process", pre=True)
def validate_process(cls, value, values: dict) -> Union[None, int]:
    """
    Validation for the process Attribute.

    args:
        value : The value which is to be validated for process.
        values : FMComponent used to retrieve the name of the component.

    Returns:
        int : The process as int, when given value is None, None is returned.

    Raises:
        ValueError : When value is set to 0 or negative.
        ValueError : When value is not int or None.
    """
    if value is None:
        return value

    if isinstance(value, int) and cls._is_valid_process_int(
        value, values.get("name")
    ):
        return value

    raise ValueError(
        f"In component '{values.get('name')}', the keyword process '{value}', is incorrect."
    )

GlobalSettings

Bases: BaseModel

Global settings for the DIMR configuration.

Attributes:

Name Type Description
logger_ncFormat int

NetCDF format type for logging.

Source code in hydrolib/core/dimr/models.py
158
159
160
161
162
163
164
165
166
class GlobalSettings(BaseModel):
    """
    Global settings for the DIMR configuration.

    Attributes:
        logger_ncFormat: NetCDF format type for logging.
    """

    logger_ncFormat: int

KeyValuePair

Bases: BaseModel

Key value pair to specify settings and parameters.

Attributes:

Name Type Description
key str

The key.

value str

The value.

Source code in hydrolib/core/dimr/models.py
23
24
25
26
27
28
29
30
31
32
class KeyValuePair(BaseModel):
    """Key value pair to specify settings and parameters.

    Attributes:
        key: The key.
        value: The value.
    """

    key: str
    value: str

Logger

Bases: BaseModel

Used to log values to the specified file in workingdir for each timestep

Attributes:

Name Type Description
workingDir Path

Directory where the log file is written.

outputFile Path

Name of the log file.

Source code in hydrolib/core/dimr/models.py
200
201
202
203
204
205
206
207
208
209
210
class Logger(BaseModel):
    """
    Used to log values to the specified file in workingdir for each timestep

    Attributes:
        workingDir: Directory where the log file is written.
        outputFile: Name of the log file.
    """

    workingDir: Path
    outputFile: Path

Parallel

Bases: ControlModel

Specification of a parallel control flow: one main component and a group of related components and couplers. Step wise execution order according to order in parallel control flow.

Attributes:

Name Type Description
startGroup StartGroup

Group of components and couplers to be executed.

start ComponentOrCouplerRef

Main component to be executed step wise (provides start time, end time and time step).

Source code in hydrolib/core/dimr/models.py
289
290
291
292
293
294
295
296
297
298
299
300
301
class Parallel(ControlModel):
    """
    Specification of a parallel control flow: one main component and a group of related components and couplers.
    Step wise execution order according to order in parallel control flow.

    Attributes:
        startGroup: Group of components and couplers to be executed.
        start: Main component to be executed step wise (provides start time, end time and time step).
    """

    _type: Literal["parallel"] = "parallel"
    startGroup: StartGroup
    start: ComponentOrCouplerRef

RRComponent

Bases: Component

Component to include the RainfallRunoff program in a DIMR control flow.

Source code in hydrolib/core/dimr/models.py
133
134
135
136
137
138
139
140
class RRComponent(Component):
    """Component to include the RainfallRunoff program in a DIMR control flow."""

    library: Literal["rr_dll"] = "rr_dll"

    @classmethod
    def get_model(cls):
        return RainfallRunoffModel

Start

Bases: ControlModel

Specification of a serial control flow: one main component.

Attributes:

Name Type Description
name str

Name of the reference to a BMI-compliant model component instance

Source code in hydrolib/core/dimr/models.py
304
305
306
307
308
309
310
311
312
313
class Start(ControlModel):
    """
    Specification of a serial control flow: one main component.

    Attributes:
        name: Name of the reference to a BMI-compliant model component instance
    """

    _type: Literal["start"] = "start"
    name: str

StartGroup

Bases: BaseModel

Specification of model components and couplers to be executed with a certain frequency.

Attributes:

Name Type Description
time str

Time frame specification for the present group: start time, stop time and frequency. Expressed in terms of the time frame of the main component.

start List[ComponentOrCouplerRef]

Ordered list of components to be executed.

coupler List[ComponentOrCouplerRef]

Oredered list of couplers to be executed.

Source code in hydrolib/core/dimr/models.py
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
class StartGroup(BaseModel):
    """
    Specification of model components and couplers to be executed with a certain frequency.

    Attributes:
        time: Time frame specification for the present group: start time, stop time and frequency.
              Expressed in terms of the time frame of the main component.
        start: Ordered list of components to be executed.
        coupler: Oredered list of couplers to be executed.
    """

    time: str
    start: List[ComponentOrCouplerRef] = []
    coupler: List[ComponentOrCouplerRef] = []

    @validator("start", "coupler", pre=True)
    def validate_start(cls, v):
        return to_list(v)

Parser

DIMRParser

A parser for DIMR xml files.

Source code in hydrolib/core/dimr/parser.py
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class DIMRParser:
    """A parser for DIMR xml files."""

    @staticmethod
    def parse(path: Path) -> dict:
        """Parses a DIMR file to a dictionary.

        Args:
            path (Path): Path to the DIMR configuration file.
        """
        if not path.is_file():
            warn(f"File: `{path}` not found, skipped parsing.")
            return {}

        parser = etree.XMLParser(
            remove_comments=True, resolve_entities=False, no_network=True
        )
        root = etree.parse(str(path), parser=parser).getroot()

        return DIMRParser._node_to_dictionary(root, True)

    @staticmethod
    def _node_to_dictionary(node: etree, ignore_attributes: bool = False):
        """
        Convert an lxml.etree node tree recursively into a nested dictionary.
        The node's attributes and child items will be added to it's dictionary.

        Args:
            node (etree): The etree node
            ignore_attributes (bool): Optional parameter; whether or not to
                                      skip the node's attributes. Default is False.
        """

        result = {} if ignore_attributes else dict(node.attrib)

        for child_node in node.iterchildren():

            key = child_node.tag.split("}")[1]

            if child_node.text and child_node.text.strip():
                value = child_node.text
            else:
                value = DIMRParser._node_to_dictionary(child_node)

            if key in result:

                if type(result[key]) is list:
                    result[key].append(value)
                else:
                    first_value = result[key].copy()
                    result[key] = [first_value, value]
            else:
                result[key] = value

        return result

parse(path) staticmethod

Parses a DIMR file to a dictionary.

Parameters:

Name Type Description Default
path Path

Path to the DIMR configuration file.

required
Source code in hydrolib/core/dimr/parser.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@staticmethod
def parse(path: Path) -> dict:
    """Parses a DIMR file to a dictionary.

    Args:
        path (Path): Path to the DIMR configuration file.
    """
    if not path.is_file():
        warn(f"File: `{path}` not found, skipped parsing.")
        return {}

    parser = etree.XMLParser(
        remove_comments=True, resolve_entities=False, no_network=True
    )
    root = etree.parse(str(path), parser=parser).getroot()

    return DIMRParser._node_to_dictionary(root, True)

Serializer

DIMRSerializer

A serializer for DIMR files.

Source code in hydrolib/core/dimr/serializer.py
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class DIMRSerializer:
    """A serializer for DIMR files."""

    @staticmethod
    def serialize(
        path: Path,
        data: dict,
        config: SerializerConfig,
        save_settings: ModelSaveSettings,
    ):
        """
        Serializes the DIMR data to the file at the specified path.

        Attributes:
            path (Path): The path to the destination file.
            data (Dict): The data to be serialized.
            config (SerializerConfig): The serialization configuration.
            save_settings (ModelSaveSettings): The model save settings.
        """

        path.parent.mkdir(parents=True, exist_ok=True)

        xmlns = "http://schemas.deltares.nl/dimr"
        xsi = "http://www.w3.org/2001/XMLSchema-instance"
        schema_location = "http://content.oss.deltares.nl/schemas/dimr-1.3.xsd"

        attrib = {e.QName(xsi, "schemaLocation"): f"{xmlns} {schema_location}"}
        namespaces = {None: xmlns, "xsi": xsi}

        root = e.Element(
            "dimrConfig",
            attrib=attrib,
            nsmap=namespaces,
        )

        path_style_converter = FilePathStyleConverter()
        DIMRSerializer._build_tree(
            root, data, config, save_settings, path_style_converter
        )

        to_string = minidom.parseString(e.tostring(root))
        xml = to_string.toprettyxml(indent="  ", encoding="utf-8")

        with path.open("wb") as f:
            f.write(xml)

    @staticmethod
    def _build_tree(
        root,
        data: dict,
        config: SerializerConfig,
        save_settings: ModelSaveSettings,
        path_style_converter: FilePathStyleConverter,
    ):
        name = data.pop("name", None)
        if name:
            root.set("name", name)

        for key, val in data.items():
            if isinstance(val, dict):
                c = e.Element(key)
                DIMRSerializer._build_tree(
                    c, val, config, save_settings, path_style_converter
                )
                root.append(c)
            elif isinstance(val, List):
                for item in val:
                    c = e.Element(key)
                    DIMRSerializer._build_tree(
                        c, item, config, save_settings, path_style_converter
                    )
                    root.append(c)
            else:
                c = e.Element(key)
                if isinstance(val, datetime):
                    c.text = val.isoformat(sep="T", timespec="auto")
                elif isinstance(val, float):
                    c.text = f"{val:{config.float_format}}"
                elif isinstance(val, Path):
                    c.text = path_style_converter.convert_from_os_style(
                        val, save_settings.path_style
                    )
                else:
                    c.text = str(val)
                root.append(c)

serialize(path, data, config, save_settings) staticmethod

Serializes the DIMR data to the file at the specified path.

Attributes:

Name Type Description
path Path

The path to the destination file.

data Dict

The data to be serialized.

config SerializerConfig

The serialization configuration.

save_settings ModelSaveSettings

The model save settings.

Source code in hydrolib/core/dimr/serializer.py
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@staticmethod
def serialize(
    path: Path,
    data: dict,
    config: SerializerConfig,
    save_settings: ModelSaveSettings,
):
    """
    Serializes the DIMR data to the file at the specified path.

    Attributes:
        path (Path): The path to the destination file.
        data (Dict): The data to be serialized.
        config (SerializerConfig): The serialization configuration.
        save_settings (ModelSaveSettings): The model save settings.
    """

    path.parent.mkdir(parents=True, exist_ok=True)

    xmlns = "http://schemas.deltares.nl/dimr"
    xsi = "http://www.w3.org/2001/XMLSchema-instance"
    schema_location = "http://content.oss.deltares.nl/schemas/dimr-1.3.xsd"

    attrib = {e.QName(xsi, "schemaLocation"): f"{xmlns} {schema_location}"}
    namespaces = {None: xmlns, "xsi": xsi}

    root = e.Element(
        "dimrConfig",
        attrib=attrib,
        nsmap=namespaces,
    )

    path_style_converter = FilePathStyleConverter()
    DIMRSerializer._build_tree(
        root, data, config, save_settings, path_style_converter
    )

    to_string = minidom.parseString(e.tostring(root))
    xml = to_string.toprettyxml(indent="  ", encoding="utf-8")

    with path.open("wb") as f:
        f.write(xml)