simstack.util package

Submodules

simstack.util.async_zip_utils module

async simstack.util.async_zip_utils.async_zip(async_gen1, async_gen2)

Async version of zip for two async generators

async simstack.util.async_zip_utils.async_zip_multiple(*async_generators)

Async version of zip for multiple async generators

simstack.util.b64mixin module

class simstack.util.b64mixin.BytesB64Mixin

Bases: object

Mixin that teaches Pydantic/ODMantic to serialise bytes fields as base-64-encoded ASCII strings when exporting to JSON (dict / response).

model_config = {'json_encoders': {<class 'bytes'>: <function BytesB64Mixin.<lambda>>}}

simstack.util.cleaned_json_schema module

simstack.util.cleaned_json_schema.cleaned_json_schema(cls)

Generates a JSON schema for the given class and its fields, but eliminates all fields which are models, embedded models, or references to models.

Parameters:

cls

Returns:

simstack.util.config_reader module

class simstack.util.config_reader.ConfigReader(db_info: DatabaseInformation | Database, resource_definition: ResourceDefinition, *, project_root: Path, git_list: list[GitRepo] | None = None)

Bases: DatabaseInformation

Represents a configuration reader that integrates with a database and a resource definition system.

async classmethod create(resource_str, db: Database, toml_reader: TomlReader, project_root: Path, **kwargs)

the idea here is to attempt to get:

1. all required data from the kwargs, if available

2. if the data is not all available, get it from the database

property docker: bool

property environment_start: str

property git_list: List[GitRepo]

property hostname

property project_root: Path

property python_paths: List[Path]

property resource: Resource

property server_token: str

property server_url: str

property ssh_key: Path

property workdir: Path

simstack.util.custom_model_dump module

async simstack.util.custom_model_dump.custom_model_dump(self, **kwargs) → Dict[str, Any]

Custom model dump method to handle the conversion of model instances to dictionaries. This method recursively traverses dictionaries and lists to convert any nested model instances to their dictionary representation.

Parameters:

• self – The model instance

• kwargs – Additional keyword arguments

Returns:

A dictionary representation of the model instance

simstack.util.database_information module

class simstack.util.database_information.DatabaseInformation(db_name: str, connection_string: str, db_type: DBType = DBType.MONGODB, server_url: str | None = None, server_token: str | None = None)

Bases: object

Represents a database and its connection information.

This class encapsulates essential details about a database, including its name and connection string. It provides a convenient method for initializing this information from a configuration file, facilitating streamlined database setups and connections.

_db_name

The name of the database.

Type:

str

_connection_string

The connection string used to access the database.

Type:

str

from_config_file(config, **kwargs)

Initialize DatabaseInformation from TOML config or a database specified in TOML.

property connection_string: str

property db_name: str

property db_type: DBType

classmethod from_config(config: Dict[str, Any], **kwargs)

Initialize DatabaseInformation from TOML config kwargs override config file.

classmethod from_db_info(db_info: DatabaseInformation)

classmethod from_db_info_or_db(db_info: 'DatabaseInformation' | 'Database')

get_information()

Returns a tuple of the initialization parameters that can be used as *args for __init__.

Returns:

(db_name, connection_string, db_type)

Return type:

tuple

property server_token: str

property server_url: str

simstack.util.db module

class simstack.util.db.Database(*, client: AsyncIOMotorClient | None = None, database_name: str | None = None, connection_string: str | None = None, engine: Any | None = None, db_type: DBType | None = None, server_url: str | None = None, server_token: str | None = None)

Bases: object

Server-owned database facade.

The server should depend on this object instead of reaching into SimStack core engine abstractions directly. The underlying persistence primitive is the plain ODMantic engine; SimStack-specific behavior lives in this facade.

async apply_resource_assignment_to_node_registry(node_registry: Any) → Any

property client: motor.motor_asyncio.AsyncIOMotorClient

async close() → None

collection(model_or_name: Any) → Any

property connection_string: str | None

property core_engine: Any

Compatibility escape hatch for SimStack core internals only.

property databae_type: DBType

property database: Any

property database_name: str

async delete(*args: Any, **kwargs: Any) → Any

async find(*args: Any, **kwargs: Any) → Any

async find_all(*args: Any, **kwargs: Any) → Any

async find_all_artifacts(node_registry: Any) → Any

async find_artifact_mappings(node_registry_path: str) → Any

async find_one(*args: Any, **kwargs: Any) → Any

async find_one_by_model_name(model_mapping: str, item_id: str | ObjectId) → Any | None

classmethod from_db_info(db_info: DatabaseInformation) → Database

get_collection(model_or_name: Any) → Any

Temporary compatibility alias for code still being migrated.

async load_task(name: str, arg_hash: str, function_hash: str) → NodeRegistry | None

Load a task based on name, arg_hash and function_hash

Parameters:

• name – Node name

• arg_hash – Hash of the arguments

• function_hash – Hash of the function

Returns:

The found NodeRegistry instance or None

async load_task_by_id(task_id: str | ObjectId) → NodeRegistry | None

Load a task based on its ID

Parameters:

task_id – The task ID

Returns:

The found NodeRegistry instance or None

async load_waiting_tasks_for_resource(resource: str) → List[NodeRegistry]

Load all waiting tasks for a specific resource

Parameters:

resource – The resource name

Returns:

List of matching NodeRegistry instances

async ping() → Any

