INI format

The ini module provides the generic logic for parsing Deltares ini based files, such as the mdu, structures files, as well as more complex files such as the boundary condition (bc) files.

Note that specific attribute files that employ this ini format often have their own dedicated module (and separate API doc page). These include:

• MDU file (hydrolib.core.dflowfm.mdu)
• External forcing file (hydrolib.core.dflowfm.ext)
• Initial fields and parameter file (hydrolib.core.dflowfm.inifield)
• Structure file (hydrolib.core.dflowfm.structure)
• 1D roughness (hydrolib.core.dflowfm.friction)
• Cross section files (hydrolib.core.dflowfm.crosssection)
• Storage node file (hydrolib.core.dflowfm.storagenode)

Following below is the documentation for the INI format base classes.

Model

DataBlockINIBasedModel

Bases: INIBasedModel

DataBlockINIBasedModel defines the base model for ini models with datablocks.

Attributes:

Name	Type	Description
datablock	Datablock	(class attribute) the actual data columns.

Source code in hydrolib/core/dflowfm/ini/models.py

class DataBlockINIBasedModel(INIBasedModel):
    """DataBlockINIBasedModel defines the base model for ini models with datablocks.

    Attributes:
        datablock (Datablock): (class attribute) the actual data columns.
    """

    datablock: Datablock = []

    _make_lists = make_list_validator("datablock")

    @classmethod
    def _get_unknown_keyword_error_manager(cls) -> Optional[UnknownKeywordErrorManager]:
        """
        The DataBlockINIBasedModel does not need to raise an error on unknown keywords.
        """
        return None

    def _to_section(
        self,
        config: DataBlockINIBasedSerializerConfig,
        save_settings: ModelSaveSettings,
    ) -> Section:
        section = super()._to_section(config, save_settings)
        section.datablock = self._to_datablock(config)
        return section

    def _to_datablock(self, config: DataBlockINIBasedSerializerConfig) -> List[List]:
        converted_datablock = []

        for row in self.datablock:
            converted_row = (
                DataBlockINIBasedModel.convert_value(value, config) for value in row
            )
            converted_datablock.append(list(converted_row))

        return converted_datablock

    @classmethod
    def convert_value(
        cls, value: Union[float, str], config: DataBlockINIBasedSerializerConfig
    ) -> str:
        if isinstance(value, float):
            return f"{value:{config.float_format_datablock}}"

        return value

    @validator("datablock")
    def _validate_no_nans_are_present(cls, datablock: Datablock) -> Datablock:
        """Validate that the datablock does not have any NaN values.

        Args:
            datablock (Datablock): The datablock to verify.

        Raises:
            ValueError: When a NaN is present in the datablock.

        Returns:
            Datablock: The validated datablock.
        """
        if any(cls._is_float_and_nan(value) for list in datablock for value in list):
            raise ValueError("NaN is not supported in datablocks.")

        return datablock

    @staticmethod
    def _is_float_and_nan(value: float) -> bool:
        return isinstance(value, float) and isnan(value)

INIBasedModel

Bases: BaseModel, ABC

INIBasedModel defines the base model for blocks/chapters inside an INIModel (*.ini file).

INIBasedModel instances can be created from Section instances obtained through parsing ini documents. It further supports adding arbitrary fields to it, which will be written to file. Lastly, no arbitrary types are allowed for the defined fields.

Attributes:

Name	Type	Description
comments	Optional[Comments]	Optional Comments if defined by the user, containing descriptions for all data fields.

Source code in hydrolib/core/dflowfm/ini/models.py

