Mondaic

salvus.opt.iteration

salvus.opt.iteration salvus opt iteration

The iteration class and related functionality.

Functions

require_gradient()

def require_gradient(
    iteration: Iteration[ModelType],
) -> HasGradient[ModelType]: ...

Type guard to ensure that the iteration has a gradient.

Parameters
  • iteration Iteration[ModelType] — The iteration to check.
Returns HasGradient[ModelType] — The iteration with a guaranteed gradient.

require_parent_id()

def require_parent_id(it: Iteration[ModelType]) -> HasParentID: ...

Type guard to ensure that the iteration has a parent ID.

Parameters
  • it Iteration[ModelType] — The iteration to check.
Returns HasParentID — The iteration with a guaranteed parent ID.

Classes

HasGradient

class HasGradient(typing.Protocol):
    def __init__(self):
        ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...

HasParentID

class HasParentID(typing.Protocol):
    def __init__(self):
        ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...

Iteration

class Iteration(
    salvus.flow.utils.serialization_helpers.SerializationMixin, typing.Generic
):
    def __init__(
        self,
        id: int,
        model: ModelType,
        events: list[str],
        misfit_configuration: str,
        parent_id: int | None = None,
        control_group_events: list[str] | None = None,
    ): ...

Iteration class to organize all tasks and data linked to one iteration of an multi-event inversion.

Parameters
  • id int — Unique identifier of the iteration.
  • model ModelType — Model of the iteration.,
  • events list[str] — Batch of events for this iteration.
  • misfit_configuration str — Misfit configuration.
  • parent_id int | None — ID of the parent iteration that this iteration is derived from.
  • control_group_events list[str] | None — Optional subset of events to be used for checking the trial model. Must be a subset of the events of the iteration. If not given, all events will be used.
Attributes
objective float

Obtain the objective value of this iteration, i.e., the sum of misfit and regularization term.

Returns: The aggregated objective value.

Methods
from_disk()
def from_disk(d: dict, path: pathlib.Path) -> Iteration[ModelType]: ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d dict — JSON dictionary of the iteration.
  • path pathlib.Path — Path to the folder containing auxiliary files.
Returns Iteration[ModelType]
from_json()
def from_json(d: dict) -> Iteration[ModelType]: ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d dict — JSON dictionary of the iteration.
Returns Iteration[ModelType]
from_parent()
def from_parent(
    id: int,
    parent: Iteration[ModelType],
    events: list[str] | None = None,
    control_group_events: list[str] | None = None,
    misfit_configuration: str | None = None,
) -> Iteration[ModelType]: ...

Derive a new iteration from a parent iteration. The model of the new iteration will be set to the final model of the parent iteration. If events and/or misfit_configuration are not specified, they will be inherited from the parent iteration.

Parameters
  • id int — Unique identifier of the new iteration.
  • parent Iteration[ModelType] — Parent iteration to inherit from.
  • events list[str] | None — List of iteration-specific events.
  • control_group_events list[str] | None — Iteration-specific control group.
  • misfit_configuration str | None — Iteration-specfic misfit configuration.
Returns Iteration[ModelType]
add_trial_model()
def add_trial_model(
    self,
    model: ModelType,
    d: dict | None = None,
    status: ModelStatus = "unknown",
) -> None: ...

Add a trial model to the iteration. This could be a proposed model update or any kind of auxiliary model that needs to be stored.

Parameters
  • model ModelType — Trial model to be stored.
  • d dict | None — Auxiliary information for the trial model.
  • status ModelStatus — Status of the model.
Returns None
get_final_model()
def get_final_model(self) -> str: ...

Obtain the name of the final model of this iteration.

Returns str — Name of the final model.
get_gradient()
def get_gradient(self, event: str, name: str | None = None) -> ModelType: ...

Obtain the mapped gradient of an event.

Parameters
  • event str — Event name.
  • name str | None — Optional name of the returned gradient.
Returns ModelType
get_model_directory()
def get_model_directory(self) -> pathlib.Path: ...

Obtain the path where all models and gradients of this iteration are stored.

Returns pathlib.Path
get_new_trial_model_id()
def get_new_trial_model_id(self) -> int: ...

Obtain the next available ID for a trial model.

Returns int — Trial model ID.
get_trial_models()
def get_trial_models(self, status: str = "all") -> list[TrialModelEntry]: ...

Obtain trial models associated with this iteration.

Parameters
  • status str — Return only models with a specific status, or all models if status="all".
Returns list[TrialModelEntry]
issue_task()
def issue_task(self, tasks: Task | typing.Sequence[Task]) -> None: ...

Issue a new set of tasks for this iteration.

Parameters
  • tasks Task | typing.Sequence[Task] — Task or list of tasks that need to be done next.
Returns None
process_current_tasks()
def process_current_tasks(self) -> None: ...

Process the outputs of all current tasks. The function assumes that all current tasks have been completed and the outputs of the tasks have been updated. This function will link the task outputs to the other data stored in the iteration object.

Returns None
set_entry_point()
def set_entry_point(self, next_task: str) -> None: ...

Modify the next entry point where this iteration will continue after all tasks have been completed.

Parameters
  • next_task str — Identifier of the next task.
Returns None
set_model_directory()
def set_model_directory(self, path: pathlib.Path) -> None: ...

Set the path to the model inventory where all models and gradients of this iteration are stored.

Parameters
  • path pathlib.Path — Path to model inventory.
Returns None
to_disk()
def to_disk(
    self, path: pathlib.Path, external_file_hash: str | None = None
) -> dict: ...

Serialize the object to dictionary that can be written to JSON. Auxiliary files will be written to the specified path.

Parameters
  • path pathlib.Path — Path to store auxiliary files.
  • external_file_hash str | None — Hash of any external files associated with this object. Can be passed here in which case it will be stored in a centralized location in the JSON file.
Returns dict
to_json()
def to_json(
    self,
    external_file_hash: str | None = None,
    timer: Timer | None = None,
    log_to_logger: bool = False,
    comm: int | None = None,
) -> dict: ...

Serialize the object to dictionary that can be written to JSON.

Parameters
  • external_file_hash str | None — Hash of any external files associated with this object. Can be passed here in which case it will be stored in a centralized location in the JSON file.
  • timer Timer | None — Execution timer.
  • log_to_logger bool — Log timings to the logger.
  • comm int | None — MPI communicator, if any.
Returns dict

TrialModelEntry

class TrialModelEntry(typing.Generic, builtins.dict):
    def __init__(self): ...

Entry describing a trial/auxiliary model.

This structure stores metadata and (potentially serialized) gradient information for auxiliary / trial models produced during an iteration.