property raw_database: Any

async reset_database() → None

Reset the database by dropping all collections and recreating them

async save(*args: Any, **kwargs: Any) → Any

async save_unchecked(*args: Any, **kwargs: Any) → Any

property server_token: str | None

property server_url: str | None

async stats() → Any

async simstack.util.db.find_all_artifacts_for_database(database: Database, node_registry: Any) → Any

simstack.util.db_logger module

class simstack.util.db_logger.DBLogHandler(connection_string: str, db_name: str, collection_name: str = 'logs')

Bases: Handler

Logging handler that stores logs in MongoDB using pymongo.

close() → None

Close the MongoDB client connection.

create_log_entry(record: LogRecord) → Dict[str, Any]

emit(record: LogRecord) → None

Process a log record by converting it to a log entry and inserting it into MongoDB.

simstack.util.db_logger.extract_resource_id(message: str) → str | None

Extract resource ID from log message.

simstack.util.db_logger.extract_task_id(message: str) → str | None

Extract task ID from log message.

simstack.util.default_from_dict module

simstack.util.default_from_dict.default_from_dict(cls_method, data: dict, **kwargs) → Any

Create an instance of the model from a dictionary. Handles nested models and enum values.

simstack.util.default_from_dict.default_from_model(cls_method, model: Model, **kwargs) → Model

simstack.util.directory_iterator module

class simstack.util.directory_iterator.DirectoryPath(root_path: str | Path, excluded_patterns: List[str] | None = None, file_extensions: List[str] | None = None)

Bases: object

Represents a directory path with excluded subdirectory patterns and provides iteration capabilities.

add_excluded_pattern(pattern: str) → None

Add an exclusion pattern.

Parameters:

pattern – Pattern to exclude

get_directories_list() → List[Path]

Get a list of all directories matching the criteria.

Returns:

List of Path objects

get_files_list() → List[Path]

Get a list of all files matching the criteria.

Returns:

List of Path objects

iterate_directories() → Iterator[Path]

Iterate over all directories in the directory tree, respecting exclusion patterns.

Yields:

Path objects for each directory found

iterate_files() → Iterator[Path]

Iterate over all files in the directory tree, respecting exclusion patterns.

Yields:

Path objects for each file found

remove_excluded_pattern(pattern: str) → None

Remove an exclusion pattern.

Parameters:

pattern – Pattern to remove

simstack.util.directory_iterator.find_python_files_with_exclusions(root_path: str, excluded_patterns: List[str] = None) → List[Path]

Find Python files using the DirectoryPath class with exclusions.

Parameters:

• root_path – Root directory to search

• excluded_patterns – Patterns to exclude

Returns:

List of Python file paths

simstack.util.docstring_parser module

class simstack.util.docstring_parser.DocstringParser(docstring: str | None)

Bases: object

Parse a docstring into structured components.

All component getters return None when the corresponding section is not present. Supported sections:

• Args: / Parameters:

• Returns:

• SimstackResult:

• CalledNodes:

• Raises:

as_dict() → Dict[str, Any]

Backward-compatible structure (always includes keys), while still using the None-returning component getters internally.

called_nodes() → List[str] | None

description() → str | None

params() → Dict[str, Dict[str, str | None]] | None

raises() → Dict[str, Dict[str, str]] | None

returns() → Dict[str, str] | None

simstack_results() → Dict[str, Dict[str, str]] | None

simstack.util.docstring_parser.parse_docstring(docstring: str | None) → Dict[str, Any]

Backward-compatible wrapper around DocstringParser.

Prefer using DocstringParser directly when you want per-section access with None when a section is missing.

simstack.util.file_hashing module

simstack.util.file_hashing.hash_file(file_path: str | Path, algorithm: str = 'sha256', chunk_size: int = 8192) → str

Calculate hash of a file on disk.

Parameters:

• file_path – Path to the file

• algorithm – Hash algorithm (md5, sha1, sha256, etc.)

• chunk_size – Size of chunks to read

Returns:

Hexadecimal digest of the hash

Return type:

str

simstack.util.file_hashing.hash_file_object(file_obj: BinaryIO, algorithm: str = 'sha256', chunk_size: int = 8192) → str

Calculate hash of a file-like object.

Parameters:

• file_obj – File-like object (must be in binary mode)

• algorithm – Hash algorithm (md5, sha1, sha256, etc.)

• chunk_size – Size of chunks to read

Returns:

Hexadecimal digest of the hash

Return type:

str

simstack.util.file_hashing.hash_string(string: str, algorithm: str = 'sha256') → str

simstack.util.file_transfer_client module

class simstack.util.file_transfer_client.DownloadResult(path: 'Path', size_bytes: 'int', checksum_sha256: 'str')

Bases: object

checksum_sha256: str

path: Path

size_bytes: int

class simstack.util.file_transfer_client.FileTransferClient(*, server_url: str, runner_token: str, timeout_seconds: int = 60)

Bases: object

Small standard-library HTTP client for runner-to-server file transfer.

It intentionally avoids adding a mandatory requests/httpx dependency to the runner environment while still streaming upload and download bodies.

chunk_size = 1048576

complete_transfer(*, transfer_id: str, target_resource_name: str, target_path: str, size_bytes: int, checksum_sha256: str) → Dict[str, Any]

create_transfer(*, file_stack_id: str, source_file_instance_id: str | None, source_resource_name: str | None, target_resource_name: str, request_type: str = 'runner_to_runner') → Dict[str, Any]