class INIBasedModel(BaseModel, ABC):
    """INIBasedModel defines the base model for blocks/chapters
    inside an INIModel (*.ini file).

    INIBasedModel instances can be created from Section instances
    obtained through parsing ini documents. It further supports
    adding arbitrary fields to it, which will be written to file.
    Lastly, no arbitrary types are allowed for the defined fields.

    Attributes:
        comments (Optional[Comments]):
            Optional Comments if defined by the user, containing
            descriptions for all data fields.
    """

    _header: str = ""
    _file_path_style_converter = FilePathStyleConverter()

    class Config:
        extra = Extra.ignore
        arbitrary_types_allowed = False

    @classmethod
    def _get_unknown_keyword_error_manager(cls) -> Optional[UnknownKeywordErrorManager]:
        return UnknownKeywordErrorManager()

    @classmethod
    def _supports_comments(cls):
        return True

    @classmethod
    def _duplicate_keys_as_list(cls):
        return False

    @classmethod
    def get_list_delimiter(cls) -> str:
        """List delimiter string that will be used for serializing
        list field values for any IniBasedModel, **if** that field has
        no custom list delimiter.

        This function should be overridden by any subclass for a particular
        filetype that needs a specific/different list separator.
        """
        return " "

    @classmethod
    def get_list_field_delimiter(cls, field_key: str) -> str:
        """List delimiter string that will be used for serializing
        the given field's value.
        The returned delimiter is either the field's custom list delimiter
        if that was specified using Field(.., delimiter=".."), or the
        default list delimiter for the model class that this field belongs
        to.

        Args:
            field_key (str): the original field key (not its alias).
        """
        delimiter = None
        if (field := cls.__fields__.get(field_key)) and isinstance(field, ModelField):
            delimiter = field.field_info.extra.get("delimiter")
        if not delimiter:
            delimiter = cls.get_list_delimiter()

        return delimiter

    class Comments(BaseModel, ABC):
        """Comments defines the comments of an INIBasedModel"""

        class Config:
            extra = Extra.allow
            arbitrary_types_allowed = False

    comments: Optional[Comments] = Comments()

    @root_validator(pre=True)
    def _validate_unknown_keywords(cls, values):
        unknown_keyword_error_manager = cls._get_unknown_keyword_error_manager()
        if unknown_keyword_error_manager:
            unknown_keyword_error_manager.raise_error_for_unknown_keywords(
                values,
                cls._header,
                cls.__fields__,
                cls._exclude_fields(),
            )
        return values

    @root_validator(pre=True)
    def _skip_nones_and_set_header(cls, values):
        """Drop None fields for known fields."""
        dropkeys = []
        for k, v in values.items():
            if v is None and k in cls.__fields__.keys():
                dropkeys.append(k)

        logger.info(f"Dropped unset keys: {dropkeys}")
        for k in dropkeys:
            values.pop(k)

        if "_header" in values:
            values["_header"] = cls._header

        return values

    @validator("comments", always=True, allow_reuse=True)
    def comments_matches_has_comments(cls, v):
        if not cls._supports_comments() and v is not None:
            logging.warning(f"Dropped unsupported comments from {cls.__name__} init.")
            v = None
        return v

    @validator("*", pre=True, allow_reuse=True)
    def replace_fortran_scientific_notation_for_floats(cls, value, field):
        if field.type_ != float:
            return value

        return FortranUtils.replace_fortran_scientific_notation(value)

    @classmethod
    def validate(cls: Type["INIBasedModel"], value: Any) -> "INIBasedModel":
        if isinstance(value, Section):
            value = value.flatten(
                cls._duplicate_keys_as_list(), cls._supports_comments()
            )

        return super().validate(value)

    @classmethod
    def _exclude_fields(cls) -> Set:
        return {"comments", "datablock", "_header"}

    def _convert_value(
        self,
        key: str,
        v: Any,
        config: INISerializerConfig,
        save_settings: ModelSaveSettings,
    ) -> str:
        if isinstance(v, bool):
            return str(int(v))
        elif isinstance(v, list):
            convert_value = lambda x: self._convert_value(key, x, config, save_settings)
            return self.__class__.get_list_field_delimiter(key).join(
                [convert_value(x) for x in v]
            )
        elif isinstance(v, Enum):
            return v.value
        elif isinstance(v, float):
            return f"{v:{config.float_format}}"
        elif isinstance(v, FileModel) and v.filepath is not None:
            return self._file_path_style_converter.convert_from_os_style(
                v.filepath, save_settings.path_style
            )
        elif v is None:
            return ""
        else:
            return str(v)

    def _to_section(
        self, config: INISerializerConfig, save_settings: ModelSaveSettings
    ) -> Section:
        props = []
        for key, value in self:
            if not self._should_be_serialized(key, value):
                continue

            field_key = key
            if key in self.__fields__:
                key = self.__fields__[key].alias

            prop = Property(
                key=key,
                value=self._convert_value(field_key, value, config, save_settings),
                comment=getattr(self.comments, field_key, None),
            )
            props.append(prop)
        return Section(header=self._header, content=props)

    def _should_be_serialized(self, key: str, value: Any) -> bool:
        if key in self._exclude_fields():
            return False

        field = self.__fields__.get(key)
        if not field:
            return value is not None

        field_type = field.type_
        if self._is_union(field_type):
            return value is not None or self._union_has_filemodel(field_type)

        if self._is_list(field_type):
            field_type = get_args(field_type)[0]

        return self._value_is_not_none_or_type_is_filemodel(field_type, value)

    @staticmethod
    def _is_union(field_type: type) -> bool:
        return get_origin(field_type) is Union

    @staticmethod
    def _union_has_filemodel(field_type: type) -> bool:
        return any(issubclass(arg, FileModel) for arg in get_args(field_type))

    @staticmethod
    def _is_list(field_type: type) -> bool:
        return get_origin(field_type) is List

    @staticmethod
    def _value_is_not_none_or_type_is_filemodel(field_type: type, value: Any) -> bool:
        return value is not None or issubclass(field_type, FileModel)

Comments

Bases: BaseModel, ABC

Comments defines the comments of an INIBasedModel

Source code in hydrolib/core/dflowfm/ini/models.py

class Comments(BaseModel, ABC):
    """Comments defines the comments of an INIBasedModel"""

    class Config:
        extra = Extra.allow
        arbitrary_types_allowed = False

get_list_delimiter() classmethod

List delimiter string that will be used for serializing list field values for any IniBasedModel, if that field has no custom list delimiter.

This function should be overridden by any subclass for a particular filetype that needs a specific/different list separator.

Source code in hydrolib/core/dflowfm/ini/models.py

@classmethod
def get_list_delimiter(cls) -> str:
    """List delimiter string that will be used for serializing
    list field values for any IniBasedModel, **if** that field has
    no custom list delimiter.

    This function should be overridden by any subclass for a particular
    filetype that needs a specific/different list separator.
    """
    return " "

get_list_field_delimiter(field_key) classmethod

List delimiter string that will be used for serializing the given field's value. The returned delimiter is either the field's custom list delimiter if that was specified using Field(.., delimiter=".."), or the default list delimiter for the model class that this field belongs to.

Parameters:

Name	Type	Description	Default
field_key	str	the original field key (not its alias).	required

Source code in hydrolib/core/dflowfm/ini/models.py

@classmethod
def get_list_field_delimiter(cls, field_key: str) -> str:
    """List delimiter string that will be used for serializing
    the given field's value.
    The returned delimiter is either the field's custom list delimiter
    if that was specified using Field(.., delimiter=".."), or the
    default list delimiter for the model class that this field belongs
    to.

    Args:
        field_key (str): the original field key (not its alias).
    """
    delimiter = None
    if (field := cls.__fields__.get(field_key)) and isinstance(field, ModelField):
        delimiter = field.field_info.extra.get("delimiter")
    if not delimiter:
        delimiter = cls.get_list_delimiter()

    return delimiter

INIModel

Bases: ParsableFileModel

INI Model representation of a *.ini file.

Typically subclasses will implement the various sorts of ini files, specifically for their fileType/contents. Child elements of this class associated with chapters/blocks in the ini file will be (sub)class of INIBasedModel.

