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.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

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.

getEventsFileFromPipelineID(pipeline_id: str) → Path

Given a pipeline id, return the path to the .events file for that 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) → 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.

getLogs(pipeline_id: str, transform_id: str, log_level: Optional[str] = None, page: int = 0) → 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.

getLogCounts(pipeline_id: str, transform_id: str) → Dict[str, int]

getJobLogCounts(pipeline_id: str) → Dict[str, int]

getJobInfo(pipeline_id: str) → Dict[str, Any]

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

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)

pipelines_endpoint(params, sessiontoken=None)

pipeline_info_endpoint(params, sessiontoken=None)

transform_info_endpoint(params, sessiontoken=None)

index_endpoint(params)

favicon_endpoints(params)

static start_simple_http_server(local_seamdir: Optional[str] = None, seamwatcher_dir: Path = PosixPath('/home/buildbot/.seamwatcher'), public: bool = False)

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.

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

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