Mondaic
This API reference is not for the latest stable Salvus version.

salvus.opt.tasks

Collection of task objects to handle asynchronous inversions.

Functions

read_task_from_json()

def read_task_from_json(d: Dict) -> salvus.opt.tasks.Task:
    ...

Utility to deserialize a task. This requires to detect the correct class name from the JSON dictionary to instantiate the task.

Parameters
  • d Dict — JSON dictionary of the task.
Returns salvus.opt.tasks.Task — Task object.

validate_keys()

def validate_keys(
    required: builtins.set,
    obtained: builtins.set,
    key: str,
    partial_update: bool = False,
) -> None:
    ...

Utility function to validate a set of keys.

Parameters
  • required builtins.set — Set of required items.
  • obtained builtins.set — Set of obtained items.
  • key str — Description of the set (just for printing nicer errors).
  • partial_update bool — Allow partial update that does not necessarily contain all required keys.
Returns None

Classes

Task

class Task(salvus.flow.utils.serialization_helpers.SerializationMixin):
    def __init__(self):
        ...

Base class to create tasks during an inversion.

Attributes
name str

Task name

Methods
from_json()
def from_json(d: Dict) -> typing_extensions.Self:
    ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d Dict — JSON dictionary of the task.
Returns typing_extensions.Self
to_json()
def to_json(self, external_file_hash: Optional[str] = None) -> Dict:
    ...

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

Parameters
  • external_file_hash Optional[str] — 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
update_status()
def update_status(self, status: str) -> None:
    ...

Update task status

Parameters
  • status str — New status.
Returns None
validate()
def validate(self, success: bool = False) -> None:
    ...

Validate status of the tasks

Parameters
  • success bool — Raise if task has not been completed (status="done").
Returns None

TaskCustom

class TaskCustom(salvus.opt.tasks.Task):
    def __init__(
        self,
        info: Dict[
            str,
            Union[
                int,
                str,
                float,
                salvus.flow.utils.serialization_helpers.SerializationMixin,
            ],
        ],
    ):
        ...

Custom task. All required information needs to be passed as dictionary that can be serialized to JSON.

Parameters
  • info Dict[str, Union[int, str, float, salvus.flow.utils.serialization_helpers.SerializationMixin]] — Custom task specification.
Attributes
name str

Task name

Methods
from_json()
def from_json(d: Dict) -> TaskCustom:
    ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d Dict — JSON dictionary of the task.
Returns TaskCustom
to_json()
def to_json(self, external_file_hash: Optional[str] = None) -> Dict:
    ...

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

Parameters
  • external_file_hash Optional[str] — 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
update_results()
def update_results(
    self,
    result: Dict[
        str,
        Union[
            int,
            str,
            float,
            salvus.flow.utils.serialization_helpers.SerializationMixin,
        ],
    ],
) -> None:
    ...

Update the results of the tasks.

Parameters
  • result Dict[str, Union[int, str, float, salvus.flow.utils.serialization_helpers.SerializationMixin]] — Task results.
Returns None
update_status()
def update_status(self, status: str) -> None:
    ...

Update task status

Parameters
  • status str — New status.
Returns None
validate()
def validate(self, success: bool = False) -> None:
    ...

Validate status of the tasks

Parameters
  • success bool — Raise if task has not been completed (status="done").
Returns None

TaskEventSelection

class TaskEventSelection(salvus.opt.tasks.Task):
    def __init__(
        self, events: Sequence[str], required_events: Sequence[str] = []
    ):
        ...

Task to select a subgroup of events as control group

Parameters
  • events Sequence[str] — List of events to choose from.
  • required_events Sequence[str] — List of required events.
Attributes
name str

Task name

Methods
from_json()
def from_json(d: Dict) -> TaskEventSelection:
    ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d Dict — JSON dictionary of the task.
Returns TaskEventSelection
to_json()
def to_json(self, external_file_hash: Optional[str] = None) -> Dict:
    ...

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

Parameters
  • external_file_hash Optional[str] — 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
update_results()
def update_results(self, selection: List[bool]) -> None:
    ...

Update the results of the tasks.

