simstack.core package

Subpackages

Submodules

simstack.core.artifacts module

class simstack.core.artifacts.ArtifactArguments(result: Any, task_id: ObjectId | None = None)[source]

Bases: object

add_attributes(func: Callable[[...], Any], *args: Any, **kwargs: Any) → None[source]
async simstack.core.artifacts.create_artifacts(artifact_arguments: ArtifactArguments, node_registry: NodeRegistry) → List[ObjectId][source]
async simstack.core.artifacts.find_all_artifacts(node_registry: NodeRegistry, db: Database) → List[ArtifactModel][source]
async simstack.core.artifacts.find_artifact_mappings(node_registry_path: str, db: Database, task_id: str | None = None) → List[ArtifactMapping][source]
async simstack.core.artifacts.find_artifacts(model_class: type[ArtifactModel], **kwargs: Any) → List[ArtifactModel][source]

Custom find function for polymorphic models

async simstack.core.artifacts.register_artifact_mapping(artifact_mapping: ArtifactMapping) → ArtifactMapping[source]
async simstack.core.artifacts.save_artifact_model(artifact: ArtifactModel) → ArtifactModel[source]

Custom save function for polymorphic models

If the item is a List type, it will also save any unsaved nested items.

simstack.core.asnyc_helper module

simstack.core.asnyc_helper.async_helper(func)[source]

simstack.core.config_file module

class simstack.core.config_file.SimstackConfig[source]

Bases: object

Singleton class that manages the path to a configuration file.

This class ensures only one instance exists throughout the application and provides a centralized way to access the configuration file path.

property config_file_path: str | None

Get the current configuration file path.

simstack.core.config_file.get_config_file() → SimstackConfig[source]

Get the singleton ConfigManager instance.

Returns:

The ConfigManager singleton instance

simstack.core.context module

simstack.core.definitions module

class simstack.core.definitions.DBType(*values)[source]

Bases: str, Enum

IN_MEMORY = 'in_memory'
MONGODB = 'mongodb'
POSTGRES = 'postgres'
SQLITE = 'sqlite'
WITH_PATH = 'with_path'
class simstack.core.definitions.TaskStatus(*values)[source]

Bases: str, Enum

COMPLETED = 'completed'
FAILED = 'failed'
RECOVERED = 'recovered'
RETRIEVED = 'retrieved'
RUNNING = 'running'
SLURM_QUEUED = 'slurm_queued'
SLURM_RUNNING = 'slurm_running'
SUBMITTED = 'submitted'
TIME_OUT = 'timeout'

simstack.core.engine module

class simstack.core.engine.AIOEngineProxy(client: motor.motor_asyncio.AsyncIOMotorClient | None = None, database: str = 'test')[source]

Bases: AIOEngine

A proxy engine that inherits all behavior from AIOEngine but overrides save.

The overridden save will: - If the model has a save member function, call it with the engine instance. - Else, if any direct part/attribute of the model has a save member function, call those with the engine instance. - Otherwise, fall back to the original AIOEngine.save implementation.

Notes: - Supports single model instances or iterables of model instances (list/tuple/set). - Avoids recursion by delegating to super().save for the fallback.

async save(obj: Any, *args, **kwargs) → Any[source]

Persist an instance to the database

This method behaves as an ‘upsert’ operation. If a document already exists with the same primary key, it will be overwritten.

All the other models referenced by this instance will be saved as well.

Parameters:

  • instance – instance to persist

  • session – An optional session to use for the operation. If not provided, an internal session will be used to persist the instance and sub-instances.

Returns:

the saved instance

Raises:

DuplicateKeyError – the instance is duplicated according to a unique index.

Note

The save operation actually modify the instance argument in place. However, the instance is still returned for convenience.

<!— #noqa: DAR401 TypeError #noqa: DAR402 DuplicateKeyError –>

async save_unchecked(obj: Any, *args, **kwargs) → Any[source]

Save without checking for custom save methods. This is a direct call to the original AIOEngine.save.

simstack.core.find_simstack_modules module

simstack.core.find_simstack_modules.find_simstack_modules()[source]

Find all packages and modules within the simstack package.

simstack.core.find_simstack_modules.walk_packages(package_name, all_modules)[source]

Walk through all packages and modules.

simstack.core.hash module

class simstack.core.hash.ComplexHash(obj: Any)[source]

Bases: object