Source code in hydrolib/core/dflowfm/ini/models.py

class INIModel(ParsableFileModel):
    """INI Model representation of a *.ini file.

    Typically subclasses will implement the various sorts of ini files,
    specifically for their fileType/contents.
    Child elements of this class associated with chapters/blocks in the
    ini file will be (sub)class of INIBasedModel.
    """

    serializer_config: INISerializerConfig = INISerializerConfig()

    general: INIGeneral

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

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

    @classmethod
    def _get_serializer(cls):
        pass  # unused in favor of direct _serialize

    @classmethod
    def _get_parser(cls) -> Callable:
        return Parser.parse_as_dict

    def _to_document(self, save_settings: ModelSaveSettings) -> Document:
        header = CommentBlock(lines=[f"written by HYDROLIB-core {version}"])
        sections = []
        for key, value in self:
            if key in self._exclude_fields() or value is None:
                continue
            if isinstance(value, list):
                for v in value:
                    sections.append(
                        v._to_section(self.serializer_config, save_settings)
                    )
            else:
                sections.append(
                    value._to_section(self.serializer_config, save_settings)
                )
        return Document(header_comment=[header], sections=sections)

    def _serialize(self, _: dict, save_settings: ModelSaveSettings) -> None:
        write_ini(
            self._resolved_filepath,
            self._to_document(save_settings),
            config=self.serializer_config,
        )

Parser

Parser

Parser defines a generic Parser for Deltares ini files.

The Parser can be configured with a ParserConfig object.

Source code in hydrolib/core/dflowfm/ini/parser.py

class Parser:
    """Parser defines a generic Parser for Deltares ini files.

    The Parser can be configured with a ParserConfig object.
    """

    class _StateType(IntEnum):
        NO_SECTION_FOUND = 0
        PARSING_PROPERTIES = 1
        PARSING_DATABLOCK = 2

    def __init__(self, config: ParserConfig) -> None:
        """Creates a new Parser configured with the provided config

        Args:
            config (ParserConfig): The configuration of this Parser
        """
        self._config = config
        self._document = Document()
        self._current_section: Optional[_IntermediateSection] = None
        self._current_header_block: Optional[_IntermediateCommentBlock] = None

        self._state = self._StateType.NO_SECTION_FOUND
        self._line_index = 0

        # TODO add invalid blocks
        self._feed_line: Dict[
            Parser._StateType, List[Tuple[Callable[[str], bool], Callable[[str], None]]]
        ] = {
            Parser._StateType.NO_SECTION_FOUND: [
                (self._is_comment, self._handle_header_comment),
                (self._is_section_header, self._handle_next_section_header),
            ],
            Parser._StateType.PARSING_PROPERTIES: [
                (self._is_comment, self._handle_section_comment),
                (self._is_section_header, self._handle_next_section_header),
                (self._is_property, self._handle_property),
                (self._is_datarow, self._handle_new_datarow),
            ],
            Parser._StateType.PARSING_DATABLOCK: [
                (self._is_section_header, self._handle_next_section_header),
                (self._is_datarow, self._handle_datarow),
            ],
        }

        self._handle_emptyline: Dict[Parser._StateType, Callable[[], None]] = {
            self._StateType.NO_SECTION_FOUND: self._finish_current_header_block,
            self._StateType.PARSING_PROPERTIES: self._noop,
            self._StateType.PARSING_DATABLOCK: self._noop,
        }

    def feed_line(self, line: str) -> None:
        """Parse the next line with this Parser.

        Args:
            line (str): The line to parse
        """
        if not self._is_empty_line(line):
            for is_line_type, handle_line_type in self._feed_line[self._state]:
                if is_line_type(line):
                    handle_line_type(line)
                    break
            else:
                # handle exception
                pass
        else:
            self._handle_emptyline[self._state]()

        self._increment_line()

    def finalize(self) -> Document:
        """Finalize parsing and return the constructed Document.

        Returns:
            Document:
                A Document describing the parsed ini file.
        """
        # TODO handle invalid block
        self._finish_current_header_block()
        self._finalise_current_section()
        return self._document

    def _increment_line(self) -> None:
        self._line_index += 1

    def _handle_next_section_header(self, line: str) -> None:
        self._finalise_current_section()
        self._handle_new_section_header(line)

        self._state = Parser._StateType.PARSING_PROPERTIES

    def _handle_new_section_header(self, line: str) -> None:
        section_header = line.strip()[1:-1].strip()
        self._current_section = _IntermediateSection(
            header=section_header, start_line=self._line_index
        )

    def _finalise_current_section(self) -> None:
        if self._current_section is not None:
            self._document.sections.append(self._current_section.finalize())

    def _handle_header_comment(self, line: str) -> None:
        if self._current_header_block is None:
            self._current_header_block = _IntermediateCommentBlock(
                start_line=self._line_index
            )

        comment = self._convert_to_comment(line)
        self._current_header_block.add_comment_line(comment)

    def _handle_section_comment(self, line: str) -> None:
        comment = self._convert_to_comment(line)
        self._current_section.add_comment(comment, self._line_index)  # type: ignore

    def _handle_property(self, line: str) -> None:
        key, valuepart = self._retrieve_key_value(line)
        if valuepart is not None:
            comment, value = self._retrieve_property_comment(valuepart.strip())
        else:
            comment, value = None, None

        prop = Property(key=key, value=value, comment=comment)
        self._current_section.add_property(prop)  # type: ignore

    def _handle_new_datarow(self, line: str) -> None:
        self._handle_datarow(line)
        self._state = Parser._StateType.PARSING_DATABLOCK

    def _handle_datarow(self, line: str) -> None:
        self._current_section.add_datarow(line.split())  # type: ignore

    def _retrieve_property_comment(self, line: str) -> Tuple[Optional[str], str]:
        """Retrieve the comment and value part from the valuestring of a key-value pair.

        The comment retrieval is complicated by the fact that in the Deltares-INI
        dialect, the comment delimiter '#' plays a double role: it may also be used
        to quote string values (for example if the contain spaces).

        Example lines that are supported:
        key = valueAndNoComment
        key = valueA  # and a simple comment
        key = #valueA with possible spaces#
        key = #valueA#  # and a simple comment
        key = #valueA# # and a complicated comment with hashes #1 example
        key = value # and a complicated comment with hashes #2.

        Keywords arguments:
            line (str) -- the partial string of the line containing both value and
                possibly a comment at the end. Note that the "key =" part must already
                have been split off, for example by _retrieve_key_value

        Returns:
            Tuple with the comment and string value, respectively. If no comment is
            present, the first tuple element is None.
        """

        if self._config.parse_comments and self._config.comment_delimiter in line:
            line = line.strip()
            parts = line.split(self._config.comment_delimiter)
            numhash = line.count(self._config.comment_delimiter)
            if numhash == 1:
                # normal value, simple comment: "key =  somevalue # and a comment "
                comment = parts[-1]
                value = parts[0]
            elif line.startswith(self._config.comment_delimiter):
                # hashed value, possible with comment: "key = #somevalue# ..."
                comment = (
                    self._config.comment_delimiter.join(parts[3:])
                    if numhash >= 3
                    else ""
                )

                value = self._config.comment_delimiter.join(parts[0:3])
            else:
                # normal value, comment with maybe more hashes: "key = somevalue #This is comment #2, or two "
                comment = self._config.comment_delimiter.join(parts[1:])
                value = parts[0]
        else:
            comment = ""
            value = line

        return (
            comment if len(comment := comment.strip()) > 0 else None,
            value if len(value := value.strip()) > 0 else None,
        )

    def _retrieve_key_value(self, line: str) -> Tuple[str, Optional[str]]:
        if "=" in line:
            key, value = line.split("=", 1)
            return key.strip(), value if len(value := value.strip()) > 0 else None
        else:
            # if no = exists, due to the previous check we know it will just be a
            # single value
            return line, None

    def _finish_current_header_block(self) -> None:
        if self._current_header_block is not None:
            self._document.header_comment.append(self._current_header_block.finalize())
            self._current_header_block = None

    def _noop(self, *_, **__) -> None:
        # no operation
        pass

    def _is_empty_line(self, line: str) -> bool:
        return len(line.strip()) == 0

    def _is_comment(self, line: str) -> bool:
        return line.strip().startswith(self._config.comment_delimiter)

    def _convert_to_comment(self, line: str) -> str:
        return line.strip()[1:].strip()

    def _is_section_header(self, line: str) -> bool:
        # a header is defined as "[ any-value ]"
        stripped = line.strip()
        return stripped.startswith("[") and stripped.endswith("]")

    def _is_property(self, line: str) -> bool:
        # we assume that we already checked wether it is a comment or
        # a section header.
        return self._config.allow_only_keywords or "=" in line

    def _is_datarow(self, _: str) -> bool:
        # we assume that we already checked whether it is either a comment,
        # section header or a property
        return self._config.parse_datablocks

    @classmethod
    def parse_as_dict(cls, filepath: Path, config: ParserConfig = None) -> dict:
        """
        Parses an INI file without a specific model type and returns it as a dictionary.

        Args:
            filepath (Path): File path to the INI-format file.
            config (ParserConfig, optional): Parser configuration to use. Defaults to None.

        Returns:
            dict: Representation of the parsed INI-file.
        """
        return cls.parse(filepath, config).flatten()

    @classmethod
    def parse(cls, filepath: Path, config: ParserConfig = None) -> Document:
        """
        Parses an INI file without a specific model type and returns it as a Document.

        Args:
            filepath (Path): File path to the INI-format file.
            config (ParserConfig, optional): Parser configuration to use. Defaults to None.

        Returns:
            Document: Representation of the parsed INI-file.
        """
        if not config:
            config = ParserConfig()
        parser = cls(config)

        with filepath.open(encoding="utf8") as f:
            for line in f:
                parser.feed_line(line)

        return parser.finalize()

