schrodinger.tasks.provenance module

Utilities for tracking the provenance of values as they are passed through one or more functions.

class schrodinger.tasks.provenance.TrackedValue(value: Any, source_id: Any)

Bases: object

The TrackedValue is the basic unit of exchange between track_provenance- decorated functions. The return value of a decoratedfunction is always a TrackedValue, which is suitable for passing in as the first argument of another decorated function.

When creating the initial TrackedValue, set source_id to any identifier (e.g. entry_id, corporate_id, index, etc.)

Variables:

  • history – A list recording the history of that generated this value.

  • failure_info – Information about the exception raised while passing the value through a function, if any.

  • metadata – A map of additional information about the value.

__init__(value: Any, source_id: Any)
Parameters:

  • value – The value to track

  • source_id – An ID for the source of the value.

makeChild(value: Any, record: Any, failure_info: schrodinger.tasks.tasks.FailureInfo | None = None) → Self
schrodinger.tasks.provenance.track_provenance(func: Callable[[...], Any]) → Callable[[...], TrackedValue]

A decorator to optionally record the provenance of a function’s return value.

If a non-TrackedValue is passed into the decorated function, the decorator will have no effect and the function will retain its original behavior.

If a TrackedValue is passed in as the first argument to the decorated function, it will return a new TrackedValue that contains a record of the value’s provenance and/or failure information.

When the tracked output of one decorated function is passed into a second decorated function (and so on), the resulting TrackedValue will contain information about the identity of the initial input, metadata being tracked and a record of the chain of operations that led to the final result.

If at any point one of the decorated functions raises an exception, failure info will be recorded in the TrackedValue output, and any subsequent functions will simply return the input TrackedValue.

Example:: >>> from schrodinger import adapter >>> from schrodinger.tasks import provenance >>> >>> @provenance.track_provenance … def get_atom_count(st, ignore_hydrogens=False): … countable_atoms = st.atom … if ignore_hydrogens: … countable_atoms = [ … atom for atom in countable_atoms if atom.atomic_number != 1 … ] … return len(countable_atoms) >>> >>> st = adapter.smiles_to_3d_structure(‘CCC’) >>> result = get_atom_count(st, ignore_hydrogens=True) # default behavior >>> result 3 >>> tracked_val = provenance.TrackedValue(st, ‘smiles_CCC’) >>> result = get_atom_count(tracked_val, ignore_hydrogens=True) >>> result.metadata[‘ignore_hydrogens’] = True >>> result.value 3 >>> result.source_id ‘smiles_CCC’ >>> result.history [‘get_atom_count’] >>> result.failure_info is None True >>> result.metadata {‘ignore_hydrogens’: True}

Alternatively, given a pre-existing get_atom_count function, one could create a tracked version:

tracked_get_atom_count = track_provenance(get_atom_count)