complex_hash(obj: Any) → str | int[source]
hash_class(cls_obj: Any) → str[source]
hash_dict(obj: Mapping[Any, Any]) → str[source]
simstack.core.hash.complex_hash_function(obj: Any) → str | int[source]
simstack.core.hash.hash_class_def(cls: type[Any]) → str[source]
simstack.core.hash.hash_function_body(func: Callable[[...], Any]) → str[source]
simstack.core.hash.hash_iterable(iterable: Iterable[Any]) → str[source]
simstack.core.hash.hash_non_callable_members(instance: Any) → dict[str, str][source]
simstack.core.hash.hash_value(value: Any) → str[source]
simstack.core.hash.is_iterable(obj: Any) → bool[source]
simstack.core.hash.is_primitive_type(obj: Any) → bool[source]

simstack.core.node module

class simstack.core.node.Node(*args: Any, **kwargs: Any)[source]

Bases: object

represents a computational task or node in the workflow, managing its lifecycle, execution environment, and interactions with the database. This class helps in defining tasks, storing their details, and ensuring they are executed either locally or remotely, with their statuses tracked within a shared database context.

Variables:

  • name – The name of the function associated with the node.

  • custom_name – A custom name for the node, generated if not specified.

  • registry_entry – The registry entry associated with the node in the database.

  • parent_id – The parent node’s unique identifier, if applicable.

  • parameters – Additional parameters for the node.

  • _func – The function represented by the node.

async execute_node_locally() → Model | SimstackResult | None[source]

Executes a specified node in the current context locally, either asynchronously or synchronously, managing task status updates, directory changes, and result persistence.

This method handles the execution of a computational task represented as a “node”. It manages the task’s status transitions, file system operations for managing working directories, and handling output results, including their persistence in a database. The method supports both asynchronous and synchronous node execution. It verifies results, handles exceptions, and manages task metadata updates.

Nodes can either return

  • a single Model instance

  • a SimstackResult instance

  • None (for failure).

  • a boolean value (for failure or success if there are no results)

There is a try-except block around the actual execution of the node which generates a log entry “node function error for node” that catches all uncaught exceptions within the node. These error are not propagated, but the task status is set to TaskStatus.FAILED.

Parameters:

self – Instance of the class invoking this method.

Raises:

Exception – for failures of the Simstack logic

Returns:

The processed result of the node execution. Depending on the task’s output, it could be of the type Model, SimstackResult, or be None if no valid result was produced.

async get_node_registry() → TaskStatus[source]

Reads or initializes the task registry entry in the database.

This method ensures that a task entry exists in the database for the current task. It computes hashes of its arguments and function, checks if a database entry already matches these hashes, and creates a new entry if no match is found. If the database is not connected, an exception is raised.

Raises:

ValueError – if the database is not connected.

Returns:

Status of the task retrieved or created.

Return type:

TaskStatus

property id: ObjectId | None
async load_results() → Model | SimstackResult | None[source]

Loads the results associated with a specific task from the database. This method verifies whether the task has valid result identifiers. If valid identifiers (results_id and results_table_name) exist, it attempts to fetch the outputs.

If the task status is not TaskStatus.COMPLETED None is returned which results in a RuntimeError If the results are not found or if the retrieval process fails, a ValueError is raised.

Raises:

ValueError – If the task has completed but lacks output identifiers (results_id or results_table_name), or if there is any error during the process of loading the results.

Returns:

The retrieved task outputs from the database.

async make_registry_entry(function_hash: str, arg_hash: str) → NodeRegistry[source]

Creates a registry entry for the node in the database.

This method is used to create a new entry in the database for the node, including its inputs and outputs. It ensures that the task is properly registered with all necessary details.

Return type:

NodeRegistry

async process_results(result: Any) → tuple[TaskStatus, Any][source]
async run_somewhere() → Model | SimstackResult | None[source]

Executes the task either locally or on a remote resource. This function ensures that if the task is meant to execute on a remote resource, it waits for the task to complete remotely and fetches its results. If the task executes locally, it directly runs the task and retrieves the results.

If any exception occurs during the execution, the status is updated to FAILED and the exception is logged.

Returns:

A single Model or a list of Model instances of the task results or None. If the task is not completed successfully.

Return type:

Model | SimstackResult

Raises:

RunTimeError – When task execution fails due to an unexpected exception.

async set_status(status: TaskStatus) → None[source]
property status: TaskStatus
simstack.core.node.compute_arg_hash(args: List[Model]) → str[source]