download_file(transfer_id: str, target_path: Path) → DownloadResult

fail_transfer(transfer_id: str, *, error_message: str, error_code: str | None = None) → Dict[str, Any]

classmethod from_context(*, required: bool = True) → FileTransferClient | None

Build a client from environment variables or context config attributes.

Environment variables are preferred because runner tokens are machine credentials and should not be stored in FileStack/FileInstance objects.

get_transfer(transfer_id: str) → Dict[str, Any]

list_transfers(*, role: str = 'source', status: str | None = 'created', limit: int = 20) → list[Dict[str, Any]]

upload_file(transfer_id: str, path: Path) → Dict[str, Any]

wait_until_uploaded(transfer_id: str, *, timeout_seconds: int | None = None, poll_interval_seconds: float | None = None) → Dict[str, Any]

exception simstack.util.file_transfer_client.FileTransferError

Bases: RuntimeError

Raised when the SimStack file transfer API cannot satisfy a request.

simstack.util.file_transfer_client.first_available_remote_location(locations: Iterable[Any], local_resource: Any) → Any | None

simstack.util.file_transfer_client.path_for_file_instance(path: Path, workdir: Path) → str

simstack.util.file_transfer_client.resolve_instance_path(raw_path: str, workdir: Path) → Path

simstack.util.file_transfer_client.resource_name(value: Any) → str

simstack.util.file_transfer_client.transfer_resource_name(value: Any) → str

Return the concrete resource name used by file transfer endpoints.

SimStack uses the symbolic resource “self” for nodes that run on the current resource. The server transfer API cannot authorize “self” because runner tokens are scoped to concrete resources such as “local” or “int-nano”. When possible, resolve “self” through the configured runner token resource before sending transfer requests or storing file instances.

simstack.util.files module

class simstack.util.files.FfspecFile(*, path: str = <odmantic.field.ODMFieldInfo object>, name: str = <odmantic.field.ODMFieldInfo object>, extension: str = <odmantic.field.ODMFieldInfo object>, protocol: str = <odmantic.field.ODMFieldInfo object>, host: str | None = <odmantic.field.ODMFieldInfo object>, port: int | None = <odmantic.field.ODMFieldInfo object>, username: str | None = <odmantic.field.ODMFieldInfo object>, ssh_auth: SshAuthMethod | None = <odmantic.field.ODMFieldInfo object>, size: int = <odmantic.field.ODMFieldInfo object>, created_at: datetime = <odmantic.field.ODMFieldInfo object>, modified_at: datetime = <odmantic.field.ODMFieldInfo object>, storage_options: Dict[str, ~typing.Any]=<odmantic.field.ODMFieldInfo object>, checksum: str | None = <odmantic.field.ODMFieldInfo object>, content_type: str | None = <odmantic.field.ODMFieldInfo object>, parent_path: str | None = <odmantic.field.ODMFieldInfo object>, is_directory: bool = <odmantic.field.ODMFieldInfo object>, children: List[str] = <odmantic.field.ODMFieldInfo object>, last_accessed: datetime | None = <odmantic.field.ODMFieldInfo object>, access_count: int = <odmantic.field.ODMFieldInfo object>, metadata: Dict[str, ~typing.Any]=<odmantic.field.ODMFieldInfo object>)

Bases: BaseModel

Representation of a file with protocol-specific details, metadata, and utility methods for remote or local operations.

This class facilitates working with files across different storage protocols such as local filesystem, SFTP, SCP, or cloud-based protocols. It provides attributes for authenticating, accessing, and manipulating files as well as methods for operations like copying files, creating URIs, and testing connections.

Variables:

• path – Full path to the file.

• name – Name of the file.

• extension – File extension.

• protocol – Storage protocol (e.g., ‘file’, ‘s3’, ‘http’, ‘sftp’, ‘scp’).

• host – Host name or address where the file is located.

• port – Port number if applicable.

• username – Username for authentication if required.

• ssh_auth – SSH authentication configuration.

• size – Size of the file in bytes.

• created_at – Creation timestamp.

• modified_at – Last modification timestamp.

• storage_options – Protocol-specific storage options.

• checksum – File content checksum.

• content_type – MIME type of the file.

• parent_path – Path to parent directory.

• is_directory – Whether this is a directory.

• children – Child file/directory paths if this is a directory.

• last_accessed – Last access timestamp.

• access_count – Number of times this file has been accessed.

• metadata – Additional metadata as key-value pairs.

access_count: int

can_connect() → bool

Test if we can connect to the remote host using the provided credentials.

Returns:

True if connection is successful, False otherwise

Return type:

bool

checksum: str | None

children: List[str]

content_type: str | None

copy_from_local(local_path: str) → None

Copy a local file to the remote location represented by this model.

Parameters:

local_path – Path to the local file

Returns:

None

copy_to_local(local_path: str | None = None) → str

Copy the remote file to a local path.

Parameters:

local_path – Local path to copy the file to. If None, a temporary file is created.

Returns:

Path to the local copy of the file

Return type:

str

created_at: datetime

extension: str

classmethod from_ssh_key(host: str, path: str, username: str, key_filename: str, protocol: str = 'sftp', port: int = 22, key_passphrase: str | None = None, host_key_policy: str = 'auto_add', **kwargs) → FfspecFile

Create a FfspecFile instance with SSH key authentication.