Parameters
  • selection List[bool] — Boolean list selecting / deselecting events.
Returns None
update_status()
def update_status(self, status: str) -> None:
    ...

Update task status

Parameters
  • status str — New status.
Returns None
validate()
def validate(self, success: bool = False) -> None:
    ...

Validate status of the tasks

Parameters
  • success bool — Raise if task has not been completed (status="done").
Returns None

TaskFinalizeIteration

class TaskFinalizeIteration(salvus.opt.tasks.Task):
    def __init__(
        self,
        info: Dict[
            str,
            Union[
                int,
                str,
                float,
                salvus.flow.utils.serialization_helpers.SerializationMixin,
            ],
        ] = {},
    ):
        ...

Custom task to finalize iteration.

Parameters
  • info Dict[str, Union[int, str, float, salvus.flow.utils.serialization_helpers.SerializationMixin]] — Custom task specification.
Attributes
name str

Task name

Methods
from_json()
def from_json(d: Dict) -> TaskFinalizeIteration:
    ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d Dict — JSON dictionary of the task.
Returns TaskFinalizeIteration
to_json()
def to_json(self, external_file_hash: Optional[str] = None) -> Dict:
    ...

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

Parameters
  • external_file_hash Optional[str] — 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
update_results()
def update_results(self) -> None:
    ...

Update the status of the tasks.

Returns None
update_status()
def update_status(self, status: str) -> None:
    ...

Update task status

Parameters
  • status str — New status.
Returns None
validate()
def validate(self, success: bool = False) -> None:
    ...

Validate status of the tasks

Parameters
  • success bool — Raise if task has not been completed (status="done").
Returns None

TaskGradients

class TaskGradients(salvus.opt.tasks.Task):
    def __init__(self, model: str, events: Sequence[str]):
        ...

Task to compute gradients for a given model and list of events.

Parameters
  • model str — Model
  • events Sequence[str] — List of events.
Attributes
name str

Task name

Methods
from_json()
def from_json(d: Dict) -> TaskGradients:
    ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d Dict — JSON dictionary of the task.
Returns TaskGradients
to_json()
def to_json(self, external_file_hash: Optional[str] = None) -> Dict:
    ...

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

Parameters
  • external_file_hash Optional[str] — 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
update_results()
def update_results(
    self, gradients: Dict[str, Any], partial_update: bool = False
) -> None:
    ...

Update the results of the tasks.

Parameters
  • gradients Dict[str, Any] — Dictionary with event-specific gradients.
  • partial_update bool — Allow partial updates for a subset of data.
Returns None
update_status()
def update_status(self, status: str) -> None:
    ...

Update task status

Parameters
  • status str — New status.
Returns None
validate()
def validate(self, success: bool = False) -> None:
    ...

Validate status of the tasks

Parameters
  • success bool — Raise if task has not been completed (status="done").
Returns None

TaskMisfits

class TaskMisfits(salvus.opt.tasks.Task):
    def __init__(self, model: str, events: Sequence[str]):
        ...

Task to compute misfits for a given model and list of events.

Parameters
  • model str — Model
  • events Sequence[str] — List of events.
Attributes
name str

Task name

Methods
from_json()
def from_json(d: Dict) -> TaskMisfits:
    ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d Dict — JSON dictionary of the task.
Returns TaskMisfits
to_json()
def to_json(self, external_file_hash: Optional[str] = None) -> Dict:
    ...

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

Parameters
  • external_file_hash Optional[str] — 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
update_results()
def update_results(
    self, misfits: Dict[str, float], partial_update: bool = False
) -> None:
    ...

Update the results of the tasks.

Parameters
  • misfits Dict[str, float] — Dictionary with event-specific misfits.
  • partial_update bool — Allow partial updates for a subset of data.
Returns None
update_status()
def update_status(self, status: str) -> None:
    ...

Update task status

Parameters
  • status str — New status.
Returns None
validate()
def validate(self, success: bool = False) -> None:
    ...

Validate status of the tasks

Parameters
  • success bool — Raise if task has not been completed (status="done").
Returns None

TaskMisfitsAndGradients

