Service Example for Reports

sal_example.py

# Required pbcommand >= 0.3.14
from pbcommand.pb_io.report import dict_to_report
from pbcommand.services import ServiceAccessLayer as S
from pbcommand.pb_io import load_report_from_json

sal = S("smrtlink-beta", 8081)

J_ID = 3349
R_UUID = "a37453c1-3463-4dbc-945a-f5cf2db1c812"

j = sal.get_job_by_id(J_ID)

r_details = sal.get_analysis_job_report_details(J_ID, R_UUID)


def load_reports(paths):
    return [load_report_from_json(p) for p in paths]


def entry_points_to_datasets(sal, epoints):
    results = []
    for e in epoints:
        if e.dataset_metatype == "PacBio.DataSet.SubreadSet":
            dataset = sal.get_subreadset_by_id(e.dataset_uuid)
            result = dataset['id'], dataset['uuid'], dataset['path']
            results.append(result)

    return results


def _only(func, ds_files):
    return [f for f in ds_files if func(f)]


def _only_report(ds_files):
    return _only(lambda f: f.file_type_id.endswith("PacBio.FileTypes.JsonReport"), ds_files)


def _only_mapping_reports(ds_files):
    return _only(lambda f: f.path.endswith("mapping_stats_report.json"), ds_files)


def job_summary(sal, job_id):

    job_datastore = sal.get_analysis_job_datastore(job_id)

    job_entry_points = sal.get_analysis_job_entry_points(job_id)

    subread_sets = entry_points_to_datasets(sal, job_entry_points)

    # If you're on the local file system this can be done
    report_ds_files = _only_report(job_datastore.files.values())
    # report_paths = [x.path for x in report_ds_files]

    # Must be on the same filesystem
    # job_reports = load_reports(report_paths)

    # Only Mapping reports
    report_uuids = [f.uuid for f in _only_mapping_reports(job_datastore.files.values())]

    reports_d = [sal.get_analysis_job_report_details(job_id, r_uuid) for r_uuid in report_uuids]

    reports = [dict_to_report(rd) for rd in reports_d]

    return job_datastore, job_entry_points, subread_sets, reports


def run_example():
    s = S("smrtlink-beta", 8081)
    job_id = 3349

    # j = sal.get_job_by_id(3349)
    # Use this
    # ds = sal.get_analysis_job_datastore(job_id)

    # entry_points = sal.get_analysis_job_entry_points(job_id)

    # don't use this.  this is list of dicts (datastore report files)
    # reports = sal.get_analysis_job_reports(job_id)
    return job_summary(s, job_id)