Parameters:

• host – Remote host name or IP address

• path – Path to the file on the remote host

• username – SSH username for authentication

• key_filename – Path to the SSH private key file

• protocol – Protocol to use, either ‘sftp’ or ‘scp’ (default: ‘sftp’)

• port – SSH port on the remote host (default: 22)

• key_passphrase – Passphrase for the SSH key if it’s encrypted (default: None)

• host_key_policy – Host key policy - ‘auto_add’, ‘strict’, or ‘ask’ (default: ‘auto_add’)

• **kwargs – Additional attributes to set on the model

Returns:

A new instance representing the remote file

Return type:

FfspecFile

get_filesystem()

Create and return a fsspec filesystem object for this file.

Returns:

An fsspec filesystem instance

Return type:

object

host: str | None

is_directory: bool

last_accessed: datetime | None

metadata: Dict[str, Any]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

modified_at: datetime

name: str

parent_path: str | None

path: str

port: int | None

protocol: str

size: int

ssh_auth: SshAuthMethod | None

storage_options: Dict[str, Any]

to_uri() → str

Convert this model to a URI/URL representation.

Note: SSH private key information is not included in the URI.

Returns:

URI string representing this file

Return type:

str

username: str | None

class simstack.util.files.SshAuthMethod(*, auth_type: str = <odmantic.field.ODMFieldInfo object>, password: str | None = <odmantic.field.ODMFieldInfo object>, key_filename: str | None = <odmantic.field.ODMFieldInfo object>, key_passphrase: str | None = <odmantic.field.ODMFieldInfo object>, allow_host_key_add: bool = <odmantic.field.ODMFieldInfo object>, host_key_policy: str = <odmantic.field.ODMFieldInfo object>, known_hosts_file: str | None = <odmantic.field.ODMFieldInfo object>, **extra_data: Any)

Bases: BaseModel

allow_host_key_add: bool

auth_type: str

host_key_policy: str

key_filename: str | None

key_passphrase: str | None

known_hosts_file: str | None

model_config = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

password: str | None

simstack.util.find_python_files module

simstack.util.find_python_files.find_python_files(directory: str) → List[str]

Returns full paths of all .py files in the given directory and its subdirectories, excluding __init__.py files.

Parameters:

directory – The root directory to search from

Returns:

List of absolute file paths to all .py files that aren’t __init__.py

simstack.util.generate_ui_schema module

simstack.util.generate_ui_schema.generate_ui_schema(cls)

Generates a UI schema that uses GenericForm for fields with ui_schema function. Also preserves any existing UI schema configurations from the class.

Parameters:

cls – The model class to generate UI schema for

Returns:

The generated UI schema

Return type:

dict

simstack.util.generic_list_mixin module

class simstack.util.generic_list_mixin.GenericListMixin

Bases: Generic[T]

Mixin class containing common functionality for list operations.

Notes on typing:

• Pure list-like operations (append/extend/…) are fully generic over T.

• Convenience helpers like find() / filter_by_size() rely on optional attributes (e.g. .name, .size). For arbitrary T, we use getattr() to keep runtime behavior flexible while remaining type-safe-ish.

append(elements: T)

clear()

copy() → List[T]

count(elements: T) → int

extend(elements: List[T] | Iterable[T] | GenericListMixin[T])

filter_by_property(property_name: str, value: Any) → Iterator[T]

filter_by_size(min_size: int = None, max_size: int = None) → Iterator[T]

find(pattern: str) → T | None

find_all(pattern: str) → Iterator[T]

index(elements: T, start: int = 0, stop: int = None) → int

insert(index: int, elements: T)

pop(index: int = -1) → T

remove(elements: T)

reverse()

sort(key=None, reverse: bool = False)

sort_by_name(reverse: bool = False)

sort_by_size(reverse: bool = False)

simstack.util.geometry_utils module

simstack.util.get_module_path module

simstack.util.get_module_path.get_module_path(file_path: str)

simstack.util.git_pull module

simstack.util.git_pull.git_pull_periodically(repo_path, interval_minutes=1, log_file=None)

Performs git pull in the specified repository at regular intervals.

Parameters:

• repo_path (str) – Path to the Git repository

• interval_minutes (int) – Time interval between pulls in minutes

• log_file (str, optional) – Path to log file. If None, logs to console.

Returns:

None

simstack.util.git_repository_status module

simstack.util.git_repository_status.get_git_status(repo_path: Path) → Dict[str, object | None]

Get the current Git short commit hash and whether the branch is up to date with its upstream.

Returns a dict:

{

“short_hash”: Optional[str], # e.g., “a1b2c3d”, None if not a git repo “branch”: Optional[str], # current branch name or “HEAD” if detached, None if not a repo “up_to_date”: Optional[bool], # True/False if upstream exists, None if unknown/no upstream “ahead”: Optional[int], # commits ahead of upstream (0 if up to date), None if unknown “behind”: Optional[int], # commits behind upstream (0 if up to date), None if unknown

}

simstack.util.import_module module

simstack.util.import_module.import_module_from_file(file_path: Path, root_dir: Path)

Import a Python file as a module.

Parameters:

• root_dir

• file_path – Path object pointing to the Python file to import

Returns:

Imported module or None if import failed

simstack.util.importer module

async simstack.util.importer.import_class(class_path: str, db: Database) → Type[Model] | None

