schrodinger.seam.viz.watcher module

This module provides a webserver that serves a webapp that displays the pipeline graph of a seam pipeline. The webapp is served at http://localhost:8000.

Usage:

$SCHRODINGER/run python3 -m schrodinger.seam.viz.server start $SCHRODINGER/run python3 -m schrodinger.seam.viz.server start path/to/local/seam

exception schrodinger.seam.viz.watcher.InvalidResponseError

Bases: Exception

class schrodinger.seam.viz.watcher.Stage(name: str, full_name: str, start: float, end: Optional[float], status: str, peak_workers: int, peak_mem_mb: Optional[float], num_total_inputs: Optional[int])

Bases: object

One stage of an executed pipeline as it appears in the snapshot.

Variables:

  • name – stage_id from the runner (e.g. "stage_03"); doubles as full_name since stages have no hierarchy.

  • full_name – same as name for stage rows.

  • start – stage start time as epoch seconds.

  • end – stage end time as epoch seconds, or None while running.

  • status – "running", "completed", or "failed".

  • peak_workers – max concurrent workers seen during the stage.

  • peak_mem_mb – max worker peak memory across workers in MB, or None if the stage finished without worker metrics.

  • num_total_inputs – total inputs the stage was sized to process, or None if not reported.

name: str
full_name: str
start: float
end: Optional[float]
status: str
peak_workers: int
peak_mem_mb: Optional[float]
num_total_inputs: Optional[int]
__init__(name: str, full_name: str, start: float, end: Optional[float], status: str, peak_workers: int, peak_mem_mb: Optional[float], num_total_inputs: Optional[int]) → None
class schrodinger.seam.viz.watcher.PipelineMetadata(name: str, id: str, status: str, user: str, host: str, subhost: Optional[str], max_workers: Optional[int], job_id: str, schrodinger_version: str, started_at: Optional[float], ended_at: Optional[float], elapsed_s: float, cpu_hours: float, license_requirements: dict)

Bases: object

Header-level fields about the run as a whole.

Variables:

  • name – human display name (job name if set, else the pipeline id).

  • id – pipeline id used to reference this run in URLs.

  • status – "Running", "Completed", or "Failed".

  • user – user who launched the run.

  • host – driver host (where the SeamRunner process lives).

  • subhost – subhost name from job control, or None.

  • max_workers – subhost worker cap from job control, or None.

  • job_id – job-control job id.

  • schrodinger_version – Schrodinger build the runner ran in.

  • started_at – pipeline start as epoch seconds.

  • ended_at – pipeline end as epoch seconds, or None while running.

  • elapsed_s – pipeline duration in seconds (0 while running).

  • cpu_hours – total CPU hours across all stages.

  • license_requirements – {license_name: tokens_required}.

name: str
id: str
status: str
user: str
host: str
subhost: Optional[str]
max_workers: Optional[int]
job_id: str
schrodinger_version: str
started_at: Optional[float]
ended_at: Optional[float]
elapsed_s: float
cpu_hours: float
license_requirements: dict
__init__(name: str, id: str, status: str, user: str, host: str, subhost: Optional[str], max_workers: Optional[int], job_id: str, schrodinger_version: str, started_at: Optional[float], ended_at: Optional[float], elapsed_s: float, cpu_hours: float, license_requirements: dict) → None
class schrodinger.seam.viz.watcher.PipelineSnapshot(pipeline: PipelineMetadata, stages: List[Stage], transform_tree: Optional[Dict[str, Any]], workers: List[Dict[str, float]], driver_mem: List[Dict[str, float]], worker_mem: List[Dict[str, Any]], seam_dir_size: List[Dict[str, float]], licenses: List[Dict[str, float]], license_types: List[str])

Bases: object

The full payload that drives the watcher page (one snapshot of one pipeline execution; the refresh button re-fetches it).

Variables:

  • pipeline – header metadata; see PipelineMetadata.

  • stages – per-stage timing/status/peaks; see Stage.

  • transform_tree – nested {name: str, full_name: str, is_stage: bool, children: list, ...} walking the user pipeline. Stage leaves (is_stage: True) carry the same fields as Stage. Stays a recursive dict because typing the recursion costs more than it gains.

  • workers – [{"t": float (epoch seconds), "queued": int, "running": int}] worker-count timeline derived from per-worker lifecycle log events; one entry per transition.

  • driver_mem – [{"t": float (epoch seconds), "mb": float}] driver-process memory timeline.

  • worker_mem – [{"worker_id": str, "points": [{"t": float (epoch seconds), "mb": float}, ...]}, ...] — one entry per worker that reported memory metrics, each carrying its own sample stream.

  • seam_dir_size – [{"t": float (epoch seconds), "gb": float}] seam-dir size at each stage end, derived from SeamDirMetrics.size_per_stage.

  • licenses – [{"t": float, <license_name>: int, ...}] license-token usage timeline derived from per-stage license requirements scaled by the running-worker count for each stage.

  • license_types – ordered list of license names appearing in licenses — used by the frontend to lay out stacked bands.