Computes a hash for a list of arguments provided, where each argument is an instance of the Model class or can be processed into a hashable format. Uses a complex hashing function for the resulting computation.

Parameters:

args (List[Model]) – A list of objects where each object must be an instance of the Model class. The objects are used to compute their respective hash values via a specified complex hashing mechanism.

Returns:

A string representation of the computed hash for the provided

list of arguments.

Return type:

str

Raises:

TypeError – If any item in the provided list is not an instance of the Model class.

simstack.core.node.default_name_generator() → str[source]
simstack.core.node.hashable_inputs(arg: Any) → dict[str, Any][source]

Get the hashable inputs for the node. This allows exclusion of some fields from the hash.

Returns:

The hashable inputs.

Return type:

dict

simstack.core.node.hashable_value(value: Any) → Any[source]
simstack.core.node.node(_func: Callable[[P], T]) → Callable[[...], T][source]
simstack.core.node.node(_func: None = None, *, name: str | None = None, version: str | None = None, cache: bool = True, **kwargs_node: Any) → Callable[[Callable[[P], T]], Callable[[...], T]]

Decorator to mark a function as a node in the computation graph. Supports both synchronous and asynchronous functions. Can be used with or without parameters: @node def func(): …

@node(name=”example”) def func(): …

async simstack.core.node.node_from_database(registry_entry: NodeRegistry) → Node | None[source]

Constructs an instance of the class from database information encoded in a registry entry.

This method retrieves input arguments and the serialized function from the database using information provided in the registry_entry. It then deserializes the function and initializes a corresponding Node instance, associating it with the given registry entry.

This function can delete the registry_entry !!! The only way that registry_entry.function_hash is “NOT INITIALIZED” is when the node is created from the frontend. No other node is listening specifically for this registry_entry to complete. If a duplicate is found the node from the duplication is returned

Parameters:

registry_entry (NodeRegistry) – The registry entry containing information necessary to reconstruct the Node instance. Includes input table names, function pickled as a string, and other metadata.

Returns:

A reconstructed Node instance based on the registry entry, or None if the deserialized function is not valid or there was an error.

Return type:

Optional[Node]

simstack.core.node_claim module

async simstack.core.node_claim.claim_submitted_node(registry_entry: NodeRegistry) → bool[source]

Automatically claim a submitted node for execution or submission.

simstack.core.node_runner module

class simstack.core.node_runner.NodeRunner(name: str, task_id: str | ObjectId, logger: Logger = None, *, status: TaskStatus = TaskStatus.RETRIEVED, error_message: str | None = None, message: str | None = None, files: List[FileStack] = <factory>, info_files: List[FileStack] = <factory>, **kwargs)[source]

Bases: SimstackResult

A task runner class that extends SimstackResult for executing subprocess commands and managing task execution.

This class provides functionality for running shell commands, collecting output files, and managing task status with comprehensive logging capabilities. It acts as a bridge between the Simstack node execution environment and external processes, ensuring that output, errors, and informational files are correctly captured and associated with the task.

task_id

Unique identifier for the task, typically a string representation of an ObjectId.

Type:

str

name

Name of the task runner instance, usually derived from the node function name.

Type:

str

logger

Logger instance for recording task activities.

Type:

logging.Logger

last_stdout

Most recent stdout from subprocess execution.

Type:

str

last_stderr

Most recent stderr from subprocess execution.

Type:

str

log_string

Cumulative log of messages recorded via the log method.

Type:

str

info_file_patterns

Set of file patterns (e.g., “*.log”) used to collect information files.

Type:

Set[str]

debug(msg)[source]

Log a debug message with task context.

Parameters:

msg (str) – Debug message to log

error(msg)[source]

Log an error message with task context and exception info.

Parameters:

msg (str) – Error message to log.

fail(msg: str) → NodeRunner[source]

Mark the task as failed and record an error message.

This method updates the task status to FAILED, sets the error message, logs the error, and ensures the cumulative log file is created.

Parameters:

msg (str) – Error message describing the failure.

Returns:

Self reference for chaining.

Return type:

NodeRunner

classmethod from_kwargs(**kwargs) → NodeRunner[source]

Create a NodeRunner instance from keyword arguments.

If ‘node_runner’ is already present in kwargs and is an instance of NodeRunner, it returns it. Otherwise, it initializes a new NodeRunner using ‘name’, ‘task_id’, and optionally ‘logger’ from kwargs.

Parameters:

**kwargs – Keyword arguments containing ‘node_runner’ or initialization parameters.