Dynamically import a class from a module using its full path. First tries to load the class from the database using ModelMapping

A pickled version of the class is used primarily :param : param class_path: class_path: Dot notation path to the class (e.g. ‘models.submodule.ClassName’) :param : param db: db: Database object

Returns:

The imported class object or None if import fails

async simstack.util.importer.import_class_by_name(class_name: str, db: Database) → Type[Model]

async simstack.util.importer.import_function(function_path: str, db: Database, task_id: ObjectId | None = None, tolerate_missing_function: bool = False) → Callable[[...], Any] | None

Dynamically import a function from a module using its full path, including a migration mechanism. load the function information using NodeModel load the pickled version if it exists if there is no pickled version, use regular import.

Parameters:

• function_path – Dot notation path to the function (e.g. ‘methods.submodule.function_name’)

• db – Database object

• task_id – Optional task id

• tolerate_missing_function – If True, return None if function is not found, otherwise raise exception

Returns:

The imported function object or None if import fails

async simstack.util.importer.import_function_by_name(function_name: str, db: Database, task_id: ObjectId) → Callable[[...], Any] | None

simstack.util.init_data_source module

async simstack.util.init_data_source.initialize_resource_from_db(resource_str: str, db: Database, workdir_self: Path) → ResourceDefinition

simstack.util.make_graph module

simstack.util.make_graph.model_to_graph_data(model, exclude_none=True, exclude_unset=True, exclude=None)

Checks if an odmantic Model contains fields that can be interpreted as graph data (multiple lists of the same length) and converts it to a format suitable for ag-graph.

Parameters:

• model – An odmantic Model instance

• exclude_none (bool) – Whether to exclude None values

• exclude_unset (bool) – Whether to exclude unset values

• exclude (set) – Fields to exclude

Returns:

A dictionary with graph data if the model contains compatible fields, otherwise None

Return type:

dict

simstack.util.make_table module

simstack.util.make_table.create_column_def(field_name, field_type, field_value, field_path)

Create a column definition for AG Grid based on field type and value.

Parameters:

• field_name – Name of the field

• field_type – Type of the field from model definition

• field_value – Actual value of the field (can be None for class-based definitions)

• field_path – Full path to the field (for nested structures)

Returns:

Column definition for AG Grid

Return type:

dict

simstack.util.make_table.is_pydantic_model(obj)

Check if an object is a Pydantic model class.

Parameters:

obj – The object to check

Returns:

True if the object is a Pydantic model class, False otherwise

Return type:

bool

simstack.util.make_table.make_column_defs_helper(model_class, table_name=None, max_recursion_level=1, drop_id=True, current_level=0, visited=None, field_prefix='')

Create column definitions for AG Grid based on a model class. Uses the same logic as table data generation but for model class. This is a helper function mapped to make_column_defs in simstack_model. The idea is that most classes can call the helper function but some classes might want to override it to return specific column definitions.

Parameters:

• model_class – The model class to process

• table_name – Optional name for the table (default: None, uses class name)

• max_recursion_level – Maximum depth for processing nested models (default: 1)

• drop_id – Whether to drop the ID field (default: True)

• current_level – Current recursion level (default: 0)

• visited – Set of objects already visited to prevent infinite recursion (default: None)

• field_prefix – Prefix for field paths in nested structures (default: “”)

Returns:

Column definitions for AG Grid

Return type:

list

simstack.util.make_table.make_column_defs_instance(model_instance: Model, table_name: str | None = None, max_recursion_level: int = 1, drop_id: bool = True, current_level: int = 0, visited: Set[int] | None = None, field_prefix: str = '') → List[Dict[str, Any]]

Create column definitions for AG Grid based on a model instance. Uses the same logic as table data generation but works on an actual instance. This allows for dynamic column generation based on actual data values.

Parameters:

• model_instance – The model instance to process

• table_name – Optional name for the table (default: None, uses class name)

• max_recursion_level – Maximum depth for processing nested models (default: 1)

• drop_id – Whether to drop the ID field (default: True)

• current_level – Current recursion level (default: 0)

• visited – Set of objects already visited to prevent infinite recursion (default: None)

• field_prefix – Prefix for field paths in nested structures (default: “”)

Returns:

Column definitions for AG Grid

Return type:

list

simstack.util.make_table.make_table_entries(model_instances, table_name=None, max_recursion_level=1, drop_id=True)

Create AG Grid table configuration for a list of model instances. This is the main function to make tables that are visible in the ui.

To generate a table in the ui the ui_schema of the class should initialize the grid_options for the table.

It is basically impossible to predict what a table should be for a class but tables can be created only from lists. The make_table function of the class should therefore call this function to generate the table for appropriate list fields and merge these manually where it makes sense.

This function calls the helper functions make_table_entries_helper and make_column_defs_helper which recursively process the model instances and their fields to generate the table data and column definitions.

for model fields these functions check if the nested model has functions make_columns or make_column_defs which are called to generate the column data and definitions, respectively

These functions do not exist by default but can be used by models to modify the default definitions where they dont work. This is typically the case if the models contain huge data sets that we do not want to display in the table. An example of this is in the Molecule class.

Parameters:

• model_instances – List of model instances to process

• table_name – Name for the table (default: None, uses class name)

• max_recursion_level – Maximum depth for processing nested models (default: 1)

• drop_id – Whether to drop the ID field (default: True)

Returns:

Dictionary with ‘tableName’, ‘rowData’ (for AG Grid), and ‘columnDefs’

Return type:

dict

simstack.util.make_table.make_table_entries_helper(model_instance, table_name=None, max_recursion_level=1, drop_id=True, current_level=0, visited=None, field_prefix='')

Create table data for AG Grid from a model instance. Handles datetime objects properly and can optionally drop ID fields.

Parameters:

• model_instance – The model instance to process

• table_name – Optional name for the table (default: None, uses class name)

• max_recursion_level – Maximum depth for processing nested models (default: 1)

• drop_id – Whether to drop the ID field (default: True)

• current_level – Current recursion level (default: 0)

• visited – Set of objects already visited to prevent infinite recursion (default: None)

• field_prefix – Prefix for field paths in nested structures (default: “”)

Returns:

Dictionary with ‘tableName’ and ‘tableData’ (rows data for AG Grid)

Return type:

dict

simstack.util.make_table.ui_add_table(ui_options: Dict[str, Any]) → Dict[str, Any]

Add table options to the UI schema for AG Grid.

the class must then implement o make_table function calling make_table :param ui_options: :return: modified ui_options

simstack.util.mappings module

class simstack.util.mappings.ModelMappingTable(mappings: List[ModelMapping])

Bases: object

get_by_mapping(mapping: str) → ModelMapping | None

get_by_name(name: str) → ModelMapping | None

async classmethod load(engine: AIOEngine)

class simstack.util.mappings.NodeMappingTable(nodes: List[NodeModel])

Bases: object

get_by_mapping(mapping: str) → NodeModel | None

get_by_name(name: str) → NodeModel | None

async classmethod load(engine: AIOEngine)

simstack.util.minimal_route_finder module

simstack.util.minimal_route_finder.find_minimal_route(routes: List[Dict[str, str]], source: str, target: str) → List[Dict[str, str]]

Find the minimal (shortest) route from source to target.

Parameters:

• routes – List of route dictionaries with ‘source’, ‘target’, and ‘host’

• source – The starting node

• target – The destination node

Returns:

A list of route dictionaries representing the shortest path from source to target, or an empty list if no path exists

simstack.util.minimal_route_finder.find_shortest_route(routes: List[Dict[str, str]], source: str, target: str) → List[Dict[str, str]]

Find the minimal (shortest) route from source to target.

Parameters:

• routes – List of route dictionaries with ‘source’, ‘target’, and ‘host’

• source – The starting node

• target – The destination node

Returns:

A list of route dictionaries representing the shortest path from source to target, or an empty list if no path exists

simstack.util.mock_db module

class simstack.util.mock_db.InMemoryAIOEngine

Bases: object

async find(model: Type[T], **query) → List[T]

async save(instance: T) → T

class simstack.util.mock_db.InMemoryCollection(engine: InMemoryAIOEngine)

Bases: object

async simstack.util.mock_db.find_one(self, model: Type[T], condition=None, **query) → T | None

simstack.util.module_path_checker module

simstack.util.module_path_checker.is_module_subpath_of_path(module_path: str, path_info_path: Path) → bool

Check whether a dot-separated module path is a subpath of an absolute path.

Parameters:

• module_path – A dot-separated path relative to the project root (e.g., “src.simstack.core.artifacts”)

• path_info_path – An absolute path below the project root (e.g., “/path/to/project/src/simstack/core”)

Returns:

True if the module path corresponds to a location under the given absolute path, False otherwise

simstack.util.object_list_mixin module

class simstack.util.object_list_mixin.ObjectListMixin

Bases: GenericListMixin[ObjectId], Generic[T]

Mixin class for lists of Model ObjectIDs. Stores ObjectId instances in elements, but allows interaction with Model instances.

append(element: T)

clear()

copy() → ObjectListMixin[T]

count(element: T) → int

async db_find_postprocess(db: Database)

Instance-level post-processing

delete_element(element: T | ObjectId)

extend(elements: List[T] | ObjectListMixin[T])

find(pattern: str) → T | None

find_all(pattern: str) → Iterator[T]

get(index: int) → T

Get an element by index, loading it from DB if necessary.

index(element: T, start: int = 0, stop: int = None) → int

insert(index: int, element: T)

pop(index: int = -1) → T

remove(element: T)

reverse()

async save(db: Database)

sort(key=None, reverse: bool = False)

simstack.util.path_manager module

class simstack.util.path_manager.PathManager(*args, **kwargs)

Bases: object

Manages paths for the SimStack application, providing mechanisms to find Python files for nodes and models. Will read only .py files. By default travers all directories below the project root. Uses DirectoryPath for efficient directory traversal.

add_path(name: str, path: Path, drops: str = '', use_pickle: bool = False) → None

Add a path to the manager.

Parameters:

• name – Name identifier for the path

• path – The directory path relative to the project root

• drops – Prefix to drop from module names (for import paths)

• use_pickle – Whether to use pickle for this path

default_excludes()

find_parent_path(path: str) → str | None

Find the parent path from self.paths that contains the given path.

Parameters:

path – The path to find the parent of

Returns:

The name of the parent path from self.paths that contains the given path, or None if no parent is found

find_python_files(path_name: str, excluded_patterns: List[str] | None = None) → List[str]

Find Python files in the specified path, excluding __init__.py files.

Parameters:

• path_name – Name of the path to search in

• excluded_patterns – Additional patterns to exclude

Returns:

