simstack.methods package

Submodules

simstack.methods.archive_files module

class simstack.methods.archive_files.ArchiveConfig(*, call_paths: list[str] = <factory>, archive_resource: str = 'archive_resource', in_memory: bool = False, use_time_window: bool = False, start_date: _datetime | None = None, end_date: _datetime | None = None, include_patterns: list[str] = <factory>, exclude_patterns: list[str] = <factory>, min_size: int = 0, filter_by_resource: bool = False, id: ObjectId = <factory>)[source]

Bases: Model

archive_resource: str = <odmantic.field.FieldProxy object>
call_paths: list[str] = <odmantic.field.FieldProxy object>
async custom_model_dump(**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

end_date: _datetime | None = <odmantic.field.FieldProxy object>
exclude_patterns: list[str] = <odmantic.field.FieldProxy object>
filter_by_resource: bool = <odmantic.field.FieldProxy object>
classmethod from_dict(data: dict, **kwargs) → Any

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

classmethod from_model(model: Model, **kwargs) → Model
id: ObjectId = <odmantic.field.FieldProxy object>
in_memory: bool = <odmantic.field.FieldProxy object>
include_patterns: list[str] = <odmantic.field.FieldProxy object>
classmethod json_schema()

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:

min_size: int = <odmantic.field.FieldProxy object>
model_config = {'arbitrary_types_allowed': False, 'collection': 'archive_config', 'extra': None, 'indexes': None, 'json_schema_extra': None, 'parse_doc_with_default_factories': False, 'str_strip_whitespace': False, 'title': None, 'validate_assignment': True, 'validate_default': True}

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

start_date: _datetime | None = <odmantic.field.FieldProxy object>
classmethod ui_make_title(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

classmethod ui_schema()

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

use_time_window: bool = <odmantic.field.FieldProxy object>
async simstack.methods.archive_files.archive_file(file_stack: FileStack, **kwargs)[source]
simstack.methods.archive_files.archive_files(file_list: FileList, **kwargs)[source]
async simstack.methods.archive_files.archive_node(archive_config: ArchiveConfig, **kwargs)[source]

Archives FileStacks matching the criteria in archive_config and deletes local instances upon success.

Parameters:

  • archive_config (ArchiveConfig) – Configuration for the archive process.

  • **kwargs – Additional arguments.

simstack.methods.archive_files.archive_one_file(file_stack: FileStack, **kwargs)[source]

simstack.methods.fire_and_forget_runner module

class simstack.methods.fire_and_forget_runner.FireAndForgetRunner(node: ~typing.Callable[[...], ~typing.Any], max_concurrency: int | None = None, *, status: ~simstack.core.definitions.TaskStatus = TaskStatus.RETRIEVED, error_message: str | None = None, message: str | None = None, files: ~typing.List[~simstack.models.files.FileStack] = <factory>, info_files: ~typing.List[~simstack.models.files.FileStack] = <factory>, **kwargs)[source]

Bases: NodeRunner

create_tasks(*args: Model)[source]
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

simstack.methods.generate_test module

async simstack.methods.generate_test.async_main()[source]
async simstack.methods.generate_test.generate_test(node_id: str, target_base: Path)[source]

Generate a test case for a given node ID.

simstack.methods.generate_test.generate_test_main()[source]
async simstack.methods.generate_test.load_models(references: List[NamedDataReference]) → Dict[str, Any][source]

Load models from the database based on references.

simstack.methods.generate_test.serialize_models(models: Dict[str, Any]) → str[source]

Serialize models to a JSON string.

simstack.methods.get_file module

simstack.methods.get_file.get_file(file_getter_args: FileGetterArgs) → FileStack[source]

Retrieves a file from a given source and saves it to the specified target path.

This function is responsible for handling the retrieval of files from a FileStack source and storing them to a local resource’s target location. It will update the FileStack

Parameters:

file_getter_args (FileGetterArgs) – The arguments for the file retrieval operation.

Returns:

The modified FileStack with the local resource.

Return type:

FileStack

Raises:

NotImplementedError – If the function is not yet implemented.

simstack.methods.mass_runner module

class simstack.methods.mass_runner.MassRunner(node: ~typing.Callable[[...], ~typing.Any], max_concurrency: int | None = None, *, status: ~simstack.core.definitions.TaskStatus = TaskStatus.RETRIEVED, error_message: str | None = None, message: str | None = None, files: ~typing.List[~simstack.models.files.FileStack] = <factory>, info_files: ~typing.List[~simstack.models.files.FileStack] = <factory>, **kwargs)[source]

Bases: NodeRunner

create_tasks(*args: Model)[source]
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

async recover_orphaned_datasets()[source]

simstack.methods.multiple_executor module

class simstack.methods.multiple_executor.MultipleExecutorInput(*, function_mapping: str, input_mappings: list[str], result_mapping: str, timeout: int = 600, inputs_data: list[list[~odmantic.bson.ObjectId]] = <factory>, id: ~odmantic.bson.ObjectId = <factory>)[source]

Bases: Model, Generic[T]

Represents a MultipleExecutor which handles multiple calculations in parallel and aggregates the results. # TODO store the inputs as ids not in memory

async append(input_models: T | List[T], with_save=True)[source]

Add the inputs for a single function call

async custom_model_dump(**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

async execute(**kwargs) → MultipleExecutorOutput[source]
classmethod from_dict(data: dict, **kwargs) → Any

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

classmethod from_function(func: Callable[[T], R])[source]
classmethod from_model(model: Model, **kwargs) → Model
function_mapping: str = <odmantic.field.FieldProxy object>
id: ObjectId = <odmantic.field.FieldProxy object>
async input_generator() → AsyncGenerator[List[Model], None][source]

Deserialize the input models

input_mappings: list[str] = <odmantic.field.FieldProxy object>
inputs_data: list[list[ObjectId]] = <odmantic.field.FieldProxy object>
classmethod json_schema()

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:

model_config = {'arbitrary_types_allowed': False, 'collection': None, 'extra': None, 'indexes': None, 'json_schema_extra': None, 'parse_doc_with_default_factories': False, 'str_strip_whitespace': False, 'title': None, 'validate_assignment': True, 'validate_default': True}

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

result_mapping: str = <odmantic.field.FieldProxy object>
timeout: int = <odmantic.field.FieldProxy object>
classmethod ui_base_schema()[source]
classmethod ui_make_title(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

classmethod ui_schema()

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

class simstack.methods.multiple_executor.MultipleExecutorOutput(*, executor_input: MultipleExecutorInput, result_mapping: str, result_ids: list[ObjectId | None] = <factory>, id: ObjectId = <factory>)[source]

Bases: Model, Generic[T, R]

Represents the output of a MultipleExecutor.

custom_model_dump() → Dict[str, Any][source]

Retrieve all results and return them as a list of dictionaries.

executor_input: MultipleExecutorInput = <odmantic.field.FieldProxy object>
classmethod from_dict(data: dict, **kwargs) → Any

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

classmethod from_model(model: Model, **kwargs) → Model
id: ObjectId = <odmantic.field.FieldProxy object>
classmethod json_schema()[source]
make_table()[source]

Create a table representation of the results with formatted columns. Returns a dictionary with table data and column definitions.

model_config = {'arbitrary_types_allowed': False, 'collection': None, 'extra': None, 'indexes': None, 'json_schema_extra': None, 'parse_doc_with_default_factories': False, 'str_strip_whitespace': False, 'title': None, 'validate_assignment': True, 'validate_default': True}

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

async result_generator() → AsyncGenerator[R | None, None][source]

Deserialize the result models

result_ids: list[ObjectId | None] = <odmantic.field.FieldProxy object>
result_mapping: str = <odmantic.field.FieldProxy object>
classmethod ui_make_title(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

classmethod ui_schema()

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

class simstack.methods.multiple_executor.PrefixFilter(name='')[source]

Bases: Filter

filter(record)[source]

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

async simstack.methods.multiple_executor.async_adder(arg: BinaryOperationInput, **kwargs) → FloatData[source]
simstack.methods.multiple_executor.get_mappings(func: Callable[[Model], Model])[source]

Decorator to map a function to a model.

async simstack.methods.multiple_executor.main()[source]
async simstack.methods.multiple_executor.multiple_executor(executor: MultipleExecutorInput, **kwargs) → MultipleExecutorOutput[source]
simstack.methods.multiple_executor.multiple_executor_adder(arg1: BinaryOperationInput, arg2: BinaryOperationInput) → BinaryOperationInput[source]

simstack.methods.switch_git module

simstack.methods.switch_git.switch_git(branch: StringData, **kwargs) → bool[source]

simstack.methods.upload_helpers module

async simstack.methods.upload_helpers.archive_upload(file_upload: FileStack, **kwargs)[source]

Asynchronously processes a file upload, extracts files from supported archive formats (.zip, .tar, .tar.gz), and returns a list of extracted files wrapped in FileStack instances. If the file is not an archive, it is added as-is to the output.

Parameters:

  • file_upload (FileStack) – The uploaded file to be processed.

  • **kwargs – Additional keyword arguments. node_runner: An optional object for logging or processing context. If supplied, it will log file paths and extraction details using its info method.

Returns:

A model containing a list of FileStack instances representing the extracted files or the original file if it is not an archive.

Return type:

FileListModel

async simstack.methods.upload_helpers.file_list_upload_all(file_stack: FileStack, **kwargs)[source]
async simstack.methods.upload_helpers.file_list_upload_test(file_list: FileListModel, **kwargs)[source]

Module contents