__init__(config)

Creates a new Parser configured with the provided config

Parameters:

Name	Type	Description	Default
config	ParserConfig	The configuration of this Parser	required

Source code in hydrolib/core/dflowfm/ini/parser.py

def __init__(self, config: ParserConfig) -> None:
    """Creates a new Parser configured with the provided config

    Args:
        config (ParserConfig): The configuration of this Parser
    """
    self._config = config
    self._document = Document()
    self._current_section: Optional[_IntermediateSection] = None
    self._current_header_block: Optional[_IntermediateCommentBlock] = None

    self._state = self._StateType.NO_SECTION_FOUND
    self._line_index = 0

    # TODO add invalid blocks
    self._feed_line: Dict[
        Parser._StateType, List[Tuple[Callable[[str], bool], Callable[[str], None]]]
    ] = {
        Parser._StateType.NO_SECTION_FOUND: [
            (self._is_comment, self._handle_header_comment),
            (self._is_section_header, self._handle_next_section_header),
        ],
        Parser._StateType.PARSING_PROPERTIES: [
            (self._is_comment, self._handle_section_comment),
            (self._is_section_header, self._handle_next_section_header),
            (self._is_property, self._handle_property),
            (self._is_datarow, self._handle_new_datarow),
        ],
        Parser._StateType.PARSING_DATABLOCK: [
            (self._is_section_header, self._handle_next_section_header),
            (self._is_datarow, self._handle_datarow),
        ],
    }

    self._handle_emptyline: Dict[Parser._StateType, Callable[[], None]] = {
        self._StateType.NO_SECTION_FOUND: self._finish_current_header_block,
        self._StateType.PARSING_PROPERTIES: self._noop,
        self._StateType.PARSING_DATABLOCK: self._noop,
    }