List of absolute file paths to Python files

classmethod from_config(config: Dict[str, Any]) → PathManager

Create a PathManager from configuration.

Parameters:

config – Configuration object containing path information

Returns:

Initialized PathManager instance

get_drops(path_name: str) → str

Get the drops value for a path.

Parameters:

path_name – Name of the path

Returns:

The drops value for the path

get_path(name: str) → Dict[str, str]

Get a path by name.

Parameters:

name – Name of the path to retrieve

Returns:

Dictionary containing path information

iterate_python_files(path_name: str, excluded_patterns: List[str] | None = None) → Iterator[Path]

Iterate over Python files in the specified path, excluding __init__.py files.

Parameters:

• path_name – Name of the path to search in

• excluded_patterns – Additional patterns to exclude

Returns:

Iterator of Path objects for Python files

reset()

simstack.util.project_root_finder module

simstack.util.project_root_finder.find_project_root(current_file=None, marker_files=('.git', 'simstack.toml', 'setup.py'), skip_files=('simstack_marker.com',)) → Path

Find the project root directory by searching for common marker files

Parameters:

• skip_files – directories with these names will be skipped

• current_file – Path to the current file, defaults to __file__ if None

• marker_files – Tuple of files/directories that indicate the project root

Returns:

Absolute path to the project root directory

simstack.util.queue_watcher module

simstack.util.queue_watcher.main() → int

simstack.util.queue_watcher.parse_spec(cmd_file: Path) → Tuple[str | None, Dict[str, str], str | None, int | None]

simstack.util.queue_watcher.process_job_from_claim(base_stem: Path, claimed_by_lock: bool) → None

Process a job using base path (without suffix). If claimed_by_lock is True, we did not rename .ready; we will remove it at the end. Otherwise, we operate on .claim and remove it.

simstack.util.queue_watcher.run_command(command: str, cwd: str | None, env: Dict[str, str], timeout: int | None, out_file: Path, err_file: Path) → int

simstack.util.queue_watcher.try_claim(ready_file: Path, retries: int = 5, backoff: float = 0.05) → Path | None

Atomically rename *.ready -> *.claim. Retries on transient failures. Returns claim path if successful, else None.

simstack.util.queue_watcher.try_lock(base_stem: Path) → bool

Create an exclusive lock file <id>.lock using O_CREAT|O_EXCL. Returns True if lock acquired, False otherwise.

simstack.util.resource_config module

class simstack.util.resource_config.ResourceConfig(config_path: Path, resource: str)

Bases: object

ResourceConfig is responsible for managing configuration settings, setup, execution, and post-processing parameters for specified resources.

This class is designed to read configuration data from a TOML file, provide an interface to access resource-specific parameters, and execute resource-related operations such as setup and running commands. It encapsulates functionality for handling temporary directories, file manipulation, and subprocess execution.

_config

The loaded configuration dictionary.

Type:

Dict[str, Any]

_resource

The name of the current resource.

Type:

str

get_postprocessing_params() → Dict[str, Any]

Returns the post-processing dict for the specified resource. Expected structure in TOML: [resource_name.post-processing] or [resource_name.postprocessing]

get_program(program_name: str) → Dict[str, Any]

Returns the dict from resource.program.name for program with name and the current resource. Expected structure in TOML: [resource_name.program.program_name]

get_setup_params() → Dict[str, Any]

Returns the setup dict for the specified resource. Expected structure in TOML: [resource_name.setup]

property os: str

Returns the OS of the current resource, defaults to ‘linux’.

run(program_name: str, input_files: List[str | FileStack] | None = None, output_files: List[str | FileStack] | None = None, node_runner: Any | None = None)

Executes the run command with optional temporary directory usage and file handling. Retrieves parameters from the configuration for the specified program.

Parameters:

• program_name – Name of the program to run.

• input_files – List of input files (str or FileStack). Overrides TOML input_files if provided.

• output_files – List of output files (str or FileStack). Overrides TOML output_files if provided.

• node_runner – Optional NodeRunner instance for execution.

setup(node_runner: Any | None = None)

Executes the setup scripts for the current resource.

property tmp_base_dir: Path

tmp_dir(task_id: ObjectId | str) → Path

simstack.util.route_finder module

simstack.util.route_finder.find_route(target, source)

Find the minimal route from source to target using available routes.

Parameters:

• target (str) – The target system

• source (str) – The source system

Returns:

A list of systems to visit in order, or None if no route exists

Return type:

list

simstack.util.runner_util module

simstack.util.runner_util.ensure_crontab_entry(command, schedule='*/10 * * * *')

Checks if a crontab entry exists and adds it if it doesn’t.

Parameters:

• command (str) – The command to be executed by cron

• schedule (str) – The cron schedule expression (default: every 10 minutes)

Returns:

True if entry was added, False if it already existed

Return type:

bool

async simstack.util.runner_util.graceful_shutdown(resource_name)

async simstack.util.runner_util.restart(resource_name)

async simstack.util.runner_util.schedule_restart(resource_name, restart_minutes)

Schedule a restart after the specified minutes

simstack.util.runner_utils module

async simstack.util.runner_utils.clean_slurm_info(resource: Resource, user: str | None = None) → None

Clean up old slurm info entries

simstack.util.runner_utils.get_job_info(job_id: str, task_id: ObjectId, resource: Resource) → SlurmInfo | None

Get job information from SLURM queue using squeue