pipeline: PipelineMetadata
stages: List[Stage]
transform_tree: Optional[Dict[str, Any]]
workers: List[Dict[str, float]]
driver_mem: List[Dict[str, float]]
worker_mem: List[Dict[str, Any]]
seam_dir_size: List[Dict[str, float]]
licenses: List[Dict[str, float]]
license_types: List[str]
__init__(pipeline: PipelineMetadata, stages: List[Stage], transform_tree: Optional[Dict[str, Any]], workers: List[Dict[str, float]], driver_mem: List[Dict[str, float]], worker_mem: List[Dict[str, Any]], seam_dir_size: List[Dict[str, float]], licenses: List[Dict[str, float]], license_types: List[str]) → None
class schrodinger.seam.viz.watcher.WatcherDirReader(local_seamdir: Optional[Path] = None, seamwatcher_dir: Path = PosixPath('/home/buildbot/.seamwatcher'))

Bases: object

Read the state of pipeline(s) from the seamwatcher directory to provide and transform information required by the webserver.

__init__(local_seamdir: Optional[Path] = None, seamwatcher_dir: Path = PosixPath('/home/buildbot/.seamwatcher'))
Parameters:

  • local_seamdir – the path to the local seam directory. If set, then all methods will read from the local seam directory instead of the seamwatcher directory.

  • seamwatcher_dir – the path to the seamwatcher directory

getUserMetricsFile(pipeline_id: str) → Optional[Path]

Given a pipeline id, return the path to the user metrics file for that pipeline. If the pipeline id is LOCAL, then return None.

getWorkerMetrics(pipeline_id: str) → Optional[ResourceMetrics]

Given a pipeline id, return a dictionary of worker metrics for that pipeline.

getDriverMemoryMetrics(pipeline_id: str) → Optional[DriverMemoryMetrics]

Given a pipeline id, return driver memory metrics for that pipeline.

Returns:

DriverMemoryMetrics or None if no driver memory data exists.

getSeamDirMetrics(pipeline_id: str) → Optional[SeamDirMetrics]

Given a pipeline id, return per-stage peak seam-dir sizes.

Returns:

SeamDirMetrics or None if no stage metrics exist yet (e.g. pipeline still in stage 1).

getDot(pipeline_id: str, view_type: str, expanded_transforms: List, color_by: str) → str

Given a pipeline id and view options, return the dot representation of the pipeline as a string.

getLeafTransforms(pipeline_id: str) → List[str]

Given a pipeline id, return a list of the leaf transforms in the pipeline.

getSchrodingerVersion(pipeline_id: str) → str

Given a pipeline id, return the Schrodinger version used to run the pipeline.

getDotFromEventsFile(events_file, viewType, expandedTransforms: list, colorBy: str)

Given a path to a .events file, return the dot representation of the pipeline as a string.

getJobLogs(pipeline_id: str, log_level: Optional[str] = None, page: int = 0, search_term: str = '') → tuple[List[dict], bool]

Given a pipeline id and transform id, return the log messages for that transform as a list of dictionaries. Each dictionary contains the keys ‘timestamp’, ‘level’, and ‘message’.

Parameters:

  • pipeline_id – the id of the pipeline

  • transform_id – the id of the transform

  • log_level – the minimum log level to return. One of ‘ERROR’, ‘WARNING’, ‘INFO’, ‘DEBUG’. If None, all log levels are returned.

  • page – the page of log messages to return. Each page contains up to 500 log messages. Page 0 is the oldest log messages.

  • search_term – optional search term to filter log messages.

getLogs(pipeline_id: str, transform_id: str, log_level: Optional[str] = None, page: int = 0, search_term: str = '') → tuple[List[dict], bool, str]

Given a pipeline id and transform id, return the log messages for that transform as a list of dictionaries. Each dictionary contains the keys ‘timestamp’, ‘level’, and ‘message’.

Parameters:

  • pipeline_id – the id of the pipeline

  • transform_id – the id of the transform

  • log_level – the minimum log level to return. One of ‘ERROR’, ‘WARNING’, ‘INFO’, ‘DEBUG’. If None, all log levels are returned.

  • page – the page of log messages to return. Each page contains up to 500 log messages. Page 0 is the oldest log messages.

  • search_term – optional search term to filter log messages.

