schrodinger.seam.reporting.output_metric_collector module

Collect workflow output metrics from completed Seam pipelines.

Metrics are declared as display data on existing workflow transforms. A declaration associates a nested dictionary path with the output of the declaring transform or one of its child transforms. The transform_label value is a Beam transform label relative to the transform containing the declaration:

from schrodinger.seam.reporting.output_metric_display_data import (
    OUTPUT_METRICS_DD_KEY,
    OutputMetric,
    OutputMetricsDisplayData,
)

class MyWorkflow(beam.PTransform):
    _GENERATE = 'Generate'
    _SAMPLE = 'RandomSample'

    def display_data(self):
        return {
            OUTPUT_METRICS_DD_KEY: OutputMetricsDisplayData(metrics=[
                OutputMetric(
                    path=('FINAL', 'NUM_GENERATED'),
                    transform_label=self._GENERATE,
                ),
                OutputMetric(
                    path=('FINAL', 'NUM_SAMPLED'),
                    transform_label=self._SAMPLE,
                ),
            ])
        }

    def expand(self, pcoll):
        generated = pcoll | self._GENERATE >> Generate()
        return generated | self._SAMPLE >> RandomSample(...)

For a multi-output transform without an untagged Beam default, set OutputMetric.output to its output tag. For a semantic count that is not the target transform’s final output, the target transform can expose a stable named source:

from schrodinger.seam.reporting.output_metric_display_data import (
    COUNT_SOURCES_DD_KEY,
    CountSource,
    CountSourcesDisplayData,
)

def display_data(self):
    return {
        COUNT_SOURCES_DD_KEY: CountSourcesDisplayData(sources=[
            CountSource(name='unique', transform_label='Deduplicate'),
        ])
    }

The owning workflow selects that source with OutputMetric(transform_label='RandomSample', count_source='unique', ...). This lets the source transform change its internal implementation without exposing its child-transform labels to workflow callers.

After a successful run, load its recorded pipeline and Seam output counts once, then collect the declared nested dictionary:

from schrodinger.seam.reporting.pipeline_snapshot import PipelineSnapshot

snapshot = PipelineSnapshot.fromSeamDir(seam_dir)
metrics = collect_output_metrics(snapshot)

The collector does not write or combine job-level metrics.json data. Callers own any workflow-specific combination of independently collected dictionaries and writing of compatibility artifacts.

Resolution is strict: missing transforms, underspecified multi-output transforms, missing or ambiguous counted producers, unknown named sources, and duplicate or conflicting dictionary paths raise ValueError.

schrodinger.seam.reporting.output_metric_collector.collect_output_metrics(snapshot: PipelineSnapshot) → dict[str, int | dict[str, int | dict[str, ForwardRef('MetricValue')]]]

Collect declared output metrics from a completed pipeline snapshot.

Parameters:

snapshot – pipeline data loaded from a completed Seam run

Raises:

ValueError – if a metric declaration cannot be resolved