populse_mia.data_manager.data_history_inspect

This module provides utilities for tracking, retrieving, and reconstructing the processing history of data files within the Mia framework. It enables users to trace the lineage of data products, including the bricks (processing steps) and intermediate files involved in their creation.

The module supports:

• History Reconstruction: Build pipelines or graphs representing the processing history of a file, including all upstream dependencies.

• Brick-Process Conversion: Convert brick database entries into lightweight “fake” processes for pipeline visualization.

• Ancestor Tracking: Identify direct and indirect ancestors of a file, including handling of temporary values and ambiguous execution times.

• Data Entry Validation: Check if filenames correspond to valid database entries or temporary placeholders.

Functions

brick_to_process(brick, project)	Convert a brick database entry into a 'fake process'.
data_history_pipeline(filename, project)	Retrieves the complete processing history of a file in the database, formatted as a "fake pipeline".
data_in_value(value, filename, project)	Determine if the specified filename is present within the given value.
find_procs_with_output(procs, filename, project)	Identify processes in the given list that have the specified filename as part of their outputs.
get_data_history(filename, project)	Retrieves the processing history for a given data file, based on get_data_history_processes().
get_data_history_bricks(filename, project)	Retrieves the complete "useful" history of a file in the database as a set of processing bricks.
get_data_history_processes(filename, project)	Retrieves the complete useful processing history of a file in the database.
get_direct_proc_ancestors(filename, project, ...)	Retrieve processing bricks referenced in the direct filename history.
get_filenames_in_value(value, project[, ...])	Extract filenames from a nested structure of lists, tuples, and dictionaries.
get_history_brick_process(brick_id, project)	Retrieve a brick from the database using its UUID and return it as a ProtoProcess instance.
get_proc_ancestors_via_tmp(proc, project, procs)	Retrieve upstream processes connected via a temporary value ("<temp>").
is_data_entry(filename, project[, allow_temp])	Check whether a filename corresponds to a valid project data entry.

Classes

ProtoProcess([brick])

A lightweight convenience class that stores a brick database entry along with additional usage information.

class populse_mia.data_manager.data_history_inspect.ProtoProcess(brick=None)

Bases: object

A lightweight convenience class that stores a brick database entry along with additional usage information.

This class encapsulates a brick database entry and tracks whether it has been used, providing a simple interface for managing brick-related data.

__init__(brick=None)

Initializes the ProtoProcess instance with a brick database entry.

Parameters:

brick (dict or None) – The brick database entry to store. Defaults to None.

populse_mia.data_manager.data_history_inspect.brick_to_process(brick, project)

Convert a brick database entry into a ‘fake process’.

This function transforms a brick database entry into a Process instance that represents its parameters and values. The process gets a name, uuid, and exec_time from the brick. This “fake process” cannot perform actual processing but serves as a representation of the brick’s traits and values.

Parameters:

• brick (dict | str) – The brick database entry to convert. If a string is provided, it is treated as the brick’s unique ID, and the corresponding brick document is retrieved from the project’s database.

• project (Project) – The project object providing access to the database and its documents.

Returns:

A Process instance representing the brick’s parameters and values. Returns None if the brick is not found.

Return type:

Process | None

populse_mia.data_manager.data_history_inspect.data_history_pipeline(filename, project)

Retrieves the complete processing history of a file in the database, formatted as a “fake pipeline”.

The generated pipeline consists of unspecialized (fake) processes, each representing a processing step with all parameters of type Any. The pipeline includes connections and traces all upstream ancestors of the file, capturing the entire processing path leading to the latest version of the file.

If the file was modified multiple times, the pipeline reflects only the relevant processing steps that contributed to the final output. Orphaned processing steps from overwritten versions are omitted.

Parameters:

• filename (str) – The name of the file whose processing history is being retrieved.

• project (Project) – The project object containing the database and relevant details.

Returns:

A Pipeline object representing the processing history, or None if no relevant history is found.

Return type:

Pipeline | None

populse_mia.data_manager.data_history_inspect.data_in_value(value, filename, project)

Determine if the specified filename is present within the given value.

This function recursively searches through the value, which can be a string, list, tuple, or dictionary, to check if it contains the specified filename. The filename can be a special placeholder <temp> or a short filename, which is a relative path within the project’s database data directory.