simstack.util.runner_utils.make_git_status_list() → List[str]

simstack.util.runner_utils.run_squeue_for_job(job_id: str) → str

simstack.util.safe_code_executor module

simstack.util.safe_code_executor.safe_code_executor(code_string: str, artifact_arguments: ArtifactArguments, timeout: int = 30) → Dict

Safely executes Python code from a string with a controlled environment.

Parameters:

• code_string (str) – The Python code to execute

• artifact_arguments (ArtifactArguments) – Artifact arguments to be used in the execution environment

• timeout (int, optional) – Maximum execution time in seconds before timeout

Returns:

A dictionary containing:

’success’ (bool): Whether execution was successful ‘result’: The return value if successful ‘error’: Error message if unsuccessful ‘error_type’: Type of error if unsuccessful

Return type:

Dict

simstack.util.sample_config module

simstack.util.sample_config.sample_config_writer_entrypoint() → None

Entry point to write a sample simstack.toml configuration file.

simstack.util.sensitive_content_filter module

simstack.util.sensitive_content_filter.filter_file_content(data: Dict[str, Any]) → Dict[str, Any]

Recursively searches through nested data structures to filter out sensitive content. Specifically looks for ‘content’ fields in dictionaries, especially within ‘files’ structures.

Parameters:

data – Any Python data structure (dict, list, etc.)

Returns:

The filtered data structure with sensitive content removed

simstack.util.setup_logging module

simstack.util.setup_logging.setup_logging(connection_string: str, db_name: str, log_level=20, console=True, log_format: str = '%(asctime)s - %(name)-15s - %(levelname)-10s - %(filename)-20s:%(lineno)4d - %(message)s')

Configures the root logger with specified logging level, handlers for both database and optionally the console.

Parameters:

• log_format

• connection_string – A string representing the database connection.

• db_name – A name of the database to log into.

• log_level – Logging verbosity level, defaulting to logging.INFO.

• console – A boolean indicating whether to log messages to the console.

Returns:

Configured root logger instance.

simstack.util.submit_to_watchdog module

class simstack.util.submit_to_watchdog.WatchdogResult(*, job_id: str, status: str, returncode: int | None, stdout: str, stderr: str, paths: dict[str, str])

Bases: BaseModel

job_id: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

paths: dict[str, str]

returncode: int | None

status: str

stderr: str

stdout: str

simstack.util.submit_to_watchdog.submit_to_watchdog(command: str, job_id: str, queue_dir: Path = PosixPath('/home/runner/work/simstack/simstack/queue'), env: dict | None = None, cwd: str | None = None) → WatchdogResult

Submit a command to be executed by the file-queue watcher. Returns a WatchdogResult with status, exit_code, stdout, stderr, and paths.

simstack.util.toml_reader module

class simstack.util.toml_reader.TomlReader(config_path: Path, config_file: Path = PosixPath('simstack.toml'))

Bases: object

build_routes()

property config

get(key: str, default=None)

get_allowed_resources()

get_git_list()

get_resource_definition(resource_str) → ResourceDefinition

Retrieves the resource definition for the given resource string from the TOML configuration. Raises ValueError if the resource is not allowed or if the resource definition is not found. Sets the allowed resources list from the config file.

get_routes(resource_str: str)

initialize_path_manager()

use_db()

simstack.util.transform_file_name module

simstack.util.transform_file_name.transform_file_name(path_str: str | Path, project_root: Path | None = None) → Path

Transform a string path by substituting environment variables.

Parameters:

• path_str – String containing path with optional environment variables ($HOME, $PROJECT, $TEMP)

• project_root – Optional project root directory. If not provided, it will be automatically detected.

Returns:

Path object with resolved environment variables

Raises:

FileNotFoundError – If the resolved path does not exist

simstack.util.ui_tools module

simstack.util.ui_tools.ui_hide_fields(ui_schema: Dict[str, Any], fields: str | List[str]) → dict

Hides specified fields in the JSON schema.

Parameters:

• ui_schema – The original JSON schema

• fields – List of field names to hide

simstack.util.ui_tools.ui_line_vector(ui_schema: Dict[str, Any], field: str) → Dict[str, Any]

Modifies the JSON schema to represent a line vector.

Parameters:

ui_schema – The original JSON schema

Returns:

Modified JSON schema with line vector representation

simstack.util.ui_tools.ui_make_foldable(ui_schema: Dict[str, Any], field: str) → dict

Adds a foldable section to the JSON schema.

Parameters:

• field – The field to be made foldable

• ui_schema – The original ui_schema schema

Returns:

Modified ui_schema with foldable section

simstack.util.ui_tools.ui_make_properties_optional(json_schema: Dict[str, Any], properties: List[str], option_field: str = 'Show More') → dict

Modifies a JSON schema to make specified properties visible only when an option is enabled.

Parameters:

• cls – The class to which the JSON schema belongs

• json_schema – The original JSON schema

• properties – List of property names to make optional

• option_field – Name of the boolean field that controls visibility

Returns:

Modified JSON schema with dependencies

simstack.util.ui_tools.ui_make_title(cls, ui_schema: Dict[str, Any], field: str, title: str) → dict

Adds a title to the JSON schema.

Parameters:

• cls – The class to which the JSON schema belongs

• ui_schema – The original ui_schema schema

• title – Title to be added

Returns:

Modified JSON schema with title

Module contents