Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(ini): allow custom parsers/serializers for ini options #2332

Merged
merged 10 commits into from
Feb 10, 2025
70 changes: 62 additions & 8 deletions antarest/study/storage/rawstudy/ini_reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,19 +15,30 @@
import typing as t
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Callable, Optional

from typing_extensions import override

from antarest.core.model import JSON

PrimitiveType = str | int | float | bool
ValueParser = Callable[[str], PrimitiveType]

def convert_value(value: str) -> t.Union[str, int, float, bool]:

def _lower_case(input: str) -> str:
return input.lower()


LOWER_CASE_PARSER: ValueParser = _lower_case


def _convert_value(value: str) -> PrimitiveType:
"""Convert value to the appropriate type for JSON."""

try:
# Infinity values are not supported by JSON, so we use a string instead.
mapping = {"true": True, "false": False, "+inf": "+Inf", "-inf": "-Inf", "inf": "+Inf"}
return t.cast(t.Union[str, int, float, bool], mapping[value.lower()])
return t.cast(PrimitiveType, mapping[value.lower()])
except KeyError:
try:
return int(value)
Expand All @@ -38,7 +49,42 @@ def convert_value(value: str) -> t.Union[str, int, float, bool]:
return value


@dataclasses.dataclass
@dataclasses.dataclass(frozen=True)
class OptionMatcher:
"""
Used to match a location in an INI file:
a None section means any section.
"""

section: Optional[str]
key: str


def any_section_option_matcher(key: str) -> OptionMatcher:
"""
Return a matcher which will match the provided key in any section.
"""
return OptionMatcher(section=None, key=key)


class ValueParsers:
def __init__(self, parsers: t.Dict[OptionMatcher, ValueParser]):
self._parsers = parsers

def find_parser(self, section: str, key: str) -> ValueParser:
if not self._parsers:
sylvlecl marked this conversation as resolved.
Show resolved Hide resolved
return _convert_value
sylvlecl marked this conversation as resolved.
Show resolved Hide resolved
possible_keys = [
OptionMatcher(section=section, key=key),
OptionMatcher(section=None, key=key),
sylvlecl marked this conversation as resolved.
Show resolved Hide resolved
]
for k in possible_keys:
if parser := self._parsers.get(k, None):
return parser
return _convert_value


@dataclasses.dataclass(frozen=True)
class IniFilter:
"""
Filter sections and options in an INI file based on regular expressions.
Expand Down Expand Up @@ -115,8 +161,8 @@ def read(self, path: t.Any, **kwargs: t.Any) -> JSON:
Parse `.ini` file to json object.

Args:
path: Path to `.ini` file or file-like object.
kwargs: Additional options used for reading.
path: Path to `.ini` file or file-like object.
options: Additional options used for reading.

Returns:
Dictionary of parsed `.ini` file which can be converted to JSON.
Expand Down Expand Up @@ -152,11 +198,17 @@ class IniReader(IReader):
This class is not compatible with standard `.ini` readers.
"""

def __init__(self, special_keys: t.Sequence[str] = (), section_name: str = "settings") -> None:
def __init__(
self,
special_keys: t.Sequence[str] = (),
section_name: str = "settings",
value_parsers: t.Dict[OptionMatcher, ValueParser] | None = None,
) -> None:
super().__init__()

# Default section name to use if `.ini` file has no section.
self._special_keys = set(special_keys)
self._value_parsers = ValueParsers(value_parsers or {})