class TaskMisfitsAndGradients(salvus.opt.tasks.Task):
    def __init__(self, model: str, events: Sequence[str]):
        ...

Task to compute misfits and gradients for a given model and list of events.

Parameters
  • model str — Model
  • events Sequence[str] — List of events.
Attributes
name str

Task name

Methods
from_json()
def from_json(d: Dict) -> TaskMisfitsAndGradients:
    ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d Dict — JSON dictionary of the task.
Returns TaskMisfitsAndGradients
to_json()
def to_json(self, external_file_hash: Optional[str] = None) -> Dict:
    ...

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

Parameters
  • external_file_hash Optional[str] — 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
update_results()
def update_results(
    self, misfits: Dict[str, float], gradients: Dict[str, Any]
) -> None:
    ...

Update the results of the tasks.

Parameters
  • misfits Dict[str, float] — Dictionary with event-specific misfits.
  • gradients Dict[str, Any] — Dictionary with event-specific gradients.
Returns None
update_status()
def update_status(self, status: str) -> None:
    ...

Update task status

Parameters
  • status str — New status.
Returns None
validate()
def validate(self, success: bool = False) -> None:
    ...

Validate status of the tasks

Parameters
  • success bool — Raise if task has not been completed (status="done").
Returns None

TaskMisfitsAndSummedGradient

class TaskMisfitsAndSummedGradient(salvus.opt.tasks.Task):
    def __init__(self, model: str, events: Sequence[str]):
        ...

Task to compute misfits and gradients for a given model and list of events and return the summed gradient.

Parameters
  • model str — Model
  • events Sequence[str] — List of events.
Attributes
name str

Task name

Methods
from_json()
def from_json(d: Dict) -> TaskMisfitsAndSummedGradient:
    ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d Dict — JSON dictionary of the task.
Returns TaskMisfitsAndSummedGradient
to_json()
def to_json(self, external_file_hash: Optional[str] = None) -> Dict:
    ...

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

Parameters
  • external_file_hash Optional[str] — 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
update_results()
def update_results(
    self, misfits: Dict[str, float], gradient: Dict[str, Any]
) -> None:
    ...

Update the results of the tasks.

Parameters
  • misfits Dict[str, float] — Dictionary with event-specific misfits.
  • gradient Dict[str, Any] — Summed gradient.
Returns None
update_status()
def update_status(self, status: str) -> None:
    ...

Update task status

Parameters
  • status str — New status.
Returns None
validate()
def validate(self, success: bool = False) -> None:
    ...

Validate status of the tasks

Parameters
  • success bool — Raise if task has not been completed (status="done").
Returns None

TaskPreconditioner

class TaskPreconditioner(salvus.opt.tasks.Task):
    def __init__(
        self, model: Union[str, salvus.opt.models.base_model.BaseModel]
    ):
        ...

Task to apply a preconditioner to a given model.

Parameters
  • model Union[str, salvus.opt.models.base_model.BaseModel] — Model to apply the preconditioner to.
Attributes
name str

Task name

Methods
from_json()
def from_json(d: Dict) -> TaskPreconditioner:
    ...

Construct object from a JSON-serializable dictionary.

Parameters
  • d Dict — JSON dictionary of the task.
Returns TaskPreconditioner
to_json()
def to_json(self, external_file_hash: Optional[str] = None) -> Dict:
    ...

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

Parameters
  • external_file_hash Optional[str] — 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
update_results()
def update_results(
    self,
    preconditioned_model: Optional[
        str, salvus.opt.models.base_model.BaseModel
    ] = None,
    info: Optional[Dict[str, str]] = None,
) -> None:
    ...

Update the results of the tasks.

Parameters
  • preconditioned_model Optional[str, salvus.opt.models.base_model.BaseModel] — Preconditioned model.
  • info Optional[Dict[str, str]] — Auxiliary information, e.g., the job name of a running simulation.
Returns None
update_status()
def update_status(self, status: str) -> None:
    ...

Update task status

Parameters
  • status str — New status.
Returns None
validate()
def validate(self, success: bool = False) -> None:
    ...

Validate status of the tasks

Parameters
  • success bool — Raise if task has not been completed (status="done").
Returns None