schrodinger.seam.reporting.cascade_collector module¶

Discover filter funnels from completed Seam pipelines.

Filter funnels are discovered automatically from the pipeline proto without hardcoded filter lists.

Discovery mechanism¶

The collector walks the pipeline proto’s canonical transform tree, starting at its declared top-level transforms and following each transform’s ordered child transforms. It looks for transforms whose display_data() contains a filter_funnel key (see FILTER_FUNNEL_DD_KEY). Each key’s value is a FilterFunnelDisplayData that declares the funnel’s filter steps as Beam step labels.

Nested funnels are supported: if a declared step itself has FilterFunnelDisplayData, its sub-steps are inlined into the parent funnel.

Declaring a funnel¶

A funnel composite declares its steps in display_data():

from schrodinger.application.transforms.display_data import (
    FilterFunnelDisplayData, FILTER_FUNNEL_DD_KEY,
    FUNNEL_INPUT_LABEL_DD_KEY,
)

class MyCascade(beam.PTransform):
    _SUBST = 'SubstructureFilter (1)'
    _DOCK = 'Dock'

    def display_data(self):
        return {
            FILTER_FUNNEL_DD_KEY: FilterFunnelDisplayData(
                filters=[
                    'PropertySpaceFilter',
                    self._SUBST,
                    self._DOCK,
                ],
            ),
            FUNNEL_INPUT_LABEL_DD_KEY: 'Inputs',
        }

    def expand(self, pcoll):
        return (pcoll
            | PropertySpaceFilter(property_ranges=self._ranges)
            | self._SUBST >> subst_filter
            | self._DOCK >> dock_transform)

Transforms like PropertySpaceFilter that have their own FilterFunnelDisplayData are automatically expanded in the report.

Titles and filtering¶

The chart title is derived from the Beam label at the call site (the last path component of the transform’s unique_name). Callers pass a titles allowlist to collect_cascades() to control which cascades appear in the report. Use a descriptive Beam label:

pcoll | 'MMP Filter Cascade' >> MyCascade(settings)

class schrodinger.seam.reporting.cascade_collector.CascadeData(title: str, input_label: str, input_count: int | None = None, filters: list[tuple[str, str]] = <factory>)¶

Bases: object

Raw cascade data extracted from a pipeline proto.

title: str¶

input_label: str¶

input_count: int | None = None¶

filters: list[tuple[str, str]]¶: List of (output_key, label) pairs for each filter step.

__init__(title: str, input_label: str, input_count: int | None = None, filters: list[tuple[str, str]] = <factory>) → None¶

schrodinger.seam.reporting.cascade_collector.collect_cascades(snapshot: PipelineSnapshot, titles: set[str]) → list[CascadeData]¶

Collect declared filter funnels in pipeline order.

Declared steps that cannot be resolved to a unique counted output are omitted. A funnel with no resolved steps is omitted as well.

Parameters:

snapshot – pipeline data loaded from a completed Seam run
titles – only return cascades whose title is in this set. Titles are derived from the Beam label at the call site (last component of unique_name).

Returns:

list of CascadeData objects in pipeline tree order