Parameters:

• value (str | list | tuple | dict) –

  The data structure to search. It can be:

  • A string representing a file path.

  • A list or tuple containing multiple file paths.

  • A dictionary where file paths are stored as values.

• filename (str) –

  The filename to search for. It can be:

  • The special placeholder “<temp>” indicating a temporary value.

  • A relative file path to the project database data directory.

• project (Project) – The project object containing the project’s folder path as an attribute (project.folder).

Returns:

True if the filename is found in the value, False otherwise.

Return type:

bool

populse_mia.data_manager.data_history_inspect.find_procs_with_output(procs, filename, project)

Identify processes in the given list that have the specified filename as part of their outputs.

This function searches through a list of processes to determine which ones have the specified filename in their output values. The results are organized by execution time.

Parameters:

• procs (iterable of ProtoProcess) – A collection of ProtoProcess instances to search through.

• filename (str) – The filename to search for within the processes outputs.

• project (Project) – An instance of the project, used to access the database folder.

Returns:

A dictionary where keys are execution times and values are lists of tuples. Each tuple contains a process and the parameter name associated with the filename. Format: {exec_time: [(process, param_name), …]}.

Return type:

dict

populse_mia.data_manager.data_history_inspect.get_data_history(filename, project)

Retrieves the processing history for a given data file, based on get_data_history_processes().

The returned dictionary contains:

• parent_files: A set of filenames representing data (direct or indirect) used to produce the given file.

• processes: A set of UUIDs of processing bricks that contributed to the file’s creation.

Parameters:

• filename (str) – The name of the file whose processing history is being retrieved.

• project (Project) – The project object containing the database and relevant details.

Returns:

A dictionary with the following keys:

• ”processes”: A set of UUIDs representing the processing bricks involved.

• ”parent_files”: A set of filenames that were used to produce the data.

Return type:

dict

populse_mia.data_manager.data_history_inspect.get_data_history_bricks(filename, project)

Retrieves the complete “useful” history of a file in the database as a set of processing bricks.

This function is a filtered version of get_data_history_processes(), similar to data_history_pipeline(), but instead of constructing a pipeline, it returns only the set of brick elements that were actually used in the relevant processing history of the file.

Parameters:

• filename (str) – The name of the file whose processing history is being retrieved.

• project (Project) – The project object containing the database and relevant details.

Returns:

A set of brick elements representing the “useful” processing steps that contributed to the final version of the given data file.

Return type:

set

populse_mia.data_manager.data_history_inspect.get_data_history_processes(filename, project)

Retrieves the complete useful processing history of a file in the database.

This function returns:

• A dictionary of processes (ProtoProcess instances), where keys are process UUIDs.

• A set of links between these processes, forming the processing graph.

Unlike data_history_pipeline(), which converts the history into a Pipeline, this function returns a lower-level representation of the processing history. Some processes encountered during history traversal may not be relevant to the final history; they are identified by their used attribute, which is set to True for processes that contribute to the current file history.

Processing bricks that are not used (for example, from previous runs where the output file was overwritten) may either be absent from the history or returned with used = False.

Parameters:

• filename (str) – The name of the file whose processing history is being retrieved.

• project (Project) – The Project containing the database and related information.

Returns:

A tuple with:

• procs (dict): Mapping of process UUIDs to ProtoProcess instances.

• links (set): Processing graph connections represented as (src_protoprocess, src_plug_name, dst_protoprocess, dst_plug_name) tuples. External connections are represented with None as src_protoprocess or dst_protoprocess.

Return type:

tuple[dict, set]

populse_mia.data_manager.data_history_inspect.get_direct_proc_ancestors(filename, project, procs, before_exec_time=None, only_latest=True, org_proc=None)

Retrieve processing bricks referenced in the direct filename history.

This function identifies the most recent processing steps that generated the given filename. If multiple processes share the same execution time, they are all retained to account for ambiguity. The function also allows filtering by execution time and excluding a specified originating process.

Parameters:

• filename (str) – The data filename to inspect.

• project (Project) – The project instance used to access the database.

• procs (dict) – Dictionary mapping process UUIDs to ProtoProcess instances. This dictionary is updated with newly retrieved processes.

