Persistence

deap_er.Checkpoint(file_name=None, dir_path=None, autoload=True, make_dir=True, raise_errors=False, hof_ind_cls=None)

Save and load evolution progress with dill.

Only attributes set on the checkpoint instance are written. The library RNG state (buffered uniforms, leftover integers, and Generator) is persisted as well. The target file is chosen at construction.

Parameters:

Name	Type	Description	Default
file_name	str | None	Checkpoint file name. Defaults to a random UUID with a .dcpf extension.	None
dir_path	Path | None	Directory for the file. Defaults to <cwd>/deap-er.	None
autoload	bool	If True and the file exists, load it during initialization. A missing file is not an error, even with raise_errors. Defaults to True.	True
make_dir	bool	If True, create missing parent directories on save. Defaults to True.	True
raise_errors	bool	If True, propagate I/O and pickle errors. Otherwise load and save return False. Defaults to False.	False
hof_ind_cls	type[Any] | None	Creator individual type used to rebuild hof from JSON on load. When set, hof is stored as JSON instead of dill. Otherwise hof is pickled like any other attribute.	None

See the class docstring for argument meanings.

Source code in deap_er/private/checkpoint.py

def __init__(
    self,
    file_name: str | None = None,
    dir_path: Path | None = None,
    autoload: bool = True,
    make_dir: bool = True,
    raise_errors: bool = False,
    hof_ind_cls: type[Any] | None = None,
) -> None:
    """See the class docstring for argument meanings."""
    if file_name is None:
        file_name = str(uuid.uuid4()) + self._ext_
    if dir_path is None:
        dir_path = Path(os.getcwd()).resolve()
        dir_path = dir_path.joinpath(self._dir_)
    self.file_path = dir_path.joinpath(file_name)
    self.raise_errors = raise_errors
    self.make_dir = make_dir
    self._hof_ind_cls_ = hof_ind_cls
    if autoload is True and self.file_path.is_file():
        self.load()

save_freq property writable

Seconds between automatic saves during range().

-1 disables saving. The default is 60 seconds.

Returns:

Type	Description
float	The current period in seconds.

last_op property

Status of the last load or save.

One of none, load_success, load_error, save_success, or save_error.

__getattr__(name)

Allow dynamically assigned checkpoint attributes.

Source code in deap_er/private/checkpoint.py

def __getattr__(self, name: str) -> Any:
    """Allow dynamically assigned checkpoint attributes."""
    raise AttributeError(name)

__setattr__(name, value)

Set a checkpoint attribute, including user-defined names.

Source code in deap_er/private/checkpoint.py

@override
def __setattr__(self, name: str, value: Any) -> None:
    """Set a checkpoint attribute, including user-defined names."""
    object.__setattr__(self, name, value)

load()

Load attributes from the checkpoint file onto this instance.

The file is read and validated before anything changes. A failed load leaves the instance and the library RNG untouched.

Returns:

Type	Description
bool	True on success, False on failure when raise_errors is False.

Raises:

Type	Description
OSError	If the file cannot be read and raise_errors is True.
CheckpointError	If the file is not a loadable checkpoint and raise_errors is True.

Source code in deap_er/private/checkpoint.py

def load(self) -> bool:
    """Load attributes from the checkpoint file onto this instance.

    The file is read and validated before anything changes. A failed
    load leaves the instance and the library RNG untouched.

    Returns:
        True on success, False on failure when ``raise_errors`` is False.

    Raises:
        OSError: If the file cannot be read and ``raise_errors`` is True.
        CheckpointError: If the file is not a loadable checkpoint and
            ``raise_errors`` is True.
    """
    file_path = self.file_path
    raise_errors = self.raise_errors
    make_dir = self.make_dir
    hof_ind_cls = self._hof_ind_cls_
    try:
        state = read_checkpoint(file_path, hof_ind_cls)
    except (OSError, CheckpointError):
        if raise_errors:
            raise
        self._last_op_ = "load_error"
        return False
    self.__dict__ = state
    self.file_path = file_path
    self.raise_errors = raise_errors
    self.make_dir = make_dir
    if hof_ind_cls is not None or "_hof_ind_cls_" not in state:
        self._hof_ind_cls_ = hof_ind_cls
    if self._rng_state_ is not None:
        rng.set_state(self._rng_state_)
    self._last_op_ = "load_success"
    return True

