Source code for

# Copyright (c) 2017-2024 Fumito Hamamura <>

# This library is free software: you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation version 3.
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# Lesser General Public License for more details.
# You should have received a copy of the GNU Lesser General Public
# License along with this library.  If not, see <>.

import pathlib
from types import MappingProxyType

def is_in_root(path: pathlib.Path):
    """Returns False if the relative path with dots go beyond root"""
    pos = path.as_posix()
    split = pos.split("/")
    level = 0
    while split:
        item = split.pop(0)
        if item == "..":
            level -= 1
        elif item == ".":
            level += 1
        if level < 0:
            return False
    return True

class BaseSharedIO:

    def __init__(self, path: pathlib.Path, manager, load_from):
        self._path = path
        self._specs = {}     # {id(spec): spec}
        self._manager = manager
        self._load_from = load_from

    def path(self):
        return self._path

    def path(self, path):
        self._manager.update_path(self, path)

    def specs(self):
        return MappingProxyType(self._specs)

    def load_from(self):
        return self._load_from

    def is_external(self):
        return self.path.is_absolute()

    def _on_write(self, path):
        raise NotImplementedError

    def _on_update_path(self, path: pathlib.Path):
        self._path = path   # override as necessary

    def _on_update_value(self, value, kwargs):
        raise NotImplementedError

    def _can_add_spec(self, spec):
        return all(c._can_add_other(spec) for c in self._specs.values())

    def _can_update_spec(self, spec, kwargs):
        return all(c._can_update_other(spec, **kwargs) for c in self._specs.values())

    def _check_sanity(self):

        key, val = next(
            (k, v) for k, v in self._manager.ios.items() if v is self)

        assert self.path == key[1]
        assert self is val

        for spec in self._specs.values():
            assert is self

    def __getstate__(self):
        return {
            "path": pathlib.PurePath(self.path),
            "specs": list(self._specs.values()),
            "manager": self._manager,
            "load_from": self._load_from

    def __setstate__(self, state):
        self._path = pathlib.Path(state["path"])
        self._manager = state["manager"]
        self._load_from = state["load_from"]
        self._specs = {id(c): c for c in state["specs"]}

    def persistent_args(self):
        return {}

class BiDict(dict):
    """Bidirectional Dictionary

    Some methods on dict, such as update(), does not work.
    def __init__(self, *args, **kwargs):
        super(BiDict, self).__init__(*args, **kwargs)
        self.inverse = {}
        for key, value in self.items():
            if value in self.inverse:
                raise ValueError("not injective mapping")
                self.inverse[value] = key

    def __setitem__(self, key, value):
        if key in self:
            if value == self[key]:
            elif value not in self.inverse:
                del self.inverse[self[key]]
                raise ValueError("not injective key-value")

        super(BiDict, self).__setitem__(key, value)
        self.inverse[value] = key

    def __delitem__(self, key):
        del self.inverse[self[key]]
        super(BiDict, self).__delitem__(key)

class IOManager:
    """A class to manage shared file

    * Create new spec
        - Create a new spec
        - Register a new spec
            - Create a SharedData if not exit
            - Register the spec to the file

    def __init__(self):
        self.ios = BiDict({})
        self.serializing = None     # Set from external

    def _check_sanity(self):
        assert len(self.ios) == len(set(id(v) for v in self.ios.values()))
        for a_io in self.ios.values():

    def _get_io(self, io_group, path: pathlib.Path):
        return self.ios.get(self._get_io_key(io_group, path), None)

    def get_ios(self, io_group):
        return {path: ios for (group, path), ios in self.ios.items()
                if group == io_group}

    def _get_io_key(self, io_group, path):
        if path.is_absolute():
            return None, path
            return io_group, path

    def _new_io(self, io_group, path, cls, load_from, **kwargs):
        a_io = cls(path, self, load_from, **kwargs)
        if not self._get_io(io_group, a_io.path):
            self.ios[self._get_io_key(io_group, a_io.path)] = a_io
            return a_io
            raise RuntimeError("must not happen")

    def _del_io(self, io_):
        if io_.specs:
            raise RuntimeError("specs must be deleted beforehand")

        key = next((k for k, v in self.ios.items() if v is io_), None)
        if key:
            del self.ios[key]

    def restore_io(self, io_group, io_):
        # Used only by restore_state in ModelImpl
        # To add unpickled IO in self.ios
        res = self._get_io(io_group, io_.path)
        if not res:
            self.ios[self._get_io_key(io_group, io_.path)] = io_

    def get_or_create_io(self, io_group, path, cls, load_from=None, **kwargs):
        a_io = self._get_io(io_group, path)
        if a_io:
            return a_io
            return self._new_io(io_group, path, cls, load_from, **kwargs)

    def new_spec(
            self, cls, io_group, path, spec_args=None, io_args=None):

        if spec_args is None:
            spec_args = {}
        if io_args is None:
            io_args = {}

        spec = cls(**spec_args)
        spec._manager = self
        spec._io = self.get_or_create_io(
            io_group, pathlib.Path(path), cls=cls.io_class, **io_args)
            self.add_spec(, spec)
            if not
                del self.ios[self._get_io_key(io_group,]

        return spec

    def add_spec(self, io_, spec):
        if id(spec) not in io_.specs:
            if io_._can_add_spec(spec):
                io_._specs[id(spec)] = spec
                raise ValueError("cannot add spec")

    def del_spec(self, spec):
        a_io =
        if id(spec) in a_io.specs:
            del a_io._specs[id(spec)]
            if not a_io.specs:

    def get_spec_from_value(self, io_group, value):
        ios = self.get_ios(io_group)
        return next(
            (spec for io_ in ios.values() for spec in io_.specs.values()
         if spec.value is value), None)

    def update_spec_value(self, spec, value, kwargs):
        if spec._can_update_value(value, kwargs):
  , kwargs)
            spec._on_update_value(value, kwargs)
            raise ValueError(
                "%s does not allow to replace its value" % repr(spec)

    def update_spec(self, spec, **kwargs):
        if, kwargs):
            raise ValueError("cannot change spec")

    def update_path(self, io_, path):
        path = pathlib.Path(path)
        group, path_old = key_old = self.ios.inverse[io_]
        if path == path_old:
            key = self._get_io_key(group, path)
            if key in self.ios:
                raise ValueError("cannot change path")
                del self.ios[key_old]
                self.ios[key] = io_

    def write_ios(self, io_group, root):
        for (group, path), a_io in self.ios.items():
            if not group or group == io_group:  # group is None if absolute path
                self.write_io(a_io, root)

    def write_io(self, io_, root):
        if not io_.path.is_absolute():
            path = root.joinpath(io_.path)
            path = io_.path

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

[docs] class BaseIOSpec: """Abstract base class for storing objects in files. .. currentmodule:: This base class is inherited from by *IOSpec* classes, such as :class:`~pandasio.PandasData` and :class:`~moduleio.ModuleData`, and defines properties shared among the child classes. The :meth:`Model.get_spec<modelx.core.model.Model.get_spec>` method returns the *IOSpec* object associated with a given object referenced in the model. :attr:`Model.iospecs<modelx.core.model.Model.iospecs>` returns a list of all the *IOSpec* associated with the objects referenced in the model. Example: The code below returns the :class:`` object associated with a DataFrame ``df`` referenced in the model:: >>> model.get_spec(df) <PandasData path='df.xlsx' file_type='excel' sheet='df1'> The code below returns a list of all the IOSpec objects associated with the objects referenced in the model:: >>> model.iospecs [<PandasData path='df.xlsx' file_type='excel' sheet='df1'>, <PandasData path='df.xlsx' file_type='excel' sheet='s1'>, <ModuleData path=''>] .. versionchanged:: 0.20.0 Renamed from ``BaseDataSpec`` to :class:`~baseio.BaseIOSpec` .. versionchanged:: 0.18.0 The ``is_hidden`` parameter is removed. .. versionchanged:: 0.18.0 Renamed from ``BaseDataClient`` to ``BaseDataSpec``. See Also: * :class:`` * :class:`` * :class:`` * :attr:`~modelx.core.model.Model.iospecs` * :meth:`~modelx.core.model.Model.get_spec` * :meth:`~modelx.core.model.Model.write` """ def __init__(self): self._manager = None self._io = None @property def io(self): """The :class:`~BaseSharedIO` object that this object is associated to""" return self._io @property def path(self): """File path that the object is written to. This property is defined in :class:`BaseIOSpec`, and returns the path to a file to which the object is written to when the model is saved by the :meth:`~modelx.core.model.Model.write` method. The returned path is an instance of the Path class defined in `pathlib`_ module in the Python standard library. The `Path`_ class is either `WindowsPath`_ or `PosixPath`_, depending on the platform the Python session is running on. This property also works as a setter to set the path. :obj:`str` or any path-like object can be given. Example: In the code below, ``df`` is a DataFrame referenced in ``model``, and the IPython sessin is running on Windows. The DataFrame is saved in an Excel file named *df.xlsx* in the model folder:: >>> model.get_spec(df).path WindowsPath('df.xlsx') The next assignment changes the file path to *files/df2.xlsx* under the model folder:: >>> model.get_spec(df).path = 'files/df2.xlsx' >>> model.get_spec(df).path WindowsPath('files/df.xlsx') See Also: * :meth:`~modelx.core.model.Model.write` * :func:`~modelx.write_model` .. _pathlib: .. _Path: .. _WindowsPath: .. _PosixPath: """ return self._io.path @path.setter def path(self, path): self._io.path = path def _check_sanity(self): assert self._io.specs[id(self)] is self assert any(self._io is d for d in self._manager.ios.values()) assert self._manager is self._io._manager def _on_load_value(self): raise NotImplementedError def _on_update_path(self, path): raise NotImplementedError def _can_update_value(self, value, kwargs): raise NotImplementedError def _on_update_value(self, value, kwargs): raise NotImplementedError def _on_update(self, kwargs): raise NotImplementedError def _on_pickle(self, state): raise NotImplementedError def _on_unpickle(self, state): raise NotImplementedError def _on_serialize(self, state): raise NotImplementedError def _on_unserialize(self, state): raise NotImplementedError def _can_add_other(self, other): raise NotImplementedError def __hash__(self): return hash(id(self)) def __getstate__(self): state = { "manager": self._manager, "_io": self._io } if hasattr(self, "_is_hidden"): state["is_hidden"] = self._is_hidden if self._manager.serializing is True: return self._on_serialize(state) elif self._manager.serializing is False: return self._on_pickle(state) else: raise RuntimeError("must not happen") def __setstate__(self, state): self._manager = state["manager"] self._io = state["_io"] if "_io" in state else state["_data"] # renamed from v0.20.0 if "is_hidden" in state: self._is_hidden = state["is_hidden"] else: # For backward compatibility with v0.12 self._is_hidden = False if self._manager.serializing is True: self._on_unserialize(state) self._manager.add_spec(self._io, self) elif self._manager.serializing is False: self._on_unpickle(state) else: RuntimeError("must not happen") def _get_attrdict(self, extattrs=None, recursive=True): result = { "type": type(self).__name__, "path": str(self._io.path), "load_from": str(self._io.load_from), "value": self.value } return result