feed_line(line)

Parse the next line with this Parser.

Parameters:

Name	Type	Description	Default
line	str	The line to parse	required

Source code in hydrolib/core/dflowfm/ini/parser.py

def feed_line(self, line: str) -> None:
    """Parse the next line with this Parser.

    Args:
        line (str): The line to parse
    """
    if not self._is_empty_line(line):
        for is_line_type, handle_line_type in self._feed_line[self._state]:
            if is_line_type(line):
                handle_line_type(line)
                break
        else:
            # handle exception
            pass
    else:
        self._handle_emptyline[self._state]()

    self._increment_line()

finalize()

Finalize parsing and return the constructed Document.

Returns:

Name	Type	Description
Document	Document	A Document describing the parsed ini file.

Source code in hydrolib/core/dflowfm/ini/parser.py

def finalize(self) -> Document:
    """Finalize parsing and return the constructed Document.

    Returns:
        Document:
            A Document describing the parsed ini file.
    """
    # TODO handle invalid block
    self._finish_current_header_block()
    self._finalise_current_section()
    return self._document

parse(filepath, config=None) classmethod

Parses an INI file without a specific model type and returns it as a Document.

Parameters:

Name	Type	Description	Default
filepath	Path	File path to the INI-format file.	required
config	ParserConfig	Parser configuration to use. Defaults to None.	None

Returns:

Name	Type	Description
Document	Document	Representation of the parsed INI-file.

Source code in hydrolib/core/dflowfm/ini/parser.py

@classmethod
def parse(cls, filepath: Path, config: ParserConfig = None) -> Document:
    """
    Parses an INI file without a specific model type and returns it as a Document.

    Args:
        filepath (Path): File path to the INI-format file.
        config (ParserConfig, optional): Parser configuration to use. Defaults to None.

    Returns:
        Document: Representation of the parsed INI-file.
    """
    if not config:
        config = ParserConfig()
    parser = cls(config)

    with filepath.open(encoding="utf8") as f:
        for line in f:
            parser.feed_line(line)

    return parser.finalize()

parse_as_dict(filepath, config=None) classmethod

Parses an INI file without a specific model type and returns it as a dictionary.

Parameters:

Name	Type	Description	Default
filepath	Path	File path to the INI-format file.	required
config	ParserConfig	Parser configuration to use. Defaults to None.	None

Returns:

Name	Type	Description
dict	dict	Representation of the parsed INI-file.

Source code in hydrolib/core/dflowfm/ini/parser.py

@classmethod
def parse_as_dict(cls, filepath: Path, config: ParserConfig = None) -> dict:
    """
    Parses an INI file without a specific model type and returns it as a dictionary.

    Args:
        filepath (Path): File path to the INI-format file.
        config (ParserConfig, optional): Parser configuration to use. Defaults to None.

    Returns:
        dict: Representation of the parsed INI-file.
    """
    return cls.parse(filepath, config).flatten()

ParserConfig

Bases: BaseModel

ParserConfig defines the configuration options of the Parser

Note that we cannot set both allow_only_keywords and parse_datablocks to True because we cannot distinguish between datablocks and key only properties. As such this will lead to a validation error.

Attributes:

Name	Type	Description
allow_only_keywords	bool	Whether to allow properties with only keys (no '=' or value). Defaults to False.
parse_datablocks	bool	Whether to allow parsing of datablocks at the bottom of sections. Defaults to False.
parse_comments	bool	Whether we allow parsing of comments defined with the comment_delimeter. Defaults to True.
comment_delimiter	str	The character or sequence of character used to define a comment. Defaults to '#'.

Source code in hydrolib/core/dflowfm/ini/parser.py

class ParserConfig(BaseModel):
    """ParserConfig defines the configuration options of the Parser

    Note that we cannot set both allow_only_keywords and parse_datablocks to True
    because we cannot distinguish between datablocks and key only properties. As
    such this will lead to a validation error.

    Attributes:
        allow_only_keywords (bool):
            Whether to allow properties with only keys (no '=' or value).
            Defaults to False.
        parse_datablocks (bool):
            Whether to allow parsing of datablocks at the bottom of sections.
            Defaults to False.
        parse_comments (bool):
            Whether we allow parsing of comments defined with the comment_delimeter.
            Defaults to True.
        comment_delimiter (str):
            The character or sequence of character used to define a comment.
            Defaults to '#'.
    """

    allow_only_keywords: bool = False
    parse_datablocks: bool = False
    parse_comments: bool = True
    comment_delimiter: str = "#"

    @validator("parse_datablocks")
    def allow_only_keywods_and_parse_datablocks_leads_should_not_both_be_true(
        cls, parse_datablocks, values
    ):
        # if both allow_only_keywords and parse_datablocks is true, we cannot
        # distinguish between the two, and the parsing will not recognise either
        # properly
        if (
            parse_datablocks
            and "allow_only_keywords" in values
            and values["allow_only_keywords"]
        ):
            raise ValueError(
                "Both parse_datablocks and allow_only_keywords should not be both True."
            )
        return parse_datablocks

Serializer

DataBlockINIBasedSerializerConfig

Bases: INISerializerConfig

Class that holds the configuration settings for INI files with data blocks serialization.

Source code in hydrolib/core/dflowfm/ini/serializer.py

