schrodinger.seam.reporting.output_metric_display_data module

Beam display-data models for finished-run output metrics.

Instances are attached to existing transforms and interpreted after the run by schrodinger.seam.reporting.output_metric_collector.

class schrodinger.seam.reporting.output_metric_display_data.CountSource(*, name: str, transform_label: Optional[str] = None, output: Optional[str] = None)

Bases: BaseModel

Names a transform output whose element count can be reported.

Parameters:

  • name – stable name referenced by OutputMetric.count_source

  • transform_label – Beam transform label relative to the transform containing the enclosing CountSourcesDisplayData; when omitted, use that transform

  • output – output tag to select when the source transform has multiple outputs; when omitted, use its sole or untagged default output

name: str
transform_label: Optional[str]
output: Optional[str]
model_config: ClassVar[ConfigDict] = {'frozen': True}

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

class schrodinger.seam.reporting.output_metric_display_data.CountSourcesDisplayData(*, sources: tuple[CountSource, ...])

Bases: BaseModel, HasDisplayData

Exposes uniquely named transform outputs as count sources.

Parameters:

sources – named transform outputs available as count sources

sources: tuple[CountSource, ...]
model_config: ClassVar[ConfigDict] = {'frozen': True}

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

classmethod fromJson(json_str: str) → Self
display_data() → dict[str, org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.DisplayData]

Returns the display data associated to a pipeline component.

It should be reimplemented in pipeline components that wish to have static display data.

Returns:

Dict[str, Any]: A dictionary containing key:value pairs. The value might be an integer, float or string value; a DisplayDataItem for values that have more data (e.g. short value, label, url); or a HasDisplayData instance that has more display data that should be picked up. For example:

{
  'key1': 'string_value',
  'key2': 1234,
  'key3': 3.14159265,
  'key4': DisplayDataItem('apache.org', url='http://apache.org'),
  'key5': subComponent
}
class schrodinger.seam.reporting.output_metric_display_data.OutputMetric(*, path: tuple[typing.Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])], ...], transform_label: Optional[str] = None, output: Optional[str] = None, count_source: Optional[str] = None)

Bases: BaseModel

Associates a nested metric path with a finished-pipeline count.

Parameters:

  • path – nested dictionary keys where the count is collected

  • transform_label – Beam transform label relative to the transform containing this metric; when omitted, use that transform

  • output – output tag to select when the target transform has multiple outputs; when omitted, use its sole or untagged default output

  • count_source – named count source exposed by the target transform; mutually exclusive with output

path: tuple[typing.Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])], ...]
transform_label: Optional[str]
output: Optional[str]
count_source: Optional[str]
model_config: ClassVar[ConfigDict] = {'frozen': True}

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

schrodinger.seam.reporting.output_metric_display_data.grouped_output_metrics(group_path: tuple[str, ...], transform_labels: Mapping[str, str]) → tuple[OutputMetric, ...]

Create output metrics that share a nested dictionary path.

For each entry in transform_labels, the mapping key is appended to group_path and the mapping value identifies the transform whose output will be counted. Transform labels are relative to the transform exposing the resulting OutputMetricsDisplayData.

Use OutputMetric directly when a metric needs an output tag or a named count source.

Parameters:

  • group_path – common prefix of the generated metric paths

  • transform_labels – mapping from final metric-path keys to transform labels

class schrodinger.seam.reporting.output_metric_display_data.OutputMetricsDisplayData(*, metrics: tuple[OutputMetric, ...])

Bases: BaseModel, HasDisplayData

Exposes finished-pipeline output counts as domain metrics.

Parameters:

metrics – associations between transform outputs and nested metric paths

metrics: tuple[OutputMetric, ...]
model_config: ClassVar[ConfigDict] = {'frozen': True}

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

classmethod fromJson(json_str: str) → Self
display_data() → dict[str, org.apache.beam.model.pipeline.v1.beam_runner_api_pb2.DisplayData]

Returns the display data associated to a pipeline component.

It should be reimplemented in pipeline components that wish to have static display data.

Returns:

Dict[str, Any]: A dictionary containing key:value pairs. The value might be an integer, float or string value; a DisplayDataItem for values that have more data (e.g. short value, label, url); or a HasDisplayData instance that has more display data that should be picked up. For example:

{
  'key1': 'string_value',
  'key2': 1234,
  'key3': 3.14159265,
  'key4': DisplayDataItem('apache.org', url='http://apache.org'),
  'key5': subComponent
}