schrodinger.seam.viz.inspect module¶

class schrodinger.seam.viz.inspect.TransformStatus¶

Bases: Enum

PENDING = 'Pending'¶

RUNNING = 'Running'¶

COMPLETED = 'Completed'¶

PARTIALLY_COMPLETED = 'Partially Completed'¶

FAILED = 'Failed'¶

class schrodinger.seam.viz.inspect.PipelineDisplayInfo(start_time: str, end_time: Optional[str], duration: str, cpu_time: str, status: str)¶

Bases: object

start_time: str¶

end_time: Optional[str]¶

duration: str¶

cpu_time: str¶

status: str¶

__init__(start_time: str, end_time: Optional[str], duration: str, cpu_time: str, status: str) → None¶

class schrodinger.seam.viz.inspect.PipelineDisplayInfoV2(*, start_time: str, end_time: Optional[str] = None, duration: str, cpu_time: str, status: PipelineStatus)¶

Bases: BaseModel

start_time: str¶

end_time: Optional[str]¶

duration: str¶

cpu_time: str¶

status: PipelineStatus¶

class schrodinger.seam.viz.inspect.PCollInfo(*, count: str, producing_transform: str, element_type: str, URN: str = 'seam:display_data:pcollinfo')¶

Bases: BaseModel

count: str¶

producing_transform: str¶

element_type: str¶

URN: str¶

class schrodinger.seam.viz.inspect.PCollectionsInfo(*, pcolls: Dict[str, PCollInfo], URN: str = 'seam:display_data:pcollectionsinfo')¶

Bases: BaseModel

pcolls: Dict[str, PCollInfo]¶

URN: str¶

class schrodinger.seam.viz.inspect.LicenseReqInfo(*, license_name: str, num_tokens: str, URN: str = 'seam:display_data:licensereqinfo')¶

Bases: BaseModel

license_name: str¶

num_tokens: str¶

URN: str¶

class schrodinger.seam.viz.inspect.LicenseRequirements(*, requirements: List[LicenseReqInfo], URN: str = 'seam:display_data:licenserequirements')¶

Bases: BaseModel

requirements: List[LicenseReqInfo]¶

URN: str¶

class schrodinger.seam.viz.inspect.StepInfo(*, local_name: str, input_pcolls: PCollectionsInfo, output_pcolls: PCollectionsInfo, cputime: str, licenses: LicenseRequirements, display_data: Optional[Dict[str, str]] = None)¶

Bases: BaseModel

local_name: str¶

input_pcolls: PCollectionsInfo¶

output_pcolls: PCollectionsInfo¶

cputime: str¶

licenses: LicenseRequirements¶

display_data: Optional[Dict[str, str]]¶

class schrodinger.seam.viz.inspect.TransformDisplayInfo(local_name: str, input_pcolls: str, output_pcolls: str, cputime: str, licenses: str, display_data: Optional[Dict[str, str]] = None)¶

Bases: object

A dataclass to hold information about a transform.

Variables:

local_name – The unqualified label of the transform.
input_pcolls – A JSON string representation of the input PCollection IDs, their counts, the producing transform, and the element type.
output_pcolls – A JSON string representation of the output PCollection IDs, their counts, the producing transform, and the element type.
cputime – The CPU time spent in the transform.
licenses – A JSON string representation of the license requirements of the transform.

local_name: str¶

input_pcolls: str¶

output_pcolls: str¶

cputime: str¶

licenses: str¶

display_data: Optional[Dict[str, str]] = None¶

__init__(local_name: str, input_pcolls: str, output_pcolls: str, cputime: str, licenses: str, display_data: Optional[Dict[str, str]] = None) → None¶

schrodinger.seam.viz.inspect.get_pipeline_info(seam_events_path: Union[Path, str]) → PipelineDisplayInfo¶: Infer the start time, end time, duration, and status of a pipeline from a seam events file.

schrodinger.seam.viz.inspect.get_pcoll_type(pipeline: Pipeline, pcoll_id: str) → str¶: Given a pipeline proto and a PCollection ID, returns a string representation of the type hint of a PCollection.