class DataBlockINIBasedSerializerConfig(INISerializerConfig):
    """Class that holds the configuration settings for INI files with data blocks serialization."""

    float_format_datablock: str = ""
    """str: The string format that will be used for float serialization of the datablock. If empty, the original number will be serialized. Defaults to an empty string.

    Examples:
        Input value = 123.456

        Format    | Output          | Description
        -------------------------------------------------------------------------------------------------------------------------------------
        ".0f"     | 123             | Format float with 0 decimal places.
        "f"       | 123.456000      | Format float with default (=6) decimal places.
        ".2f"     | 123.46          | Format float with 2 decimal places.
        "+.1f"    | +123.5          | Format float with 1 decimal place with a + or  sign.
        "e"       | 1.234560e+02    | Format scientific notation with the letter 'e' with default (=6) decimal places.
        "E"       | 1.234560E+02    | Format scientific notation with the letter 'E' with default (=6) decimal places.
        ".3e"     | 1.235e+02       | Format scientific notation with the letter 'e' with 3 decimal places.
        "<15"     | 123.456         | Left aligned in space with width 15
        "^15.0f"  |       123       | Center aligned in space with width 15 with 0 decimal places.
        ">15.1e"  |         1.2e+02 | Right aligned in space with width 15 with scientific notation with 1 decimal place.
        "*>15.1f" | **********123.5 | Right aligned in space with width 15 with 1 decimal place and fill empty space with *
        "%"       | 12345.600000%   | Format percentage with default (=6) decimal places.     
        ".3%"     | 12345.600%      | Format percentage with 3 decimal places.  

        More information: https://docs.python.org/3/library/string.html#format-specification-mini-language
    """

float_format_datablock = '' class-attribute instance-attribute

str: The string format that will be used for float serialization of the datablock. If empty, the original number will be serialized. Defaults to an empty string.

Examples:

Input value = 123.456

Format | Output | Description

".0f" | 123 | Format float with 0 decimal places. "f" | 123.456000 | Format float with default (=6) decimal places. ".2f" | 123.46 | Format float with 2 decimal places. "+.1f" | +123.5 | Format float with 1 decimal place with a + or sign. "e" | 1.234560e+02 | Format scientific notation with the letter 'e' with default (=6) decimal places. "E" | 1.234560E+02 | Format scientific notation with the letter 'E' with default (=6) decimal places. ".3e" | 1.235e+02 | Format scientific notation with the letter 'e' with 3 decimal places. "<15" | 123.456 | Left aligned in space with width 15 "^15.0f" | 123 | Center aligned in space with width 15 with 0 decimal places. ">15.1e" | 1.2e+02 | Right aligned in space with width 15 with scientific notation with 1 decimal place. ">15.1f" | ***123.5 | Right aligned in space with width 15 with 1 decimal place and fill empty space with * "%" | 12345.600000% | Format percentage with default (=6) decimal places.
".3%" | 12345.600% | Format percentage with 3 decimal places.

More information: https://docs.python.org/3/library/string.html#format-specification-mini-language

INISerializerConfig

Bases: SerializerConfig

SerializerConfig defines the configuration options of the Serializer

Attributes:

Name	Type	Description
section_indent	int	The number of spaces with which whole sections should be indented. Defaults to 0.
property_indent	int	The number of spaces with which properties should be indented relative to the section header (i.e. the full indent equals the section_indent plus property_indent). Defaults to 4.
datablock_indent	int	The number of spaces with which datablock rows are indented relative to the section header (i.e. the full indent equals the section_indent plus datablock_indent). Defaults to 8.
datablock_spacing	int	The number of spaces between datablock columns. Note that there might be additional offset to ensure . is lined out. Defaults to 2.
comment_delimiter	str	The character used to delimit comments. Defaults to '#'.
skip_empty_properties	bool	Whether or not to skip properties with a value that is empty or None. Defaults to True.

Source code in hydrolib/core/dflowfm/ini/serializer.py

class INISerializerConfig(SerializerConfig):
    """SerializerConfig defines the configuration options of the Serializer

    Attributes:
        section_indent (int):
            The number of spaces with which whole sections should be indented.
            Defaults to 0.
        property_indent (int):
            The number of spaces with which properties should be indented relative to
            the section header (i.e. the full indent equals the section_indent plus
            property_indent). Defaults to 4.
        datablock_indent (int):
            The number of spaces with which datablock rows are indented relative to
            the section header (i.e. the full indent equals the section_indent plus
            datablock_indent). Defaults to 8.
        datablock_spacing (int):
            The number of spaces between datablock columns. Note that there might be
            additional offset to ensure . is lined out. Defaults to 2.
        comment_delimiter (str):
            The character used to delimit comments. Defaults to '#'.
        skip_empty_properties (bool):
            Whether or not to skip properties with a value that is empty or None. Defaults to True.
    """

    section_indent: int = 0
    property_indent: int = 0
    datablock_indent: int = 8
    datablock_spacing: int = 2
    comment_delimiter: str = "#"
    skip_empty_properties: bool = True

    @property
    def total_property_indent(self) -> int:
        """The combined property indentation, i.e. section_indent + property_indent"""
        return self.section_indent + self.property_indent

    @property
    def total_datablock_indent(self) -> int:
        """The combined datablock indentation, i.e. section_indent + datablock_indent"""
        return self.section_indent + self.datablock_indent

total_datablock_indent property

The combined datablock indentation, i.e. section_indent + datablock_indent

total_property_indent property

The combined property indentation, i.e. section_indent + property_indent

MaxLengths

Bases: BaseModel

MaxLengths defines the maxmimum lengths of the parts of a section

Attributes:

Name	Type	Description
key	int	The maximum length of all the keys of the properties within a section. If no properties are present it should be 0.
value	int	The maximum length of all the non None values of the properties within a section. If no properties are present, or all values are None, it should be 0.
datablock	Optional[Sequence[int]]	The maximum length of the values of each column of the Datablock. If no datablock is present it defaults to None.

Source code in hydrolib/core/dflowfm/ini/serializer.py

