#!/usr/bin/env python3 # Copyright (c) 2023, Pacific Biosciences of California, Inc. ## # All rights reserved. ## # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are met: # * Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # * Redistributions in binary form must reproduce the above copyright # notice, this list of conditions and the following disclaimer in the # documentation and/or other materials provided with the distribution. # * Neither the name of Pacific Biosciences nor the names of its # contributors may be used to endorse or promote products derived from # this software without specific prior written permission. ## # NO EXPRESS OR IMPLIED LICENSES TO ANY PARTY'S PATENT RIGHTS ARE GRANTED BY # THIS LICENSE. THIS SOFTWARE IS PROVIDED BY PACIFIC BIOSCIENCES AND ITS # CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT # NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A # PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL PACIFIC BIOSCIENCES OR # ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, # EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, # PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; # OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, # WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR # OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF # ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. """ SMRT Link REST API reference client implementation, for version 12.0.0 or newer. This is written to be self-contained without any dependencies on internal PacBio libraries, and can be copied, modified, and redistributed without limitation (see module comments for license terms). The SmrtLinkClient does not cover the entire API, but the code is intended to be easily extended and translated into other languages. For simplicity and readability, the returned objects are weakly typed lists and dicts containing basic Python types (str, int, float, bool, None). The Swagger documentation in the SMRT Link GUI online help (https://servername:8243/sl/docs/services) provides a comprehensive listing of endpoints and data models. Software requirements: Python >= 3.9; 'requests' module Example module usage: - Creating a client object, including authentication:: client = SmrtLinkClient.connect("localhost", user="pbuser", password="XXX") - Get all analysis jobs associated with a run:: run = client.get_run(run_uuid) collections = client.get_run_collections(run_uuid) jobs = [] for collection in collections: jobs.extend(client.get_dataset_jobs(collection["ccsId"])) - Import a HiFi dataset XML, and fetch a list of all datasets that were loaded (including any demultiplexed children):: import_job = client.create_import_dataset_job(xml_file) finished_job = client.poll_for_successful_job(import_job["id"]) datasets = client.get_consensusreadsets(jobId=import_job["id"]) - Retrieve barcoded sample metrics for a run collection and dump to CSV:: reports = client.get_run_collection_reports(run_uuid, collection_uuid) for r in reports: if r["reportTypeId"].split(".")[-1] == "report_barcode": report_uuid = r["dataStoreFile"]["uuid"] report = client.load_datastore_report_file(report_uuid) bc_table = report["tables"][0] headers = [c["header"] for c in bc_table["columns"]] col_data = [c["values"] for c in bc_table["columns"]] table_rows = [[c[i] for c in col_data] for i in len(col_data[0])] write_csv(csv_file_name, headers, table_rows) - Retrieve loading metrics for a run collection and return as a dict:: reports = client.get_run_collection_reports(run_uuid, collection_uuid) for r in reports: if "loading" in r["reportTypeId"]: report_uuid = r["dataStoreFile"]["uuid"] report = client.load_datastore_report_file(report_uuid) print({attr["id"]:attr["value"] for attr in report["attributes"]}) - Combine a sample split across multiple cells: DS_TYPE = "PacBio.DataSet.ConsensusReadSet" datasets = client.get_consensusreadsets(bioSampleName="MySample1234") job = client.create_merge_datasets_job([d["uuid"] for d in datasets]) job = client.poll_for_successful_job(job["id"]) datastore = client.get_job_datastore(job["id"]) merged_datasets = [f for f in datastore if f["fileTypeId"] == DS_TYPE] # alternately: merged_datasets = client.get_consensusreadsets(jobId=job["id"]) - Find the Run QC reports associated with an analysis job:: entry_points = client.get_job_entry_points(job_id) movie_names = set([]) for entry_point in entry_points: if entry_point["datasetType"] == "PacBio.DataSet.ConsensusReadSet": dataset = client.get_consensusreadset(entry_point["datasetUUID"]) movie_names.append(dataset["metadataContextId"]) qc_reports = [] for movie_name in movie_names: runs = client.get_runs(movieName=movie_name) if len(runs) == 1: collections = client.get_run_collection(runs[0]["unique_id"]) for collection in collections: if collection["context"] == movie_name: reports = client.get_collection_reports(run_id, collection["unique_id"]) qc_reports.append(reports) - Poll every 10 minutes until a collection is complete, then launch a HiFi Mapping job using the official PacBio hg38 reference, and poll until it completes successfully:: collection = client.get_run_collection(run_id, collection_id) while True: dataset = client.get_dataset_search(collection["ccsId"]) if dataset: break else: time.sleep(600) job = client.create_analysis_job({ "name": "My Mapping Job", "pipelineId": "cromwell.workflows.pb_align_ccs", "entryPoints": [ { "entryId": "eid_ccs", "datasetId": collection["ccsId"], "fileTypeId": "PacBio.DataSet.ConsensusReadSet" }, { "entryId": "eid_ref_dataset", "datasetId": "ba3866bf-2aba-7c99-0570-0d6709174e4a", "fileTypeId": "PacBio.DataSet.ReferenceSet" } ], "taskOptions": [], "workflowOptions": [] }) job = client.poll_for_successful_job(job["id"]) """ from abc import ABC, abstractmethod import urllib.parse import argparse import logging import json import time import os import sys # This is the only non-standard dependency import requests __all__ = [ "SmrtLinkClient", "add_smrtlink_server_args", "get_smrtlink_client_from_args", ] log = logging.getLogger(__name__) class Constants: API_PORT = 8243 H_CT_AUTH = "application/x-www-form-urlencoded" H_CT_JSON = "application/json" # SSL is good and we should not disable it by default DEFAULT_VERIFY = True def refresh_on_401(f): """ Method decorator to trigger a token refresh when an HTTP 401 error is received. """ def wrapper(self, *args, **kwds): try: return f(self, *args, **kwds) except requests.exceptions.HTTPError as e: if e.response.status_code == 401: self.refresh() return f(self, *args, **kwds) else: raise return wrapper def _disable_insecure_warning(): """ Workaround to silence SSL warnings when the invoker has explicitly requested insecure mode. """ from urllib3.exceptions import ProtocolError, InsecureRequestWarning # pylint: disable=import-error # To disable the ssl cert check warning import urllib3 import warnings warnings.warn("You are running with some SSL security features disabled (verify=False). Please note that this is considered risky and exposes you to remote attacks that can steal your authorization credentials or sensitive data.", InsecureRequestWarning) urllib3.disable_warnings( InsecureRequestWarning) # pylint: disable=no-member class RESTClient(ABC): """ Base class for interacting with any REST API that communicates primarily in JSON. """ PROTOCOL = "http" def __init__(self, host, port, verify=Constants.DEFAULT_VERIFY): self.host = host self.port = port self._verify = verify @abstractmethod def refresh(self): ... @property def headers(self): return {"Content-Type": Constants.H_CT_JSON} @property def base_url(self): return f"{self.PROTOCOL}://{self.host}:{self.port}" def to_url(self, path): """Convert an API method path to the full server URL""" return f"{self.base_url}{path}" def _get_headers(self, other_headers={}): headers = dict(self.headers) if other_headers: headers.update(other_headers) return headers @refresh_on_401 def _http_get(self, path, params=None, headers={}): if isinstance(params, dict): if len(params) == 0: params = None else: # get rid of queryParam=None elements params = {k: v for k, v in params.items() if v is not None} url = self.to_url(path) log.info(f"Method: GET {path}") log.debug(f"Full URL: {url}") response = requests.get(url, params=params, headers=self._get_headers(headers), verify=self._verify) log.debug(response) response.raise_for_status() return response @refresh_on_401 def _http_post(self, path, data, headers={}): url = self.to_url(path) log.info(f"Method: POST {path} {data}") log.debug(f"Full URL: {url}") response = requests.post(url, data=json.dumps(data), headers=self._get_headers(headers), verify=self._verify) log.debug(response) response.raise_for_status() return response @refresh_on_401 def _http_put(self, path, data, headers={}): url = self.to_url(path) log.info(f"Method: PUT {path} {data}") log.debug(f"Full URL: {url}") response = requests.put(url, data=json.dumps(data), headers=self._get_headers(headers), verify=self._verify) log.debug(response) response.raise_for_status() return response @refresh_on_401 def _http_delete(self, path, headers={}): url = self.to_url(path) log.info(f"Method: DELETE {path}") log.debug(f"Full URL: {url}") response = requests.delete(url, headers=self._get_headers(headers), verify=self._verify) log.debug(response) response.raise_for_status() return response @refresh_on_401 def _http_options(self, path, headers={}): url = self.to_url(path) log.info(f"Method: OPTIONS {url}") log.debug(f"Full URL: {url}") response = requests.options(url, headers=self._get_headers(headers), verify=self._verify) log.debug(response) response.raise_for_status() return response def get(self, path, params=None, headers={}): """Generic JSON GET method handler""" return self._http_get(path, params, headers).json() def post(self, path, data, headers={}): """Generic JSON POST method handler""" return self._http_post(path, data, headers).json() def put(self, path, data, headers={}): """Generic JSON PUT method handler""" return self._http_put(path, data, headers).json() def delete(self, path, headers={}): """Generic JSON DELETE method handler""" return self._http_delete(path, headers).json() def options(self, path, headers={}): """ OPTIONS handler, used only for getting CORS settings in ReactJS. Since the response body is empty, the return value is the response headers (as a dict). """ return self._http_options(path, headers).headers() def execute_call(self, method, path, data, headers): """Execute any supported JSON-returning HTTP call by name""" if method == "GET": return self.get(path, headers=headers) elif method == "POST": return self.post(path, data=data, headers=headers) elif method == "PUT": return self.put(path, data=data, headers=headers) elif method == "DELETE": return self.get(path, headers=headers) elif method == "OPTIONS": return self.options(path, headers=headers) else: raise ValueError(f"Method '{method}' not supported") class AuthenticatedClient(RESTClient): """ Base class for REST clients that require authorization via the Oauth2 token interface. """ def __init__(self, host, port, username, password, verify=Constants.DEFAULT_VERIFY): super(AuthenticatedClient, self).__init__(host, port, verify) self._user = username self._oauth2 = self.get_authorization_token(username, password) @property def auth_token(self): return self._oauth2["access_token"] @property def refresh_token(self): return self._oauth2["refresh_token"] @property def headers(self): """Dict of default HTTP headers for all endpoints""" return { "Content-Type": Constants.H_CT_JSON, "Authorization": f"Bearer {self.auth_token}" } @abstractmethod def get_authorization_token(self, username, password): ... class SmrtLinkClient(AuthenticatedClient): """ Class for executing methods on the secure (authenticated) SMRT Link REST API, via API gateway """ PROTOCOL = "https" JOBS_PATH = "/smrt-link/job-manager/jobs" def __init__(self, *args, **kwds): if not kwds.get("verify", Constants.DEFAULT_VERIFY): _disable_insecure_warning() super(SmrtLinkClient, self).__init__(*args, **kwds) @staticmethod def connect(host, username, password, verify=Constants.DEFAULT_VERIFY): """ Convenience method for instantiating a client using the default API port 8243 """ return SmrtLinkClient(host=host, port=Constants.API_PORT, username=username, password=password, verify=verify) @property def headers(self): return { "Content-Type": Constants.H_CT_JSON, "Authorization": f"Bearer {self.auth_token}", "X-User-ID": self._user } def to_url(self, path): return f"{self.base_url}/SMRTLink/2.0.0{path}" def get_authorization_token(self, username, password): """ Request an Oauth2 authorization token from the SMRT Link API server, which is actually a proxy to Keycloak. This token will enable access to all API methods that are allowed for the role of the authorized user. """ auth_d = dict(username=username, password=password, grant_type="password") resp = requests.post(f"{self.base_url}/token", data=auth_d, headers={"Content-Type": Constants.H_CT_AUTH}, verify=self._verify) resp.raise_for_status() t = resp.json() log.info("Access token: {}...".format(t["access_token"][0:40])) log.debug("Access token: {}".format(t["access_token"])) return t def refresh(self): """ Attempt to refresh the authorization token, using the refresh token obtained in a previous request. """ log.info("Requesting new access token using refresh token") auth_d = dict(grant_type="refresh_token", refresh_token=self.refresh_token) resp = requests.post(self.to_url("/token"), data=auth_d, headers={"Content-Type": Constants.H_CT_AUTH}, verify=self._verify) resp.raise_for_status() t = resp.json() log.info("Access token: {}...".format(t["access_token"][0:40])) log.debug("Access token: {}".format(t["access_token"])) self._oauth2 = t return t # ----------------------------------------------------------------- # ADMINISTRATION def get_status(self): """Get status of server backend""" return self.get("/status") def set_system_config_param(self, key, value): """ Set a SMRT Link system configuration parameter, such as the instrument user password """ body = dict(key=key, value=value) return self.post("/smrt-link/admin-config", body) def get_swagger_api(self): """Fetch the Swagger API definition for the server""" return self.get("/smrt-link/swagger") def get_software_manifests(self): """Get a list of software components and versions""" return self.get("/smrt-link/manifests") def get_software_manifest(self, component_id): """Retrieve version information for a specific component""" return self.get(f"/smrt-link/manifests/{component_id}") def get_instrument_connections(self): """ Get a list of Revio instrument connections. """ return self.get("/smrt-link/instrument-config/connections") def create_instrument_connection(self, host, secret_key, name="Revio Instrument"): """ Create a new Revio instrument connection in SMRT Link. The server will automatically perform a full handshake and configure ICS. """ body = { "host": host, "serial": "unknown", # NOTE the server gets this from ICS "name": name, "credentials": secret_key } return self.post("/smrt-link/instrument-config/connections", body) def update_instrument_connection(self, instrument_id, update_d): """Update an existing Revio instrument connection""" path = f"/smrt-link/instrument-config/connections/{instrument_id}" return self.put(path, update_d) def connect_instrument(self, instrument_id): """Activate an existing Revio instrument connection""" return self.update_instrument_connection(instrument_id, dict(isConnected=True)) def delete_instrument_connection(self, id_name_or_serial): """Delete an instrument connection record""" return self.delete(f"/smrt-link/instrument-config/connections/{id_name_or_serial}") def get_instrument_states(self): """ Return a list of instrument states, complex objects that include configuration details and run progress. Connected instruments should send state updates once per minute. """ return self.get("/smrt-link/instruments") def get_instrument_state(self, serial): """ Return the last recorded state for a specific instrument by serial number. """ return self.get(f"/smrt-link/instruments/{serial}") def delete_instrument_state(self, serial): """ Remove an instrument from the Instruments status page in SMRT Link (but not from the Instrument Settings page) """ return self.delete(f"/smrt-link/instruments/{serial}") # ----------------------------------------------------------------- # RUNS def get_runs(self, **search_params): """ Get a list of all PacBio instrument runs, with optional search parameters. Partial list of supported search parameters: name (partial matches supported) reserved (boolean, true means selected on instrument) instrumentType (Revio, Sequel2e, or Sequel2) chipType (8mChip or 25mChip) collectionUuid (retrieve the run for a specific collection) movieName """ return self.get("/smrt-link/runs", dict(search_params)) def get_run(self, run_id): """Retrieve a PacBio instrument run description by UUID""" return self.get(f"/smrt-link/runs/{run_id}") def get_run_xml(self, run_id): """ Retrieve the XML data model for a PacBio instrument run """ return self._http_get(f"/smrt-link/runs/{run_id}/datamodel").text def get_run_collections(self, run_id): """Retrieve a list of collections/samples for a run""" return self.get(f"/smrt-link/runs/{run_id}/collections") def get_run_collection(self, run_id, collection_id): """Retrieve metadata for a single collection in a run""" return self.get(f"/smrt-link/runs/{run_id}/collections/{collection_id}") def get_run_from_collection_id(self, collection_id): """ Convenience method wrapping get_runs(), for retrieving a run based on collection UUID alone. Returns None if no matching run is found. """ runs = self.get_runs(collectionUuid=collection_id) return None if len(runs) == 0 else runs[0] def get_run_collection_reports(self, run_id, collection_id): """ Get all reports associated with a run collection Introduced in SMRT Link 13.0 """ return self.get(f"/smrt-link/runs/{run_id}/collections/{collection_id}/reports") def get_run_collection_barcodes(self, run_id, collection_id): """Get a list of barcoded samples associated with a run collection""" return self.get(f"/smrt-link/runs/{run_id}/collections/{collection_id}/barcodes") def get_run_collection_hifi_reads(self, run_id, collection_id): """ Retrieve the HiFi dataset that is the primary output of a PacBio instrument run. """ collection = self.get_run_collection(run_id, collection_id) return self.get_consensusreadset(collection["ccsId"]) def get_run_collection_hifi_reads_barcoded_datasets(self, run_id, collection_id, barcode_name=None, biosample_name=None): """ Retrieve the demultiplexed "child" datasets for a PacBio instrument run, optionally filtering by barcode name (e.g. 'bc2001--bc2001') or biosample name. """ collection = self.get_run_collection(run_id, collection_id) return self.get_barcoded_child_datasets(collection["ccsId"], barcode_name=barcode_name, biosample_name=biosample_name) def get_run_reports(self, run_id): """ Get all collection-level reports associated with a run. Introduced in SMRT Link 13.0 """ return self.get(f"/smrt-link/runs/{run_id}/reports") def get_run_design(self, run_id): """Return the run design JSON object used by the SMRT Link GUI""" return self.get(f"/smrt-link/run-design/{run_id}") def import_run_design_csv(self, csv_file): """ Import a Run CSV definition and return the run-design data model. This is the officially supported interface for creating a Run Design programatically. """ csv_d = {"content": open(csv_file, "rt").read()} return self.post("/smrt-link/import-run-design", csv_d) def delete_run(self, run_id): """Delete a PacBio run description by UUID""" return self.delete(f"/smrt-link/runs/{run_id}") def import_run_xml(self, xml_file): """ Post a Run XML directly to the API. This is not officially supported as an integration mechanism, but is useful for transferring Run QC results between servers. """ return self.post("/smrt-link/runs", {"dataModel": open(xml_file).read()}) def update_run_xml(self, xml_file, run_id, is_reserved=None): """ Update a Run data model XML. This endpoint is what the Revio and Sequel II/IIe instruments use to communicate run progress, and to mark a run as "reserved" by a particular instrument. It can be used as a workaround for updating the status of incomplete runs after manual XML edits. """ opts_d = {"dataModel": open(xml_file).read()} if is_reserved is not None: opts_d["reserved"] = is_reserved return self.post(f"/smrt-link/runs/{run_id}", opts_d) # ----------------------------------------------------------------- # CHEMISTRY BUNDLE def get_active_bundle_metadata(self, bundle_type): """ Return the metadata for the current version of a bundle """ return self.get(f"/smrt-link/bundles/{bundle_type}/active") def get_chemistry_bundle_metadata(self): """ Return the metadata for the current version of the chemistry-pb bundle used by Sample Setup, Run Design, and the instrument software """ return self.get_active_bundle_metadata("chemistry-pb") def get_active_bundle_file(self, bundle_type, relative_path): """ Retrieve the contents of a file in a bundle """ relpath = urllib.parse.quote(relative_path) path = f"/smrt-link/bundles/{bundle_type}/active/files/{relpath}" return self._http_get(path).text def get_chemistry_bundle_file(self, relative_path): """ Retrieve the contents of a file in the chemistry bundle. The list of consumables is in 'definitions/PacBioAutomationConstraints.xml'; settings for Run Design applications are in 'RunDesignDefaults.json'. """ return self.get_active_bundle_file("chemistry-pb", relative_path) # ----------------------------------------------------------------- # MISC SERVICES def download_datastore_file(self, file_uuid): path = f"/smrt-link/datastore-files/{file_uuid}/download" return self._http_get(path).content def load_datastore_report_file(self, file_uuid): """ Convenience wrapper for downloading a datastore report.json file in memory and and converting to Python objects. """ return json.loads(self.download_datastore_file(file_uuid)) def download_file_resource(self, file_uuid, resource_name): """ Retrieve another file (usually a PNG file) referenced by a datastore file (usually a Report) and return the raw data """ path = f"/smrt-link/datastore-files/{file_uuid}/resources" return self._http_get(path, params={"relpath": resource_name}).content # ----------------------------------------------------------------- # DATASETS def _get_datasets_by_type(self, dataset_type, **query_args): return self.get(f"/smrt-link/datasets/{dataset_type}", params=dict(query_args)) def _get_dataset_by_type_and_id(self, dataset_type, dataset_id): return self.get(f"/smrt-link/datasets/{dataset_type}/{dataset_id}") def _get_dataset_resources_by_type_and_id(self, dataset_type, dataset_id, resource_type): return self.get(f"/smrt-link/datasets/{dataset_type}/{dataset_id}/{resource_type}") def get_consensusreadsets(self, **query_args): """ Retrieve a list of HiFi datasets, with optional search parameters. Partial list of supported search terms: name bioSampleName wellSampleName metadataContextId (movie name) String searches are always case-insensitive. Most of the non-timestamp string fields in the data model are searchable with partial strings by adding the prefix 'like:' to the search term, thus: client.get_consensusreadsets(bioSampleName="like:HG002") The refixes 'not:' (inequality), 'unlike:', 'start:', and 'end:' are also recognized. For numerical fields, 'not:', 'lt:', 'lte:', 'gt:', 'gte:' are supported, plus 'range:{start},{end}'. """ return self._get_datasets_by_type("ccsreads", **query_args) def get_consensusreadsets_by_movie(self, movie_name): """ Retrieve a list of HiFi datasets for a unique movie name (AKA 'context' or 'metadataContextId'), such as 'm84001_230601_123456' """ return self.get_consensusreadsets(metadataContextId=movie_name) def get_barcoded_child_datasets(self, parent_dataset_id, barcode_name=None, biosample_name=None): """Get a list of demultiplexed children (if any) of a HiFi dataset""" return self.get_consensusreadsets(parentUuid=parent_dataset_id, dnaBarcodeName=barcode_name, bioSampleName=biosample_name) def get_subreadsets(self, **query_args): """ Retrieve a list of CLR/subread datasets, with optional search parameters. (DEPRECATED) """ return self._get_datasets_by_type("subreads", **query_args) def get_referencesets(self, **query_args): """Get a list of ReferenceSet datasets""" return self._get_datasets_by_type("references", **query_args) def get_barcodesets(self, **query_args): """Get a list of BarcodeSet datasets (including MAS-Seq adapters and Iso-Seq primers)""" return self._get_datasets_by_type("barcodes", **query_args) def get_consensusreadset(self, dataset_id): """Get a HiFi dataset by UUID or integer ID""" return self._get_dataset_by_type_and_id("ccsreads", dataset_id) def get_subreadset(self, dataset_id): """Get a CLR (subread) dataset by UUID or integer ID (DEPRECATED)""" return self._get_dataset_by_type_and_id("subreads", dataset_id) def get_referenceset(self, dataset_id): """GET /smrt-link/datasets/references/{dataset_id}""" return self._get_dataset_by_type_and_id("references", dataset_id) def get_barcodeset(self, dataset_id): """GET /smrt-link/datasets/barcodes/{dataset_id}""" return self._get_dataset_by_type_and_id("barcodes", dataset_id) def get_consensusreadset_reports(self, dataset_id): """Get a list of reports associated with a HiFi dataset""" return self._get_dataset_resources_by_type_and_id("ccsreads", dataset_id, "reports") def get_barcodeset_contents(self, dataset_id): """ Retrieve the entire contents of a BarcodeSet dataset, as a raw FASTA string (not JSON!) """ path = f"/smrt-link/datasets/barcodes/{dataset_id}/contents" return self._http_get(path).text def get_barcodeset_record_names(self, dataset_id): """Retrieve a list of barcode/primer/adapter names in a BarcodeSet""" return self._get_dataset_resources_by_type_and_id("barcodes", dataset_id, "record-names") def get_dataset_metadata(self, dataset_id): """Retrieve a type-independent dataset metadata object""" return self._get_dataset_by_type_and_id("meta", dataset_id) def get_dataset_jobs(self, dataset_id): """Get a list of analysis jobs that used the specified dataset as input""" return self._get_dataset_resources_by_type_and_id("meta", dataset_id, "jobs") def get_dataset_search(self, dataset_id): """ Retrieve a single dataset if it is present in the database, or None if it is missing, without triggering an HTTP 404 error in the latter case. """ result_d = self.get(f"/smrt-link/datasets/search/{dataset_id}") if not result_d: # empty dict is the "not found" response return None return result_d # ----------------------------------------------------------------- # JOBS def _get_jobs_by_type(self, job_type, params=None): return self.get(f"{self.JOBS_PATH}/{job_type}", params=params) def _post_job_by_type(self, job_type, opts_d): return self.post(f"{self.JOBS_PATH}/{job_type}", opts_d) def _get_job_by_type_and_id(self, job_type, job_id): return self.get(f"{self.JOBS_PATH}/{job_type}/{job_id}") def _get_job_resources_by_type_and_id(self, job_type, job_id, resource_type): return self.get(f"{self.JOBS_PATH}/{job_type}/{job_id}/{resource_type}") def _get_job_resource_by_type_and_id(self, job_type, job_id, resource_type, resource_id): return self.get(f"{self.JOBS_PATH}/{job_type}/{job_id}/{resource_type}/{resource_id}") # NOTE our API routing is deliberately very permissive due to # limitations of the underlying toolkits, so we often use the # 'analysis' endpoints for GETting individual job attributes, # regardless of job type def get_job(self, job_id): """ Retrieve a job of any type by integer ID or UUID. """ return self._get_job_by_type_and_id("analysis", job_id) def get_job_reports(self, job_id): """ Get a list of reports generated by a job. The UUIDs of these objects can be used to retrieve the full report content using get_job_report() """ return self._get_job_resources_by_type_and_id("analysis", job_id, "reports") def get_job_report(self, job_id, report_uuid): """ Retrieve the Report data for a specific report output by a job. """ return self._get_job_resource_by_type_and_id("analysis", job_id, "reports", report_uuid) def download_job_report_resource(self, job_id, report_uuid, resource_name): """ Retrieve a plot (usually a PNG file) referenced by a Report object, and return the raw image data """ path = f"{self.JOBS_PATH}/analysis/{job_id}/reports/{report_uuid}/resources" return self._http_get(path, params={"relpath": resource_name}).content def get_job_datastore(self, job_id): """ Get a list of all exposed output files generated by a job """ return self._get_job_resources_by_type_and_id("analysis", job_id, "datastore") def get_job_entry_points(self, job_id): """Get the dataset UUIDs used to create the job""" return self._get_job_resources_by_type_and_id("analysis", job_id, "entry-points") def get_job_datasets(self, job_id): """Alias for get_job_entry_points(job_id)""" return self.get_job_entry_points(job_id) def get_job_options(self, job_id): """Get the options model used to create the job""" return self._get_job_resources_by_type_and_id("analysis", job_id, "options") def download_job_datastore_file(self, job_id, file_id): """ Get the raw content for a specific file in the job datastore. Note that this is effectively redundant with download_datastore_file. """ path = f"{self.JOBS_PATH}/analysis/{job_id}/datastore/{file_id}/download" return self._http_get(path).content def get_analysis_jobs(self, **search_params): """ Get a list of standalone analysis jobs, with optional search/filter parameters. This is equivalent to the "flat view" in the SMRT Analysis job catalog. """ return self._get_jobs_by_type("analysis", search_params) def get_analysis_jobs_by_state(self, state): """ Get a list of analysis jobs in a specified state. Supported states: CREATED SUBMITTED RUNNING SUCCESSFUL FAILED TERMINATED ABORTED """ return self.get_analysis_jobs(state=state) def get_analysis_jobs_by_parent(self, multi_job_id): """ Get all children of the multi-analysis job whose ID is specified """ return self.get_analysis_jobs(parentMultiJobId=multi_job_id) def get_smrt_analysis_nested_jobs(self, **search_params): """ Get the 'nested' view of analysis jobs displayed by default in the SMRT Analysis jobs catalog, where children of a multi-analysis job are hidden at the top level """ return self.get("/smrt-link/job-manager/analysis-jobs", params=search_params) def create_analysis_job(self, opts_d): """ Submit a SMRT Analysis job to be run as soon as possible. Requires that all input datasets have already been imported. Job options schema: pipelineId: string such as 'cromwell.workflows.pb_align_ccs' name: job name string entryPoints: list of dataset entry points taskOptions: list of workflow task options projectId: int or null presetId: string or null Entry point model: entryId: pre-set identifier, can be any of 'eid_ccs', 'eid_barcode', 'eid_ref_dataset', 'eid_barcode_2', or 'eid_subread' fileTypeId: dataset MetaType, from the top-level XML tag datasetId: dataset UniqueID (UUID) Task/workflow option model: optionId: string ID such as 'mapping_min_length' value: string, float, int, bool, or occasionally null optionTypeId: type of 'value' field """ # NOTE workflowOptions is still technically required by the REST API # but will become optional in future releases if "workflowOptions" not in opts_d: opts_d["workflowOptions"] = [] return self._post_job_by_type("analysis", opts_d) def terminate_analysis_job(self, job_id): """Immediately terminate a running analysis job.""" return self.post(f"{self.JOBS_PATH}/analysis/{job_id}/terminate", {}) def get_import_dataset_jobs(self, **search_params): """Get a list of import-dataset jobs""" return self._get_jobs_by_type("import-dataset", search_params) def _post_job_with_path_opt(self, job_type, opts_d_or_path, key="path"): opts_d = opts_d_or_path if isinstance(opts_d_or_path, str): opts_d = {key: opts_d_or_path} return self._post_job_by_type(job_type, opts_d) def create_import_dataset_job(self, opts_d_or_path): """ Submit an import-dataset job, using the path to the dataset XML. :param opts_d_or_path: either a dictionary containing at least the 'path' field, or an actual path string """ return self._post_job_with_path_opt("import-dataset", opts_d_or_path) def create_import_datasets_zip_job(self, opts_d): """ Submit an import-datasets-zip job, using the path to the zipfile exported by another SMRT Link server. :param opts_d: a dictionary containing at least the 'zipFile' field """ return self._post_job_by_type("import-datasets-zip", opts_d) def create_import_collection_job(self, opts_d_or_path): """ Submit an import-collection job, using the path to the instrument file moviename.reports.zip. In SMRT Link Lite this is used to populate the Run QC and Data Management pages, without direct access to the XML files. :param opts_d_or_path: either a dictionary containing at least the 'path' field, or an actual path string """ return self._post_job_with_path_opt("import-collection", opts_d_or_path) def create_merge_datasets_job(self, opts_d_or_dataset_ids): """ Submit a merge-datasets job, using two or more unique IDs for ConsensusReadSet datasets. :param opts_d_or_dataset_ids: either a dictionary containing at least the 'ids' field, or a list of integers or UUID strings """ opts_d = opts_d_or_dataset_ids if isinstance(opts_d_or_dataset_ids, list): opts_d = {"ids": opts_d_or_dataset_ids} return self._post_job_by_type("merge-datasets", opts_d) def get_pipelines(self, public_only=True): """ Retrieve a list of SMRT Analysis workflow interface descriptions """ def _is_private(tags): hidden_tags = {"dev", "internal", "alpha", "obsolete"} return len(set(tags).intersection(hidden_tags)) > 0 pipelines = self.get("/smrt-link/resolved-pipeline-templates") if public_only: return [p for p in pipelines if not _is_private(p["tags"])] else: return pipelines def get_pipeline(self, pipeline_id): """ Retrieve a SMRT Analysis workflow interface description by ID, such as 'pb_align_ccs' or 'cromwell.workflows.pb_align_ccs' """ if not pipeline_id.startswith("cromwell.workflows"): pipeline_id = f"cromwell.workflows.{pipeline_id}" return self.get(f"/smrt-link/resolved-pipeline-templates/{pipeline_id}") def poll_for_successful_job(self, job_id, sleep_time=10, max_time=28800): """ Poll a submitted services job of any type until it completes successfully within the specified timeout, or raise an exception. This lacks the error handling that we need for heavily used servers. """ t_start = time.time() final_states = {"SUCCESSFUL", "FAILED", "TERMINATED", "ABORTED"} while True: job = self.get_job(job_id) log.debug(f"Current job: {job}") state = job["state"] if state in final_states: log.info(f"Job {job_id} is in state {state}, polling complete") break else: t_current = time.time() if t_current - t_start > max_time: raise RuntimeError( f"Polling time ({max_time}s) exceeded, aborting") log.debug(f"Sleeping {sleep_time}s until next status check") time.sleep(sleep_time) if state != "SUCCESSFUL": raise RuntimeError(f"Job {job_id} exited with state {state}") return job # ----------------------------------------------------------------- # OTHER def upload_file(self, file_path): files_d = { "upload_file": open(file_path, "rb") } url = self.to_url("/smrt-link/uploader") log.info(f"Method: POST /smrt-link/uploader {files_d}") log.debug(f"Full URL: {url}") # XXX the Content-Type header is added automatically by the requests # library - sending application/json results in a 404 response from # the akka-http API backend headers = { "Authorization": f"Bearer {self.auth_token}" } response = requests.post(url, files=files_d, headers=headers, verify=self._verify) log.debug(response) response.raise_for_status() return response.json() ######################################################################## # COMMAND LINE TOOLING # def get_smrtlink_client_from_args(args): """ Instantiate a client using the argparse object, when defined with the add_smrtlink_server_args function """ return SmrtLinkClient( host=args.host, port=args.port, username=args.user, password=args.password, verify=not args.insecure) def add_smrtlink_server_args(p): """ Add argparse arguments for connecting to a SMRT Link REST API, for developers who want to use this client in other CLI programs """ DEFAULT_HOST = os.environ.get("PB_SERVICE_HOST", "localhost") DEFAULT_PORT = int(os.environ.get("PB_SERVICE_PORT", Constants.API_PORT)) DEFAULT_USER = os.environ.get("PB_SERVICE_AUTH_USER", None) DEFAULT_PASSWORD = os.environ.get("PB_SERVICE_AUTH_PASSWORD", None) p.add_argument("--host", action="store", default=DEFAULT_HOST, help="SL Server Hostname") p.add_argument("--port", action="store", type=int, default=DEFAULT_PORT, help="SL Server Port") p.add_argument("--user", action="store", default=DEFAULT_USER, help="SL Server User Name") p.add_argument("--password", action="store", default=DEFAULT_PASSWORD, help="SL Server Password") p.add_argument("-k", "--insecure", action="store_true", default=not Constants.DEFAULT_VERIFY, help="Run in insecure mode without validating the SSL certificate") return p def _main(argv=sys.argv): """ Command line utility for generic method calls returning JSON: $ smrtlink_client.py GET /smrt-link/runs \\ --host smrtlink.lab.company.com \\ --user pbuser --password XXX [ { "uniqueId": "2707e6c2-8bc8-4d84-98b3-5f112e7d77ff", "name": "Run_2023_06_01_84001", ... }, ... ] """ def _validate_api_path(s): if not s.startswith("/"): raise ValueError(f"Invalid URL path {s}") elif s.startswith("/SMRTLink"): raise ValueError( f"Please use the base API path, without /SMRTLink/1.0.0") return s parser = argparse.ArgumentParser(_main.__doc__) add_smrtlink_server_args(parser) parser.add_argument("method", choices=["GET", "POST", "PUT", "DELETE", "OPTIONS"], help="HTTP method (GET, POST, PUT, DELETE, OPTIONS)") parser.add_argument("path", type=_validate_api_path, help="API method path, e.g. /smrt-link/runs") parser.add_argument("-d", "--data", type=json.loads, default={}, help="JSON request body (POST/PUT only)") parser.add_argument("-H", "--header", dest="headers", action="append", default=[], help="Optional HTTP header (multiple allowed)") parser.add_argument("-v", "--verbose", action="store_true", default=False, help="Verbose mode logging (INFO)") parser.add_argument("--debug", action="store_true", default=False, help="Debug mode logging (DEBUG)") parser.add_argument("--quiet", action="store_true", default=False, help="Quiet mode logging (ERROR)") args = parser.parse_args(argv[1:]) log_level = logging.WARN if args.debug: log_level = logging.DEBUG elif args.verbose: log_level = logging.INFO elif args.quiet: log_level = logging.ERROR logging.basicConfig(level=log_level, stream=sys.stdout, format="[%(levelname)s] %(asctime)-15sZ %(message)s") if args.insecure: _disable_insecure_warning() client = get_smrtlink_client_from_args(args) headers_d = dict([(s.strip() for s in h.split(":")) for h in args.headers]) response = client.execute_call(args.method, args.path, args.data, headers_d) print(json.dumps(response, indent=2)) return 0 if __name__ == "__main__": sys.exit(_main(sys.argv))