Returns:

An existing or newly created NodeRunner instance.

Return type:

NodeRunner

info(msg)[source]

Log an info message with task context.

Parameters:

msg (str) – Info message to log.

log(msg: str)[source]

Log a message and append it to the cumulative log string.

This method records a message in the internal log_string and also logs it as a debug message using the logger. The cumulative log string can later be saved as a log file when the task finishes.

Parameters:

msg (str) – Message to log.

async make_info_files(*args, cwd: str | Path = '')[source]

Collect and process information files based on patterns and explicit file paths.

This method processes the provided arguments to identify file patterns and explicit file paths, then collects all matching files into FileStack objects for later use.

Parameters:

  • *args – Variable arguments that can be: - File patterns (strings containing ‘*’) - Explicit file paths (existing readable files)

  • cwd (str | Path, optional) – The directory where to look for files. Defaults to current directory.

Note

Files are added to the info_files list as FileStack objects. Only readable files are processed.

model_config = {'extra': 'allow'}

Represents the results from a Simstack operation.

This class serves to encapsulate the status, messages, and files resulting from a Simstack operation. It provides structured attributes for the error message, operation message, and categorized files, ensuring clarity in the representation of these aspects. It also supports additional fields through flexible configuration.

Variables:

  • status – Indicates the status of the task, defaulting to TaskStatus.COMPLETED.

  • error_message – Optionally stores an error message if the task encountered issues.

  • message – An optional general message providing additional information about the task.

  • files – A list of FileStack objects that represent the primary files involved in the operation. Defaults to an empty list.

  • info_files – A list of FileStack objects that provide additional or informational files pertaining to the operation. Defaults to an empty list. These files are shown in the results but not passed to the calling function

submit_to_watchdog(name: str, command: str) → bool[source]

Submit a command to an external watchdog process for execution.

This method prepares the necessary files and environment to submit a command to a watchdog process that will handle its execution. It uses a specified queue directory to manage job files and signals.

Parameters:

  • name (str) – Name identifier for the watchdog job (used for log file naming).

  • command (str) – Command to be executed by the watchdog.

Returns:

True if the submission and execution were successful, False otherwise.

Return type:

bool

Note

  • The method sets the task status to FAILED if the submission or execution fails.

  • Job logs are automatically added to the info_files collection.

subprocess(name: str, command: str | List[str], cwd: str = '') → bool[source]

Execute a shell command as a subprocess and capture its output.

This method runs a shell command, captures stdout and stderr, creates a log file with the execution details, and adds the log file to the info_files collection.

Parameters:

  • name (str) – Name identifier for the subprocess (used for log file naming)

  • command (str) – Shell command to execute

  • cwd (str, optional) – Working directory for command execution. Defaults to current directory.

Returns:

True if the subprocess completed successfully (return code 0), False otherwise

Return type:

bool

Note

  • Creates a log file named “{name}.log” containing command details and output

  • Updates last_stdout and last_stderr attributes with the most recent output

  • Log file is automatically added to info_files collection

succeed(msg: str = '') → NodeRunner[source]

Mark the task as successfully completed.

This method updates the task status to COMPLETED, sets an optional success message, logs the success, and ensures the cumulative log file is created.

Parameters:

msg (str, optional) – Success message. Defaults to “”.

Returns:

Self reference for chaining.

Return type:

NodeRunner

warning(msg)[source]

Log a warning message with task context.

Parameters:

msg (str) – Warning message to log.

simstack.core.process_results module

async simstack.core.process_results.process_result_helper(result: SimstackResult | Model | bool | BooleanData, task_id: str = 'NA') → Tuple[List[NamedDataReference], List[Model]][source]

Computes the results_references and returns a List[Model]. It works if the result is a SimstackResult, a single Model or a bool.

simstack.core.recompute_artifacts module

async simstack.core.recompute_artifacts.recompute_artifacts(node_registry: NodeRegistry)[source]

Recomputes artifacts for a node and all its children recursively.

First processes all children nodes recursively, then removes and recomputes the current node’s artifacts.

Parameters:

node_registry (NodeRegistry) – The registry entry for the node to process

simstack.core.reset_db module

async simstack.core.reset_db.main()[source]

simstack.core.resource_assignment module

class simstack.core.resource_assignment.ResourceAssignmentResolution(parameters: simstack.models.parameters.Parameters, normalized_call_path: str, matched_rule: simstack.models.resource_assignment.ResourceAssignmentRule | None = None)[source]

