Source code for eko.io.metadata
"""Define `eko.EKO` metadata."""
import logging
import os
import pathlib
from dataclasses import dataclass
from typing import Optional
import yaml
from packaging.version import parse
from .. import version as vmod
from ..interpolation import XGrid
from . import v1, v2
from .dictlike import DictLike
from .paths import InternalPaths
from .types import EvolutionPoint as EPoint
logger = logging.getLogger(__name__)
[docs]
@dataclass
class Metadata(DictLike):
"""Manage metadata, and keep them synced on disk.
It is possible to have a metadata view, in which the path is not actually
connected (i.e. it is set to ``None``). In this case, no update will be
possible, of course.
Note
----
Unfortunately, for nested structures it is not possible to detect a change
in their attributes, so a call to :meth:`update` has to be performed
manually.
"""
origin: EPoint
"""Inital scale."""
xgrid: XGrid
"""Interpolation grid."""
# tagging information
_path: Optional[pathlib.Path] = None
"""Path to the open dir."""
version: str = vmod.__version__
"""Library version used to create the corresponding file."""
data_version: int = vmod.__data_version__
"""Specs version, to which the file adheres."""
[docs]
@classmethod
def load(cls, path: os.PathLike):
"""Load metadata from open folder.
Parameters
----------
path: os.PathLike
the path to the open EKO folder
Returns
-------
Metadata
loaded metadata
"""
path = pathlib.Path(path)
paths = InternalPaths(path)
# read raw file first to catch version
raw = yaml.safe_load(paths.metadata.read_text(encoding="utf-8"))
version = parse(raw["version"])
# patch if necessary
if version.major == 0 and version.minor == 13:
raw = v1.update_metadata(paths, raw)
elif version.major == 0 and version.minor == 14:
raw = v2.update_metadata(paths, raw)
# now we are ready
content = cls.from_dict(raw)
content._path = path
return content
[docs]
def update(self):
"""Update the disk copy of metadata."""
if self._path is None:
logger.info("Impossible to set metadata, no file attached.")
else:
with open(InternalPaths(self._path).metadata, "w", encoding="utf8") as fd:
yaml.safe_dump(self.raw, fd)
@property
def path(self):
"""Access temporary dir path.
Raises
------
RuntimeError
if path has not been initialized before
"""
if self._path is None:
raise RuntimeError(
"Access to EKO directory attempted, but not dir has been set."
)
return self._path
@path.setter
def path(self, value: pathlib.Path):
"""Set temporary dir path."""
self._path = value
@property
def raw(self):
"""Override default :meth:`DictLike.raw` representation to exclude
path."""
return self.public_raw