# List of keys which should be parsed as list.
self._section_name = section_name
Expand Down Expand Up @@ -313,10 +365,12 @@ def _handle_option(self, ini_filter: IniFilter, section: str, key: str, value: s
def _append_option(self, section: str, key: str, value: str) -> None:
self._curr_sections.setdefault(section, {})
values = self._curr_sections[section]
parser = self._value_parsers.find_parser(section, key)
parsed = parser(value)
if key in self._special_keys:
values.setdefault(key, []).append(convert_value(value))
values.setdefault(key, []).append(parsed)
else:
values[key] = convert_value(value)
values[key] = parsed
self._curr_option = key


Expand Down
49 changes: 45 additions & 4 deletions antarest/study/storage/rawstudy/ini_writer.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,18 +12,51 @@

import ast
import configparser
import typing as t
from pathlib import Path
from typing import Callable, Dict, List, Optional

from typing_extensions import override

from antarest.core.model import JSON
from antarest.study.storage.rawstudy.ini_reader import OptionMatcher, PrimitiveType
sylvlecl marked this conversation as resolved.
Show resolved Hide resolved

# Value serializers may be used to customize the way INI options are serialized
ValueSerializer = Callable[[str], PrimitiveType]


def _lower_case(input: str) -> str:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This also exists inside the ini_reader, we could import it or put this inside the utility file

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I slightly prefer to keep it separate because it has 2 different usages (parsing and serializing)

return input.lower()


LOWER_CASE_SERIALIZER: ValueSerializer = _lower_case


class ValueSerializers:
def __init__(self, serializers: Dict[OptionMatcher, ValueSerializer]):
self._serializers = serializers

def find_serializer(self, section: str, key: str) -> Optional[ValueSerializer]:
if not self._serializers:
return None
possible_keys = [
OptionMatcher(section=section, key=key),
OptionMatcher(section=None, key=key),
]
for k in possible_keys:
if parser := self._serializers.get(k, None):
return parser
return None


class IniConfigParser(configparser.RawConfigParser):
def __init__(self, special_keys: t.Optional[t.List[str]] = None) -> None:
def __init__(
self,
special_keys: Optional[List[str]] = None,
value_serializers: Optional[ValueSerializers] = None,
) -> None:
super().__init__()
self.special_keys = special_keys
self._value_serializers = value_serializers or ValueSerializers({})

# noinspection SpellCheckingInspection
@override
Expand All @@ -41,6 +74,9 @@ def _write_line( # type:ignore
value = self._interpolation.before_write( # type:ignore
self, section_name, key, value
)
if self._value_serializers:
if serializer := self._value_serializers.find_serializer(section_name, key):
value = serializer(value)
if value is not None or not self._allow_no_value: # type:ignore
value = delimiter + str(value).replace("\n", "\n\t")
else:
Expand Down Expand Up @@ -70,8 +106,13 @@ class IniWriter:
Standard INI writer.
"""

def __init__(self, special_keys: t.Optional[t.List[str]] = None):
def __init__(
self,
special_keys: Optional[List[str]] = None,
value_serializers: Optional[Dict[OptionMatcher, ValueSerializer]] = None,
):
self.special_keys = special_keys
self._value_serializers = ValueSerializers(value_serializers or {})

def write(self, data: JSON, path: Path) -> None:
"""
Expand All @@ -81,7 +122,7 @@ def write(self, data: JSON, path: Path) -> None:
data: JSON content.
path: path to `.ini` file.
"""
config_parser = IniConfigParser(special_keys=self.special_keys)
config_parser = IniConfigParser(special_keys=self.special_keys, value_serializers=self._value_serializers)
config_parser.read_dict(data)
with path.open("w") as fp:
config_parser.write(fp)
Expand Down
32 changes: 31 additions & 1 deletion tests/storage/repository/antares_io/reader/test_ini_reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,13 @@
import textwrap
from pathlib import Path

from antarest.study.storage.rawstudy.ini_reader import IniReader, SimpleKeyValueReader
from antarest.study.storage.rawstudy.ini_reader import (
LOWER_CASE_PARSER,
IniReader,
OptionMatcher,
SimpleKeyValueReader,
any_section_option_matcher,
)


class TestIniReader:
Expand Down Expand Up @@ -324,6 +330,30 @@ def test_read__filtered_option(self, tmp_path) -> None:
expected = {"part1": {"bar": "hello"}, "part2": {"bar": "salut"}}
assert actual == expected

def test_read__with_custom_parser(self, tmp_path):
path = Path(tmp_path) / "test.ini"
path.write_text(
textwrap.dedent(
"""
[part1]
bar = Hello

[part2]
bar = Hello
"""
)
)

value_parsers = {OptionMatcher("part2", "bar"): LOWER_CASE_PARSER}
sylvlecl marked this conversation as resolved.
Show resolved Hide resolved
actual = IniReader(value_parsers=value_parsers).read(path)
expected = {"part1": {"bar": "Hello"}, "part2": {"bar": "hello"}}
assert actual == expected

value_parsers = {any_section_option_matcher("bar"): LOWER_CASE_PARSER}
actual = IniReader(value_parsers=value_parsers).read(path)
expected = {"part1": {"bar": "hello"}, "part2": {"bar": "hello"}}
assert actual == expected


class TestSimpleKeyValueReader:
def test_read(self) -> None:
Expand Down
31 changes: 30 additions & 1 deletion tests/storage/repository/antares_io/writer/test_ini_writer.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@

import pytest

from antarest.study.storage.rawstudy.ini_writer import IniWriter
from antarest.study.storage.rawstudy.ini_reader import OptionMatcher, any_section_option_matcher
sylvlecl marked this conversation as resolved.
Show resolved Hide resolved
from antarest.study.storage.rawstudy.ini_writer import LOWER_CASE_SERIALIZER, IniWriter


@pytest.mark.unit_test
Expand Down Expand Up @@ -59,3 +60,31 @@ def test_write(tmp_path: str, ini_cleaner: Callable) -> None:
writer.write(json_data, path)

assert ini_cleaner(ini_content) == ini_cleaner(path.read_text())


@pytest.mark.unit_test
def test_write_with_custom_serializer(tmp_path: str, ini_cleaner: Callable) -> None:
path = Path(tmp_path) / "test.ini"

serializers = {any_section_option_matcher("group"): LOWER_CASE_SERIALIZER}
writer = IniWriter(value_serializers=serializers)

expected = """
[part1]
group = gas

[part2]
group = gas

[part3]
other = Gas
"""

json_data = {
"part1": {"group": "Gas"},
"part2": {"group": "Gas"},
"part3": {"other": "Gas"},
}
writer.write(json_data, path)

assert ini_cleaner(path.read_text()) == ini_cleaner(expected)
Loading