save()

Write this instance's attributes to the checkpoint file.

Overwrites an existing file through a uniquely named staging file that is fsynced before the rename, so concurrent writers and power loss never leave a torn checkpoint. Creates parent directories when make_dir is True.

Returns:

Type	Description
bool	True on success, False on failure when raise_errors is False.

Raises:

Type	Description
OSError	If the file cannot be written and raise_errors is True.
Exception	Whatever serialization raises (for example dill.PickleError, ValueError, or RecursionError) when raise_errors is True.

Source code in deap_er/private/checkpoint.py

def save(self) -> bool:
    """Write this instance's attributes to the checkpoint file.

    Overwrites an existing file through a uniquely named staging
    file that is fsynced before the rename, so concurrent writers
    and power loss never leave a torn checkpoint. Creates parent
    directories when ``make_dir`` is True.

    Returns:
        True on success, False on failure when ``raise_errors`` is False.

    Raises:
        OSError: If the file cannot be written and ``raise_errors`` is True.
        Exception: Whatever serialization raises (for example
            ``dill.PickleError``, ``ValueError``, or ``RecursionError``)
            when ``raise_errors`` is True.
    """
    try:
        self._rng_state_ = rng.get_state()
        if self.make_dir:
            self.file_path.parent.mkdir(parents=True, exist_ok=True)
        _dict_ = vars(self).copy()
        for key in self._omit_:
            _dict_.pop(key, None)
        hof = _dict_.get("hof")
        if self._hof_ind_cls_ is not None and isinstance(hof, HallOfFame):
            _dict_["_hof_json_"] = hof.to_json()
            del _dict_["hof"]
        write_atomic(self.file_path, serialize_state(_dict_))
    except Exception:
        if self.raise_errors:
            raise
        self._last_op_ = "save_error"
        return False
    self._last_op_ = "save_success"
    return True

range(generations)

Yield generation indices, saving periodically when enabled.

Continues from the last stored counter. When save_freq is not -1, the checkpoint is written about every save_freq seconds and once more after the last yield. save_freq may be changed while iterating.

Parameters:

Name	Type	Description	Default
generations	int	Number of generations to yield.	required

Returns:

Type	Description
Iterator[int]	A generator of integer generation indices.

Raises:

Type	Description
ValueError	If generations is negative.

Source code in deap_er/private/checkpoint.py

def range(self, generations: int) -> Iterator[int]:
    """Yield generation indices, saving periodically when enabled.

    Continues from the last stored counter. When ``save_freq`` is
    not ``-1``, the checkpoint is written about every
    ``save_freq`` seconds and once more after the last yield.
    ``save_freq`` may be changed while iterating.

    Args:
        generations: Number of generations to yield.

    Returns:
        A generator of integer generation indices.

    Raises:
        ValueError: If ``generations`` is negative.
    """
    if generations < 0:
        raise ValueError("Iterations argument cannot be a negative number.")
    from_ = self._range_counter_ + 1
    to_excl = self._range_counter_ + generations + 1
    last_save = time.time()
    for i in range(from_, to_excl):
        yield i
        self._range_counter_ = i
        if self.save_freq == -1:
            continue
        now = time.time()
        if now - last_save >= self._save_freq_:
            last_save = now
            self.save()
    if self.save_freq != -1:
        self.save()

is_loaded()

Return whether the last operation was a successful load.

Returns:

Type	Description
bool	True if last_op is load_success.

Source code in deap_er/private/checkpoint.py

def is_loaded(self) -> bool:
    """Return whether the last operation was a successful load.

    Returns:
        True if ``last_op`` is ``load_success``.
    """
    return self._last_op_ == "load_success"

is_saved()

Return whether the last operation was a successful save.

Returns:

Type	Description
bool	True if last_op is save_success.

Source code in deap_er/private/checkpoint.py

def is_saved(self) -> bool:
    """Return whether the last operation was a successful save.

    Returns:
        True if ``last_op`` is ``save_success``.
    """
    return self._last_op_ == "save_success"