feature: Add config parsing

Author: tgross03

pyproject.toml

@@ -4,9 +4,7 @@ description = "Simulation of radio skies to create astrophysical data sets"
 readme = "README.rst"
 requires-python = ">=3.11"
 license = { text = "MIT" }
-authors = [
-  { name = "radionets Developers" },
-]
+authors = [{ name = "radionets Developers" }]
 maintainers = [
   { name = "Kevin Schmitz", email = "kevin2.schmiz@tu-dortmund.de" },
   { name = "Christian Arauner", email = "christian.arauner@tu-dortmund.de" },
@@ -49,6 +47,7 @@ radiosim = "radiosim.tools.cli:main"
 
 [project.optional-dependencies]
 torch = ["torch", "torchvision"]
+ppdisks = ["fargopy", "tomllib", "tomli-w"]
 
 [dependency-groups]
 dev = [
@@ -94,11 +93,7 @@ requires = ["hatch-vcs", "hatchling"]
 build-backend = "hatchling.build"
 
 [tool.coverage.run]
-omit = [
-  "docs/*",
-  "src/radiosim/_version.py",
-  "src/radiosim/version.py",
-]
+omit = ["docs/*", "src/radiosim/_version.py", "src/radiosim/version.py"]
 
 [tool.coverage.xml]
 output = "coverage.xml"
@@ -125,12 +120,12 @@ extend-exclude = ["tests"]
 
 [tool.ruff.lint]
 extend-select = [
-  "I",  # isort
-  "E",  # pycodestyle
-  "F",  # Pyflakes
+  "I",   # isort
+  "E",   # pycodestyle
+  "F",   # Pyflakes
   "UP",  # pyupgrade
-  "B",  # flake8-bugbear
-  "SIM",  # flake8-simplify
+  "B",   # flake8-bugbear
+  "SIM", # flake8-simplify
 ]
 ignore = ["B905", "UP038"]
 fixable = ["ALL"]

radiosim/ppdisk/__init__.py

@@ -0,0 +1,6 @@
+from .fargopy import FargopyConfiguration
+from .parser import Parser
+from .toml import TOMLConfiguration
+from .variables import Variables
+
+__all__ = ["TOMLConfiguration", "FargopyConfiguration", "Variables", "Parser"]

radiosim/ppdisk/config/__init__.py

@@ -0,0 +1,183 @@
+import warnings
+from dataclasses import dataclass
+from pathlib import Path
+
+import numpy as np
+
+from radiosim.ppdisk.config import Parser, Variables
+
+__all__ = ["FargoParameterConfig", "FargoParameterEntry"]
+
+
+@dataclass
+class FargoParameterEntry:
+    key: str
+    value: object
+    comment: str
+
+    def get_line(self, max_key_len: int, max_value_len: int):
+        return (
+            f"{self.key:<{max_key_len + 2}}{self.value:<{max_value_len + 2}}"
+            f"{self.comment if self.comment is not None else ''}\n"
+        )
+
+
+class FargoParameterConfig:
+    def __init__(self, setup: str, autosave: bool = False):
+        self._path: Path = Variables.get("FARGO_ROOT") / f"setups/{setup}/{setup}.par"
+        self._autosave: bool = autosave
+
+        if not self._path.exists():
+            raise NameError(f"The given setup '{setup}' does not exist!")
+
+        self._parameters: dict[str, FargoParameterEntry] = dict()
+
+        self.load()
+        if self._parameters["Setup"].value != setup:
+            warnings.warn(
+                "The given setup name exists but the 'Setup' parameter in the"
+                " config gives a different name. A missmatch might lead to "
+                "execution problems.",
+                stacklevel=1,
+            )
+
+    def _get_entries(self) -> list[FargoParameterEntry]:
+        values = []
+        for _key, value in self._parameters.items():
+            if isinstance(value, dict):
+                values.extend(list(value.values()))
+            else:
+                values.append(value)
+
+        return values
+
+    def load(self) -> None:
+        with open(self._path) as file:
+            lines = file.readlines()
+
+            current_category = None
+            for line in lines:
+                if line.strip() == "":
+                    continue
+
+                if line.startswith("### ") and "[" in line and "]" in line:
+                    current_category = (
+                        line.removeprefix("### ").split("[")[1].split("]")[0]
+                    )
+                    self._parameters[current_category] = dict()
+                    continue
+
+                components = line.split()
+
+                if len(components) < 2:
+                    continue
+
+                entry = FargoParameterEntry(
+                    key=components[0],
+                    value=Parser().parse(components[1]),
+                    comment=None if len(components) == 2 else " ".join(components[2:]),
+                )
+
+                if current_category is None:
+                    self._parameters[components[0]] = entry
+                else:
+                    self._parameters[current_category][components[0]] = entry
+
+    def save(self) -> None:
+        with open(self._path) as file:
+            old_content = file.read()
+        with open(self._path, "w") as file:
+            try:
+                key_lens = []
+                value_lens = []
+                for entry in self._get_entries():
+                    key_lens.append(len(str(entry.key)))
+                    value_lens.append(len(str(entry.value)))
+
+                max_key_len = np.max(key_lens)
+                max_value_len = np.max(value_lens)
+
+                lines = []
+
+                for key, entry in self._parameters.items():
+                    if isinstance(entry, dict):
+                        lines.append("\n")
+                        lines.append(f"### [{key}]\n")
+                        lines.append("\n")
+
+                        for _subkey, subentry in self._parameters[key].items():
+                            lines.append(
+                                subentry.get_line(
+                                    max_key_len=max_key_len, max_value_len=max_value_len
+                                )
+                            )
+                    else:
+                        lines.append(
+                            entry.get_line(
+                                max_key_len=max_key_len, max_value_len=max_value_len
+                            )
+                        )
+
+                file.writelines(lines)
+            except Exception as e:
+                warnings.warn(
+                    "An error occured while saving. Rolling back configuration files.",
+                    stacklevel=1,
+                )
+                file.write(old_content)
+                raise e
+
+    def __getitem__(self, key: str) -> FargoParameterEntry:
+        key_components = key.split(".")
+
+        match len(key_components):
+            case 1:
+                return self._parameters[key_components[0]]
+            case 2:
+                return self._parameters[key_components[0]][key_components[1]]
+            case _:
+                if len(key_components) > 2:
+                    raise KeyError(
+                        "The maximum depth of a config key is 2 (catgeory -> entry)!"
+                    )
+
+    def __setitem__(self, key: str, value: object) -> None:
+        key_components = key.split(".")
+
+        match len(key_components):
+            case 1:
+                if isinstance(value, dict):
+                    self._parameters[key_components[0]] = value
+                    return None
+                if isinstance(value, FargoParameterEntry):
+                    self._parameters[key_components[0]] = value
+                elif isinstance(
+                    self._parameters[key_components[0]], FargoParameterEntry
+                ):
+                    self._parameters[key_components[0]].value = value
+                else:
+                    raise TypeError(
+                        "Values at root level must either be a dict or a valid entry!"
+                    )
+            case 2:
+                if isinstance(value, FargoParameterEntry):
+                    self._parameters[key_components[0]][key_components[1]] = value
+                elif isinstance(
+                    self._parameters[key_components[0]][key_components[1]],
+                    FargoParameterEntry,
+                ):
+                    self._parameters[key_components[0]][key_components[1]].value = value
+                else:
+                    raise TypeError(
+                        "This key does not point to a valid entry! Enter an instance "
+                        "of a 'FargoParameterEntry'"
+                    )
+
+            case _:
+                if len(key_components) > 2:
+                    raise KeyError(
+                        "The maximum depth of a config key is 2 (catgeory -> entry)!"
+                    )
+
+        if self._autosave:
+            self.save()

radiosim/ppdisk/config/fargo.py

@@ -0,0 +1,66 @@
+import shutil
+import subprocess
+from pathlib import Path
+
+
+def _parse_fpc_value(value: object) -> str:
+    match value:
+        case str():
+            return f'"{value}"'
+        case float() | int() | bool():
+            return value
+        case _:
+            raise TypeError(
+                f"The value '{value}' is not parsable for the fargopy config!"
+            )
+
+
+class FargopyConfiguration:
+    def __init__(self):
+        self.path: Path = Path("~/.fargopy/fargopyrc").expanduser()
+
+        if not self.exists():
+            self.reset()
+
+    def exists(self) -> bool:
+        return self.path.is_file()
+
+    def get_content(self) -> dict:
+        fargopyrc = dict()
+        with open(self.path) as file:
+            exec(file.read(), dict(), fargopyrc)
+        return fargopyrc
+
+    def reset(self) -> None:
+        try:
+            import fargopy as fp
+
+            fp.initialize(options="configure")
+        except Exception:
+            print(
+                "The fargopy configuration seems to be corrupted or is gone. "
+                "Re-running initial fargopy configuration. This could take a moment..."
+            )
+
+            if self.exists() or self.path.parent.exists():
+                shutil.rmtree(self.path.parent)
+
+            subprocess.run(["ifargopy"], shell=True)
+
+            print("Finished regeneration.")
+
+        self["FP_FARGO3D_CLONECMD"] = "git clone https://github.com/FARGO3D/fargo3d.git"
+
+    def __getitem__(self, i: str):
+        return self.get_content()[i]
+
+    def __setitem__(self, key: str, value: object):
+        _ = self[key]
+        with open(self.path) as f:
+            content = f.readlines()
+
+        with open(self.path, "w") as f:
+            for i in range(len(content)):
+                if content[i].startswith(key):
+                    content[i] = f"{key} = {_parse_fpc_value(value)}\n"
+            f.writelines(content)

radiosim/ppdisk/config/fargopy.py

@@ -0,0 +1,47 @@
+__all__ = ["Parser"]
+
+
+class Parser:
+    def __init__(self, bool_true: str = "yes", bool_false: str = "no"):
+        self._type_parsers = [
+            BoolParser(true_val=bool_true, false_val=bool_false),
+            TypeParser(parse_type=int),
+            TypeParser(parse_type=float),
+            TypeParser(parse_type=str),
+        ]
+
+    def parse(self, value: str):
+        for parser in self._type_parsers:
+            try:
+                return parser.parse(value=value)
+            except Exception:
+                continue
+
+
+class BoolParser:
+    def __init__(self, true_val: str, false_val: str):
+        self._true_val: str = true_val
+        self._false_val: str = false_val
+
+    def parse(self, value: str) -> bool:
+        true_check = value == self._true_val
+        false_check = value == self._false_val
+
+        if true_check:
+            return True
+        if false_check:
+            return False
+
+        if not true_check and not false_check:
+            raise ValueError(
+                f"Invalid boolean value: '{value}'! "
+                f"Valid values: {self._true_val} (True) or {self._false_val} (False)"
+            )
+
+
+class TypeParser:
+    def __init__(self, parse_type: type):
+        self._parse_type = parse_type
+
+    def parse(self, value: str):
+        return self._parse_type(value)