class MaxLengths(BaseModel):
    """MaxLengths defines the maxmimum lengths of the parts of a section

    Attributes:
        key (int):
            The maximum length of all the keys of the properties within a section.
            If no properties are present it should be 0.
        value (int):
            The maximum length of all the non None values of the properties within a
            section. If no properties are present, or all values are None, it should
            be 0.
        datablock (Optional[Sequence[int]]):
            The maximum length of the values of each column of the Datablock.
            If no datablock is present it defaults to None.
    """

    key: int
    value: int
    datablock: Optional[Sequence[int]] = None

    @classmethod
    def from_section(cls, section: Section) -> "MaxLengths":
        """Generate a MaxLengths instance from the given Section

        Args:
            section (Section): The section of which the MaxLengths are calculated

        Returns:
            MaxLengths: The MaxLengths corresponding with the provided section
        """
        properties = list(p for p in section.content if isinstance(p, Property))

        keys = (prop.key for prop in properties)
        values = (prop.value for prop in properties if prop.value is not None)

        max_key_length = max((len(k) for k in keys), default=0)
        max_value_length = max((len(v) for v in values), default=0)
        max_datablock_lengths = MaxLengths._of_datablock(section.datablock)

        return cls(
            key=max_key_length,
            value=max_value_length,
            datablock=max_datablock_lengths,
        )

    @staticmethod
    def _of_datablock(datablock: Optional[Datablock]) -> Optional[Sequence[int]]:
        if datablock is None or len(datablock) < 1:
            return None

        datablock_columns = map(list, zip(*datablock))
        datablock_column_lengths = (map(len, column) for column in datablock_columns)  # type: ignore
        max_lengths = (max(column) for column in datablock_column_lengths)

        return tuple(max_lengths)

from_section(section) classmethod

Generate a MaxLengths instance from the given Section

Parameters:

Name	Type	Description	Default
section	Section	The section of which the MaxLengths are calculated	required

Returns:

Name	Type	Description
MaxLengths	MaxLengths	The MaxLengths corresponding with the provided section

Source code in hydrolib/core/dflowfm/ini/serializer.py

@classmethod
def from_section(cls, section: Section) -> "MaxLengths":
    """Generate a MaxLengths instance from the given Section

    Args:
        section (Section): The section of which the MaxLengths are calculated

    Returns:
        MaxLengths: The MaxLengths corresponding with the provided section
    """
    properties = list(p for p in section.content if isinstance(p, Property))

    keys = (prop.key for prop in properties)
    values = (prop.value for prop in properties if prop.value is not None)

    max_key_length = max((len(k) for k in keys), default=0)
    max_value_length = max((len(v) for v in values), default=0)
    max_datablock_lengths = MaxLengths._of_datablock(section.datablock)

    return cls(
        key=max_key_length,
        value=max_value_length,
        datablock=max_datablock_lengths,
    )

SectionSerializer

SectionSerializer provides the serialize method to serialize a Section

The entrypoint of this method is the serialize method, which will construct an actual instance and serializes the Section with it.

Source code in hydrolib/core/dflowfm/ini/serializer.py

class SectionSerializer:
    """SectionSerializer provides the serialize method to serialize a Section

    The entrypoint of this method is the serialize method, which will construct
    an actual instance and serializes the Section with it.
    """

    def __init__(self, config: INISerializerConfig, max_length: MaxLengths):
        """Create a new SectionSerializer

        Args:
            config (SerializerConfig): The config describing the serialization options
            max_length (MaxLengths): The max lengths of the section being serialized
        """
        self._config = config
        self._max_length = max_length

    @classmethod
    def serialize(cls, section: Section, config: INISerializerConfig) -> Lines:
        """Serialize the provided section with the given config

        Args:
            section (Section): The section to serialize
            config (SerializerConfig): The config describing the serialization options

        Returns:
            Lines: The iterable lines of the serialized section
        """
        serializer = cls(config, MaxLengths.from_section(section))
        return serializer._serialize_section(section)

    @property
    def config(self) -> INISerializerConfig:
        """The SerializerConfig used while serializing the section."""
        return self._config

    @property
    def max_length(self) -> MaxLengths:
        """The MaxLengths of the Section being serialized by this SectionSerializer."""
        return self._max_length

    def _serialize_section(self, section: Section) -> Lines:
        header_iterable = self._serialize_section_header(section.header)
        properties = self._serialize_content(section.content)
        datablock = self._serialize_datablock(section.datablock)

        return chain(header_iterable, properties, datablock)

    def _serialize_section_header(self, section_header: str) -> Lines:
        indent = " " * (self.config.section_indent)
        yield f"{indent}[{section_header}]"

    def _serialize_content(self, content: Iterable[ContentElement]) -> Lines:
        elements = (self._serialize_content_element(elem) for elem in content)
        return chain.from_iterable(elements)

    def _serialize_content_element(self, elem: ContentElement) -> Lines:
        if isinstance(elem, Property):
            return self._serialize_property(elem)
        else:
            indent = self.config.total_property_indent
            delimiter = self.config.comment_delimiter
            return _serialize_comment_block(elem, delimiter, indent)

    def _serialize_property(self, property: Property) -> Lines:
        if self.config.skip_empty_properties and str_is_empty_or_none(property.value):
            return

        indent = " " * (self._config.total_property_indent)
        key_ws = _get_offset_whitespace(property.key, self.max_length.key)
        key = f"{property.key}{key_ws} = "

        value_ws = _get_offset_whitespace(property.value, self.max_length.value)

        if property.value is not None:
            value = f"{property.value}{value_ws}"
        else:
            value = value_ws

        comment = (
            f" # {property.comment}"
            if not str_is_empty_or_none(property.comment)
            else ""
        )

        yield f"{indent}{key}{value}{comment}".rstrip()

    def _serialize_datablock(self, datablock: Optional[Datablock]) -> Lines:
        if datablock is None or self.max_length.datablock is None:
            return []

        indent = " " * self._config.total_datablock_indent
        return (self._serialize_row(row, indent) for row in datablock)

    def _serialize_row(self, row: DatablockRow, indent: str) -> str:
        elem_spacing = " " * self.config.datablock_spacing
        elems = (self._serialize_row_element(elem, i) for elem, i in zip(row, count()))

        return indent + elem_spacing.join(elems).rstrip()

    def _serialize_row_element(self, elem: str, index: int) -> str:
        max_length = self.max_length.datablock[index]  # type: ignore
        whitespace = _get_offset_whitespace(elem, max_length)
        return elem + whitespace