getLogCounts(pipeline_id: str, transform_id: str) → Dict[str, int]
getJobLogCounts(pipeline_id: str) → Dict[str, int]
getJobInfo(pipeline_id: str) → Dict[str, Any]
getPipelineSummary(pipeline_id: str) → Dict[str, Any]

Return a summary dict for a pipeline, suitable for the landing page table.

getPipelineOptions(pipeline_id: str) → Dict[str, Any]
getWorkers(pipeline_id: str) → Dict[str, Any]

Aggregate worker lifecycle, bundle, and autoscale events into {"workers": [...], "autoscale": [...]} for the Workers tab. Each worker record carries the static per-worker license profile for its stage (no per-worker license events exist).

getPipelineSnapshot(pipeline_id: str) → PipelineSnapshot

Build the snapshot that drives the watcher page. The endpoint adapter converts the returned dataclass to a dict for the wire.

class schrodinger.seam.viz.watcher.ResponseType

Bases: Enum

TEXT = 'text/plain'
JSON = 'application/json'
HTML = 'text/html'
ZIP = 'application/zip'
SVG = 'image/svg+xml'
class schrodinger.seam.viz.watcher.EndpointInfo(path: str, request_schema: Dict[str, Any], response_type: ResponseType)

Bases: object

path: str
request_schema: Dict[str, Any]
response_type: ResponseType
__init__(path: str, request_schema: Dict[str, Any], response_type: ResponseType) → None
exception schrodinger.seam.viz.watcher.InvalidRequestError

Bases: Exception

schrodinger.seam.viz.watcher.endpoint(endpoint_name: str, request_schema: dict = None, response_type: ResponseType = ResponseType.JSON, accepts_sessiontoken: bool = False)

Decorator to add an endpoint to the server.

Parameters:

  • endpoint_name – the name of the endpoint

  • request_schema – the schema for the request parameters

  • response_type – the type of the response

  • accepts_sessiontoken – whether the endpoint accepts a session token

Returns:

the decorated function

class schrodinger.seam.viz.watcher.WatcherWebServer(local_seamdir: Optional[str] = None, seamwatcher_dir: Path = PosixPath('/home/buildbot/.seamwatcher'))

Bases: _BaseServer

SERVER_ROOT = '/'
__init__(local_seamdir: Optional[str] = None, seamwatcher_dir: Path = PosixPath('/home/buildbot/.seamwatcher'))
sessiontokenHasAccess(sessiontoken: Optional[str], pipeline_id: str) → bool

Check if a session token has access to a pipeline. By default, all session tokens have access to all pipelines. Subclasses can override this method to implement access control.

Parameters:

  • sessiontoken – the session token to check

  • pipeline_id – the pipeline id to check

dot_endpoint(params, sessiontoken=None)
logs_endpoint(params, sessiontoken=None)
joblogs_endpoint(params, sessiontoken=None)
jobloglevelcounts_endpoint(params, sessiontoken=None)
loglevelcounts_endpoint(params, sessiontoken=None)
workers_endpoint(params, sessiontoken=None)
diagnostics_endpoint(params, sessiontoken=None)
pipelines_endpoint(params, sessiontoken=None)
pipelines_summary_endpoint(params, sessiontoken=None)
pipeline_snapshot_endpoint(params, sessiontoken=None)
pipeline_info_endpoint(params, sessiontoken=None)
transform_info_endpoint(params, sessiontoken=None)
index_endpoint(params)
favicon_endpoints(params)
classmethod start_simple_http_server(local_seamdir: Optional[str] = None, seamwatcher_dir: Path = PosixPath('/home/buildbot/.seamwatcher'), public: bool = False, port: int = 8443, server_root: str = '/')

Start a simple HTTP server that serves the watcher webapp. Only intended for local use, not for production servers that might require more security.

Parameters:

  • local_seamdir – the path to a local seam directory. If set, the server will make the local seamdir available to the webapp.

  • seamwatcher_dir – Path to the seamwatcher directory

  • public – Whether to bind to all interfaces (True) or localhost only (False)

  • port – Port number to bind the server to

class schrodinger.seam.viz.watcher.ThreadedTCPServer(server_address, RequestHandlerClass, bind_and_activate=True)

Bases: ThreadingMixIn, HTTPServer

This server class uses a mixin to create a new thread for each request. This is a simple way to enable concurrency without writing complex threading logic.

daemon_threads = True
schrodinger.seam.viz.watcher.get_parser() → ArgumentParser
exception schrodinger.seam.viz.watcher.GCloudNotInstalledError(message=None)

Bases: Exception

Custom exception raised when the ‘gcloud’ command is not found on the system. Provides detailed instructions on how to install the Google Cloud CLI.

__init__(message=None)
schrodinger.seam.viz.watcher.main(args: Optional[List[str]] = None)