schrodinger.seam.viz.inspect.get_licenses(pipeline: Pipeline, transform_id: str) → dict[str, str]¶

Get the license requirements of a transform in a pipeline.

Returns a key-value mapping of license names to the number of tokens required as strings for printing. e.g. {“LIGPREP_MAIN”: “2”}

schrodinger.seam.viz.inspect.get_transform_info_from_pipeline(user_pipeline, transform_id, metrics)¶

schrodinger.seam.viz.inspect.get_transform_info(seam_events_path: Union[Path, str], transform_id: str) → TransformDisplayInfo¶: Infer the input PCollection IDs, output PCollection IDs, CPU time, and licenses of a transform from a seam events file.

schrodinger.seam.viz.inspect.get_transform_info_v2(seam_events_path: Union[Path, str], transform_id: str) → StepInfo¶

schrodinger.seam.viz.inspect.get_leaf_transforms(pipeline: Pipeline) → List[str]¶: Get the leaf transform IDs in a pipeline. A leaf transform is a transform that does not have any child transforms.

schrodinger.seam.viz.inspect.get_consuming_transforms(pipeline: Pipeline, pcoll_id: str) → List[str]¶: Get the Transform IDs that consume a given PCollection ID in a pipeline.

schrodinger.seam.viz.inspect.get_producing_transform(pipeline: Pipeline, pcoll_id: str) → Optional[str]¶: Get the Transform ID that produces a given PCollection ID in a pipeline. If no such transform exists, return None.

schrodinger.seam.viz.inspect.get_pipeline_start_time(seam_events_path: Union[Path, str]) → float¶: Infer the start time of a pipeline from seam events file.

schrodinger.seam.viz.inspect.get_raw_user_labels(pipeline_proto: Pipeline) → Dict[str, str]¶: Given a user pipeline proto (i.e. a pipeline proto that has not been optimized), returns a mapping from user labels to stage ids (aka stage ids) that they correspond to.

schrodinger.seam.viz.inspect.get_annotated_user_labels(transform_proto: PTransform) → list¶

Given a transform proto of a stage (i.e. a transform proto that has been in an optimized pipeline), returns a list of all the user labels that have been annotated to that stage.

For example, in the following pipeline:

with beam.Pipeline() as p:
    _ = (p
        | 'A' >> beam.Create([1, 2, 3])
        | 'B' >> beam.Map(lambda x: x + 1)
        | 'C' >> beam.Map(lambda x: x + 1)
        | 'D' >> beam.Map(lambda x: x + 1)
    )

If ‘B’, ‘C’, and ‘D’ are all fused into the same stage, then resulting transform proto for that stage will return [‘B’, ‘C’, ‘D’].

schrodinger.seam.viz.inspect.get_user_label_to_stage_mapping(optimized_pipeline: Pipeline) → Dict[str, set[str]]¶: Given an optimized pipeline proto, returns a mapping from user labels to the stages that they are in.

schrodinger.seam.viz.inspect.get_transform_unique_name_to_stage_mapping(optimized_pipeline: Pipeline) → Dict[str, set[str]]¶: Given an optimized pipeline proto, returns a mapping from transform unique names to the stage that they are in.

schrodinger.seam.viz.inspect.get_inferred_statuses(xform_to_stage_mapping: Dict[str, set], running: List[str], complete: List[str], failed: List[str]) → Dict[str, str]¶

Given a mapping from user labels to the stages that they are in, and a list of running stages and completed stages, returns a mapping from user labels to statuses.

For example, if the transforms “A”, “B”, and “C” are in stages “stage_1”, and “stage_2” represented by the following mapping:

{
    "A": {"stage_1"},
    "B": {"stage_1"},
    "C": {"stage_2"}
}

And “stage_1” is complete and “stage_2” is running, then the following mapping will be returned:

{
    "A": "Completed",
    "B": "Completed",
    "C": "Running"
}