• before_exec_time (datetime | None) – If specified, only processing bricks executed before this time are considered.

• only_latest (bool) – If True (default), keeps only the latest processes found in the history. If before_exec_time is specified, retains only the latest before that time.

• org_proc (ProtoProcess) – The originating process, which is excluded from execution time filtering but included in the ancestor list.

Returns:

A dictionary mapping brick UUIDs to ProtoProcess instances.

Return type:

dict

populse_mia.data_manager.data_history_inspect.get_filenames_in_value(value, project, allow_temp=True)

Extract filenames from a nested structure of lists, tuples, and dictionaries.

This function parses the given value, which can be a nested combination of lists, tuples, and dictionaries, to retrieve all filenames referenced within it. Only filenames that are valid database entries or the special <temp> value (if allow_temp is True) are retained. Other filenames are considered read-only static data and are not included in the results.

Parameters:

• value (str | list | tuple | dict) – The value to parse. It can be a single string, a list, a tuple, a dictionary, or any nested combination of these types.

• project (Project) – The project object providing access to the database.

• allow_temp (bool) – If True, includes the temporary filename <temp> in the results. Defaults to True.

Returns:

A set of filenames that are valid database entries or the temporary filename <temp> (if allowed).

Return type:

set[str]

populse_mia.data_manager.data_history_inspect.get_history_brick_process(brick_id, project, before_exec_time=None)

Retrieve a brick from the database using its UUID and return it as a ProtoProcess instance.

This function fetches a brick from the database using its unique identifier (UUID). It returns the brick as a ProtoProcess instance if the brick has been executed (its execution status is “Done”) and, if specified, its execution time is not later than before_exec_time. If the brick does not meet these criteria or is not found in the database, the function returns None.

Parameters:

• brick_id (str) – The unique identifier (UUID) of the brick to retrieve.

• project (Project) – The project object providing access to the database.

• before_exec_time –

  An execution time filter. If provided, bricks

  executed after this timestamp are discarded.

  type before_exec_time:

  str | None

Returns:

A ProtoProcess instance representing the brick if it meets the criteria; otherwise, None.

Return type:

ProtoProcess | None

populse_mia.data_manager.data_history_inspect.get_proc_ancestors_via_tmp(proc, project, procs)

Retrieve upstream processes connected via a temporary value (“<temp>”).

This function is intended for internal use within get_data_history_processes and data_history_pipeline. It attempts to identify upstream processes connected to the given process (proc) through a temporary filename.

The function first searches the direct history of the process’s output files. If no matching process is found, it searches the entire database of bricks, which may be slower for large databases. Matching is based on the temporary filename and processing time, which can be error-prone.

Parameters:

• proc (ProtoProcess) – The process whose ancestors need to be determined.

• project (Project) – The project object providing access to the database and other necessary functionalities for processing.

• procs (dict) – A dictionary of processes, where keys are process IDs and values are ProtoProcess instances.

Returns:

A tuple with:

• new_procs (dict): Mapping of process UUIDs to ProtoProcess instances.

• links (set): Processing graph connections represented as (src_protoprocess, src_plug_name, dst_protoprocess, dst_plug_name) tuples. External connections are represented with None as src_protoprocess or dst_protoprocess.

Return type:

tuple[dict, set]

Contains:

Inner functions:

• _get_tmp_param: Identifies a process parameter associated with a temporary value.

populse_mia.data_manager.data_history_inspect.is_data_entry(filename, project, allow_temp=True)

Check whether a filename corresponds to a valid project data entry.

A valid data entry is a file whose path belongs to the project directory and whose relative path exists as a document in the project’s current database collection. The special value "<temp>" is also considered valid when allow_temp is True.

The function returns the path relative to the project folder for valid database entries. It returns "<temp>" unchanged for temporary entries, and None for invalid or unknown files.

Parameters:

• filename (str) – Absolute file path or the special value "<temp>" to check.

• project (Project) – Project instance providing access to the project folder and database.

• allow_temp (bool) – Whether the special value "<temp>" should be accepted as a valid entry. Defaults to True.

Returns:

The path relative to the project folder if filename is a valid database entry, "<temp>" for a temporary entry, or None if the file is not a valid entry.

Return type:

str | None