Bases: object

matched_rule: ResourceAssignmentRule | None = None
normalized_call_path: str
parameters: Parameters
async simstack.core.resource_assignment.apply_resource_assignment_to_node_registry(db: Database, node_registry: NodeRegistry, *, parent_parameters: Parameters | None = None) → ResourceAssignmentResolution[source]
simstack.core.resource_assignment.empty_slurm_parameters() → SlurmParameters[source]
simstack.core.resource_assignment.normalize_and_validate_effective_parameters(parameters: Parameters | None) → None[source]
simstack.core.resource_assignment.normalize_call_path(call_path: str | None) → str[source]
async simstack.core.resource_assignment.resolve_resource_assignment(db: Database, *, call_path: str | None, base_parameters: Parameters | None, parent_parameters: Parameters | None = None) → ResourceAssignmentResolution[source]

simstack.core.resources module

class simstack.core.resources.AllowedResources[source]

Bases: object

Singleton class that holds a list of allowed resource strings.

This class is problematic because the @node decorator is called before the config is read. @node may contain Parameters which set default values for resources. The solution is to create the AllowedResources singleton class here, and set the resources from context.initialize. Before context.initialize is called, any resource is allowed.

However, resources are validated on read, so before any node is executed, the resources must be set. AllowedResources can be set only once.

add_resource(resource: str) → None[source]

Add a single resource to the list.

clear_resources() → None[source]

Clear all resources from the list.

get_resources() → List[str][source]

Get the list of allowed resources.

has_resource(resource: str) → bool[source]

Check if a resource exists in the list.

property initialized
remove_resource(resource: str) → None[source]
set_resources(resources: List[str]) → None[source]

Set the list of allowed resources.

simstack.core.route_table module

class simstack.core.route_table.RouteTable[source]

Bases: object

add_route_set(source: str, targets: List[str]) → None[source]
clear_routes() → None[source]
classmethod get_instance() → RouteTable[source]

simstack.core.run_docker module

async simstack.core.run_docker.run_docker(registry_entry: NodeRegistry) → bool[source]

simstack.core.run_node module

async simstack.core.run_node.run_node_from_id(node_id: str, resource_str: str, project_root: str = None)[source]

Run a single node by its ID from the database

simstack.core.run_node.run_node_main()[source]

simstack.core.runner module

async simstack.core.runner.async_main(args: Namespace) → None[source]

Async entry point

async simstack.core.runner.initialize_default_resource() → ResourceDefinition | None[source]

Checks if the current resource is the default one. If so, syncs the node and model tables based on config.toml.

simstack.core.runner.runner_main() → None[source]

simstack.core.simstack_result module

class simstack.core.simstack_result.SimstackResult(*, status: TaskStatus = TaskStatus.RETRIEVED, error_message: str | None = None, message: str | None = None, files: List[FileStack] = <factory>, info_files: List[FileStack] = <factory>, **extra_data: Any)[source]

Bases: BaseModel

error_message: str | None
files: List[FileStack]
info_files: List[FileStack]
message: str | None
model_config = {'extra': 'allow'}

Represents the results from a Simstack operation.

This class serves to encapsulate the status, messages, and files resulting from a Simstack operation. It provides structured attributes for the error message, operation message, and categorized files, ensuring clarity in the representation of these aspects. It also supports additional fields through flexible configuration.

Variables:

  • status – Indicates the status of the task, defaulting to TaskStatus.COMPLETED.

  • error_message – Optionally stores an error message if the task encountered issues.

  • message – An optional general message providing additional information about the task.

  • files – A list of FileStack objects that represent the primary files involved in the operation. Defaults to an empty list.

  • info_files – A list of FileStack objects that provide additional or informational files pertaining to the operation. Defaults to an empty list. These files are shown in the results but not passed to the calling function

status: TaskStatus

simstack.core.submit_node module

simstack.core.submit_node.make_executable(file_path: str | PathLike[str]) → None[source]
async simstack.core.submit_node.submit_node(registry_entry: NodeRegistry) → bool[source]

Submit a node to the SLURM queue

simstack.core.task_id module

simstack.core.task_id.clear_task_id() → None[source]

Clear the current task ID.

simstack.core.task_id.get_task_id() → ObjectId | None[source]

Get the current task ID.

simstack.core.task_id.set_task_id(task_id: ObjectId | None) → None[source]

Set the current task ID.

Module contents