config property

The SerializerConfig used while serializing the section.

max_length property

The MaxLengths of the Section being serialized by this SectionSerializer.

__init__(config, max_length)

Create a new SectionSerializer

Parameters:

Name	Type	Description	Default
config	SerializerConfig	The config describing the serialization options	required
max_length	MaxLengths	The max lengths of the section being serialized	required

Source code in hydrolib/core/dflowfm/ini/serializer.py

def __init__(self, config: INISerializerConfig, max_length: MaxLengths):
    """Create a new SectionSerializer

    Args:
        config (SerializerConfig): The config describing the serialization options
        max_length (MaxLengths): The max lengths of the section being serialized
    """
    self._config = config
    self._max_length = max_length

serialize(section, config) classmethod

Serialize the provided section with the given config

Parameters:

Name	Type	Description	Default
section	Section	The section to serialize	required
config	SerializerConfig	The config describing the serialization options	required

Returns:

Name	Type	Description
Lines	Lines	The iterable lines of the serialized section

Source code in hydrolib/core/dflowfm/ini/serializer.py

@classmethod
def serialize(cls, section: Section, config: INISerializerConfig) -> Lines:
    """Serialize the provided section with the given config

    Args:
        section (Section): The section to serialize
        config (SerializerConfig): The config describing the serialization options

    Returns:
        Lines: The iterable lines of the serialized section
    """
    serializer = cls(config, MaxLengths.from_section(section))
    return serializer._serialize_section(section)

Serializer

Serializer serializes Document to its corresponding lines.

Source code in hydrolib/core/dflowfm/ini/serializer.py

class Serializer:
    """Serializer serializes Document to its corresponding lines."""

    def __init__(self, config: INISerializerConfig):
        """Creates a new Serializer with the provided configuration.

        Args:
            config (SerializerConfig): The configuration of this Serializer.
        """
        self._config = config

    def serialize(self, document: Document) -> Lines:
        """Serialize the provided document into an iterable of lines.

        Args:
            document (Document): The Document to serialize.

        Returns:
            Lines: An iterable returning each line of the serialized Document.
        """
        header_iterable = self._serialize_document_header(document.header_comment)

        serialize_section = lambda s: SectionSerializer.serialize(s, self._config)
        sections = (serialize_section(section) for section in document.sections)
        sections_with_spacing = Serializer._interweave(sections, [""])
        sections_iterable = chain.from_iterable(sections_with_spacing)

        return chain(header_iterable, sections_iterable)

    def _serialize_document_header(self, header: Iterable[CommentBlock]) -> Lines:
        delimiter = self._config.comment_delimiter
        serialize = lambda cb: _serialize_comment_block(cb, delimiter)
        blocks = (serialize(block) for block in header)
        blocks_with_spacing = Serializer._interweave(blocks, [""])

        return chain.from_iterable(blocks_with_spacing)

    @staticmethod
    def _interweave(iterable: Iterable, val: Any) -> Iterable:
        # Interweave the provided iterable with the provided value:
        # iterable_element, val, iterable_element, val, ...

        # Note that this will interweave with val without making copies
        # as such it is the same object being interweaved.
        return chain.from_iterable(zip(iterable, repeat(val)))

__init__(config)

Creates a new Serializer with the provided configuration.

Parameters:

Name	Type	Description	Default
config	SerializerConfig	The configuration of this Serializer.	required

Source code in hydrolib/core/dflowfm/ini/serializer.py

def __init__(self, config: INISerializerConfig):
    """Creates a new Serializer with the provided configuration.

    Args:
        config (SerializerConfig): The configuration of this Serializer.
    """
    self._config = config

serialize(document)

Serialize the provided document into an iterable of lines.

Parameters:

Name	Type	Description	Default
document	Document	The Document to serialize.	required

Returns:

Name	Type	Description
Lines	Lines	An iterable returning each line of the serialized Document.

Source code in hydrolib/core/dflowfm/ini/serializer.py

def serialize(self, document: Document) -> Lines:
    """Serialize the provided document into an iterable of lines.

    Args:
        document (Document): The Document to serialize.

    Returns:
        Lines: An iterable returning each line of the serialized Document.
    """
    header_iterable = self._serialize_document_header(document.header_comment)

    serialize_section = lambda s: SectionSerializer.serialize(s, self._config)
    sections = (serialize_section(section) for section in document.sections)
    sections_with_spacing = Serializer._interweave(sections, [""])
    sections_iterable = chain.from_iterable(sections_with_spacing)

    return chain(header_iterable, sections_iterable)

write_ini(path, document, config)

Write the provided document to the specified path

If the provided path already exists, it will be overwritten. If the parent folder do not exist, they will be created.

Parameters:

Name	Type	Description	Default
path	Path	The path to which the document should be written.	required
document	Document	The document to serialize to the specified path.	required
config	INISerializerConfig	The configuration settings for the serializer.	required

Source code in hydrolib/core/dflowfm/ini/serializer.py

def write_ini(path: Path, document: Document, config: INISerializerConfig) -> None:
    """Write the provided document to the specified path

    If the provided path already exists, it will be overwritten. If the parent folder
    do not exist, they will be created.

    Args:
        path (Path): The path to which the document should be written.
        document (Document): The document to serialize to the specified path.
        config (INISerializerConfig): The configuration settings for the serializer.
    """

    serializer = Serializer(config)

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

    with path.open("w", encoding="utf8") as f:

        for line in serializer.serialize(document):
            f.write(line + "\n")