Further import and format cleanups

arteymix · arteymix · commit 781243f275d5 · 2025-03-18T12:02:08.000-07:00
diff --git a/rnaseq_pipeline/gsheet.py b/rnaseq_pipeline/gsheet.py
@@ -1,18 +1,16 @@
-import argparse
+import logging
 import logging
 import os
 import os.path
 import pickle
-import sys
-from os.path import dirname, expanduser, join
-from pkg_resources import resource_filename
+from os.path import join
 
-from googleapiclient.discovery import build
-from google_auth_oauthlib.flow import InstalledAppFlow
-from google.auth.transport.requests import Request
-import luigi
 import pandas as pd
 import xdg.BaseDirectory
+from google.auth.transport.requests import Request
+from google_auth_oauthlib.flow import InstalledAppFlow
+from googleapiclient.discovery import build
+from pkg_resources import resource_filename
 
 SCOPES = ['https://www.googleapis.com/auth/spreadsheets.readonly']
 CREDENTIALS_FILE = resource_filename('rnaseq_pipeline', 'credentials.json')
@@ -47,7 +45,8 @@ def retrieve_spreadsheet(spreadsheet_id, sheet_name):
     service = build('sheets', 'v4', credentials=_authenticate(), cache_discovery=None)
 
     # Retrieve the documents contents from the Docs service.
-    rnaseq_pipeline_queue = service.spreadsheets().values().get(spreadsheetId=spreadsheet_id, range=sheet_name).execute()
+    rnaseq_pipeline_queue = service.spreadsheets().values().get(spreadsheetId=spreadsheet_id,
+                                                                range=sheet_name).execute()
 
     # this will fail if people add new columns
     df = pd.DataFrame(rnaseq_pipeline_queue['values'][1:], columns=rnaseq_pipeline_queue['values'][0])
diff --git a/rnaseq_pipeline/platforms.py b/rnaseq_pipeline/platforms.py
@@ -13,7 +13,7 @@ def get_trim_single_end_reads_task(r1, dest, **kwargs):
         pass
 
     @abstractmethod
-    def get_trim_paired_reads_task(r1,r2, r1_dest, r2_dest, **kwargs):
+    def get_trim_paired_reads_task(r1, r2, r1_dest, r2_dest, **kwargs):
         pass
 
 class BgiPlatform(Platform):
@@ -33,18 +33,18 @@ def __init__(self, instrument):
 
     def get_trim_single_end_reads_task(self, r1, dest, **kwargs):
         return cutadapt.TrimReads(
-                r1,
-                dest,
-                adapter_3prime=BgiPlatform.FORWARD_FILTER,
-                **kwargs)
+            r1,
+            dest,
+            adapter_3prime=BgiPlatform.FORWARD_FILTER,
+            **kwargs)
 
     def get_trim_paired_reads_task(self, r1, r2, r1_dest, r2_dest, **kwargs):
         return cutadapt.TrimPairedReads(
-                r1, r2,
-                r1_dest, r2_dest,
-                adapter_3prime=BgiPlatform.FORWARD_FILTER,
-                reverse_adapter_3prime=BgiPlatform.REVERSE_FILTER,
-                **kwargs)
+            r1, r2,
+            r1_dest, r2_dest,
+            adapter_3prime=BgiPlatform.FORWARD_FILTER,
+            reverse_adapter_3prime=BgiPlatform.REVERSE_FILTER,
+            **kwargs)
 
 class IlluminaPlatform(Platform):
     """
@@ -59,18 +59,18 @@ def __init__(self, instrument):
 
     def get_trim_single_end_reads_task(self, r1, dest, **kwargs):
         return cutadapt.TrimReads(
-                r1,
-                dest,
-                adapter_3prime=IlluminaPlatform.UNIVERSAL_ADAPTER,
-                **kwargs)
+            r1,
+            dest,
+            adapter_3prime=IlluminaPlatform.UNIVERSAL_ADAPTER,
+            **kwargs)
 
     def get_trim_paired_reads_task(self, r1, r2, r1_dest, r2_dest, **kwargs):
         return cutadapt.TrimPairedReads(
-                r1, r2,
-                r1_dest, r2_dest,
-                adapter_3prime=IlluminaPlatform.UNIVERSAL_ADAPTER,
-                reverse_adapter_3prime=IlluminaPlatform.UNIVERSAL_ADAPTER,
-                **kwargs)
+            r1, r2,
+            r1_dest, r2_dest,
+            adapter_3prime=IlluminaPlatform.UNIVERSAL_ADAPTER,
+            reverse_adapter_3prime=IlluminaPlatform.UNIVERSAL_ADAPTER,
+            **kwargs)
 
 class IlluminaNexteraPlatform(Platform):
     """
@@ -85,11 +85,11 @@ def __init__(self, instrument):
 
     def get_trim_single_end_reads_task(self, r1, dest, **kwargs):
         return cutadapt.TrimReads(
-                r1,
-                dest,
-                cut=12,
-                adapter_3prime=IlluminaNexteraPlatform.NEXTERA_ADAPTER,
-                **kwargs)
+            r1,
+            dest,
+            cut=12,
+            adapter_3prime=IlluminaNexteraPlatform.NEXTERA_ADAPTER,
+            **kwargs)
 
     def get_trim_paired_reads_task(self, r1, r2, r1_dest, r2_dest, **kwargs):
         raise NotImplementedError
diff --git a/rnaseq_pipeline/sources/arrayexpress.py b/rnaseq_pipeline/sources/arrayexpress.py
@@ -1,11 +1,11 @@
-from urllib.request import urlretrieve
 import os
 from os.path import join
+from urllib.request import urlretrieve
 
 import luigi
-from luigi.task import WrapperTask
 import pandas as pd
 from bioluigi.tasks.utils import TaskWithOutputMixin
+from luigi.task import WrapperTask
 
 from ..config import rnaseq_pipeline
 from ..platforms import IlluminaPlatform
@@ -21,11 +21,13 @@ class DownloadArrayExpressFastq(luigi.Task):
     def run(self):
         with self.output().temporary_path() as dest_filename:
             urlretrieve(self.fastq_url,
-                        reporthook=lambda numblocks, blocksize, totalsize: self.set_progress_percentage(100.0 * numblocks * blocksize / totalsize),
+                        reporthook=lambda numblocks, blocksize, totalsize: self.set_progress_percentage(
+                            100.0 * numblocks * blocksize / totalsize),
                         filename=dest_filename)
 
     def output(self):
-        return luigi.LocalTarget(join(cfg.OUTPUT_DIR, cfg.DATA, 'arrayexpress', self.sample_id, os.path.basename(self.fastq_url)))
+        return luigi.LocalTarget(
+            join(cfg.OUTPUT_DIR, cfg.DATA, 'arrayexpress', self.sample_id, os.path.basename(self.fastq_url)))
 
 class DownloadArrayExpressSample(TaskWithOutputMixin, WrapperTask):
     experiment_id = luigi.Parameter()
@@ -47,8 +49,10 @@ class DownloadArrayExpressExperiment(TaskWithOutputMixin, WrapperTask):
 
     def run(self):
         # store metadata locally under metadata/arrayexpress/<experiment_id>.sdrf.txt
-        ae_df = pd.read_csv('http://www.ebi.ac.uk/arrayexpress/files/{0}/{0}.sdrf.txt'.format(self.experiment_id), sep='\t')
+        ae_df = pd.read_csv('http://www.ebi.ac.uk/arrayexpress/files/{0}/{0}.sdrf.txt'.format(self.experiment_id),
+                            sep='\t')
         ae_df = ae_df[ae_df['Comment[LIBRARY_STRATEGY]'] == 'RNA-Seq']
         # FIXME: properly handle the order of paired FASTQs
-        yield [DownloadArrayExpressSample(experiment_id=self.experiment_id, sample_id=sample_id, fastq_urls=s['Comment[FASTQ_URI]'].sort_values().tolist())
+        yield [DownloadArrayExpressSample(experiment_id=self.experiment_id, sample_id=sample_id,
+                                          fastq_urls=s['Comment[FASTQ_URI]'].sort_values().tolist())
                for sample_id, s in ae_df.groupby('Comment[ENA_SAMPLE]')]
diff --git a/rnaseq_pipeline/sources/gemma.py b/rnaseq_pipeline/sources/gemma.py
@@ -3,8 +3,8 @@
 import os
 from os.path import join
 
-from bioluigi.tasks.utils import DynamicTaskWithOutputMixin, DynamicWrapperTask
 import luigi
+from bioluigi.tasks.utils import DynamicTaskWithOutputMixin, DynamicWrapperTask
 from luigi.util import requires
 
 from .geo import DownloadGeoSample
@@ -33,9 +33,12 @@ def run(self):
             accession = sample['accession']['accession']
             external_database = sample['accession']['externalDatabase']['name']
             if external_database == 'GEO':
-                download_sample_tasks.append(DownloadGeoSample(accession, metadata=dict(experiment_id=self.experiment_id, sample_id=accession)))
+                download_sample_tasks.append(
+                    DownloadGeoSample(accession, metadata=dict(experiment_id=self.experiment_id, sample_id=accession)))
             elif external_database == 'SRA':
-                download_sample_tasks.append(DownloadSraExperiment(accession, metadata=dict(experiment_id=self.experiment_id, sample_id=accession)))
+                download_sample_tasks.append(DownloadSraExperiment(accession,
+                                                                   metadata=dict(experiment_id=self.experiment_id,
+                                                                                 sample_id=accession)))
             else:
                 logger.warning('Downloading %s from %s is not supported.', accession, external_database)
                 continue
@@ -51,7 +54,9 @@ def run(self):
                     continue
 
                 if len(sample.output()) == 0:
-                    logger.warning('GEO sample %s has no associated FASTQs from which batch information can be extracted.', sample.sample_id)
+                    logger.warning(
+                        'GEO sample %s has no associated FASTQs from which batch information can be extracted.',
+                        sample.sample_id)
                     continue
 
                 # TODO: find a cleaner way to obtain the SRA run accession
diff --git a/rnaseq_pipeline/sources/geo.py b/rnaseq_pipeline/sources/geo.py
@@ -6,29 +6,25 @@
 import logging
 import os
 import re
-import requests
 import tarfile
 import tempfile
-
 from datetime import timedelta
 from functools import lru_cache
 from os.path import join
-from subprocess import Popen
 from urllib.parse import urlparse, parse_qs
 from xml.etree import ElementTree
 
 import luigi
 import requests
-
 from bioluigi.tasks.utils import DynamicTaskWithOutputMixin, DynamicWrapperTask, TaskWithMetadataMixin
 from luigi.util import requires
 
+from .sra import DownloadSraExperiment
 from ..config import rnaseq_pipeline
 from ..miniml_utils import collect_geo_samples, collect_geo_samples_info
-from ..platforms import Platform, BgiPlatform, IlluminaPlatform
+from ..platforms import BgiPlatform, IlluminaPlatform
 from ..targets import ExpirableLocalTarget
 from ..utils import RerunnableTaskMixin
-from .sra import DownloadSraExperiment
 
 cfg = rnaseq_pipeline()
 
@@ -53,7 +49,8 @@ def match_geo_platform(geo_platform):
         return BgiPlatform(geo_platform_title.split(' ')[0])
 
     # Illumina HiSeq X and NextSeq 550 platforms are not prefixed with Illumina
-    illumina_regex = [r'Illumina (.+) \(.+\)', r'(HiSeq X .+) \(.+\)', r'(NextSeq 550) \(.+\)', r'(NextSeq 2000) \(.+\)']
+    illumina_regex = [r'Illumina (.+) \(.+\)', r'(HiSeq X .+) \(.+\)', r'(NextSeq 550) \(.+\)',
+                      r'(NextSeq 2000) \(.+\)']
 
     for r in illumina_regex:
         illumina_match = re.match(r, geo_platform_title)
@@ -85,7 +82,8 @@ def run(self):
             f.write(res.text)
 
     def output(self):
-        return ExpirableLocalTarget(join(cfg.OUTPUT_DIR, cfg.METADATA, 'geo', '{}.xml'.format(self.gsm)), ttl=timedelta(days=14))
+        return ExpirableLocalTarget(join(cfg.OUTPUT_DIR, cfg.METADATA, 'geo', '{}.xml'.format(self.gsm)),
+                                    ttl=timedelta(days=14))
 
 @requires(DownloadGeoSampleMetadata)
 class DownloadGeoSample(DynamicTaskWithOutputMixin, DynamicWrapperTask):
@@ -131,22 +129,25 @@ class DownloadGeoSeriesMetadata(TaskWithMetadataMixin, RerunnableTaskMixin, luig
     def run(self):
         if self.output().is_stale():
             logger.info('%s is stale, redownloading...', self.output())
-        res = requests.get('https://ftp.ncbi.nlm.nih.gov/geo/series/'+ self.gse[:-3] + 'nnn/' + self.gse + '/miniml/' + self.gse + '_family.xml.tgz', stream=True)
+        res = requests.get('https://ftp.ncbi.nlm.nih.gov/geo/series/' + self.gse[
+                                                                        :-3] + 'nnn/' + self.gse + '/miniml/' + self.gse + '_family.xml.tgz',
+                           stream=True)
         res.raise_for_status()
         # we need to use a temporary file because Response.raw does not allow seeking
         with tempfile.TemporaryFile() as tmp:
             for chunk in res.iter_content(chunk_size=1024):
                 tmp.write(chunk)
             tmp.seek(0)
-            with tarfile.open(fileobj=tmp, mode='r:gz') as fin, self.output().temporary_path() as fpath, open(fpath, 'wb') as f:
+            with tarfile.open(fileobj=tmp, mode='r:gz') as fin, self.output().temporary_path() as fpath, open(fpath,
+                                                                                                              'wb') as f:
                 reader = fin.extractfile(self.gse + '_family.xml')
                 while chunk := reader.read(1024):
                     f.write(chunk)
 
-
     def output(self):
         # TODO: remove the _family suffix
-        return ExpirableLocalTarget(join(cfg.OUTPUT_DIR, cfg.METADATA, 'geo', '{}_family.xml'.format(self.gse)), ttl=timedelta(days=14))
+        return ExpirableLocalTarget(join(cfg.OUTPUT_DIR, cfg.METADATA, 'geo', '{}_family.xml'.format(self.gse)),
+                                    ttl=timedelta(days=14))
 
 @requires(DownloadGeoSeriesMetadata)
 class DownloadGeoSeries(DynamicTaskWithOutputMixin, DynamicWrapperTask):
@@ -177,7 +178,9 @@ def run(self):
         with self.output().open('w') as info_out:
             for sample in samples:
                 if len(sample.output()) == 0:
-                    logger.warning('GEO sample %s has no associated FASTQs from which batch information can be extracted.', sample.sample_id)
+                    logger.warning(
+                        'GEO sample %s has no associated FASTQs from which batch information can be extracted.',
+                        sample.sample_id)
                     continue
 
                 # TODO: find a cleaner way to obtain the SRA run accession
diff --git a/rnaseq_pipeline/sources/local.py b/rnaseq_pipeline/sources/local.py
@@ -1,9 +1,9 @@
-from glob import glob
 import os
+from glob import glob
 from os.path import join
 
-from bioluigi.tasks.utils import DynamicTaskWithOutputMixin, DynamicWrapperTask
 import luigi
+from bioluigi.tasks.utils import DynamicTaskWithOutputMixin, DynamicWrapperTask
 
 from ..config import rnaseq_pipeline
 
@@ -23,11 +23,12 @@ def platform(self):
 
     def output(self):
         # we sort to make sure that pair ends are in correct order
-        return [luigi.LocalTarget(f) for f in sorted(glob(join(cfg.OUTPUT_DIR, cfg.DATA, 'local', self.experiment_id, self.sample_id, '*.fastq.gz')))]
+        return [luigi.LocalTarget(f) for f in
+                sorted(glob(join(cfg.OUTPUT_DIR, cfg.DATA, 'local', self.experiment_id, self.sample_id, '*.fastq.gz')))]
 
 class DownloadLocalExperiment(DynamicTaskWithOutputMixin, DynamicWrapperTask):
     experiment_id = luigi.Parameter()
 
     def run(self):
         yield [DownloadLocalSample(self.experiment_id, os.path.basename(f))
-                for f in glob(join(cfg.OUTPUT_DIR, cfg.DATA, 'local', self.experiment_id, '*'))]
+               for f in glob(join(cfg.OUTPUT_DIR, cfg.DATA, 'local', self.experiment_id, '*'))]
diff --git a/rnaseq_pipeline/webviewer/__init__.py b/rnaseq_pipeline/webviewer/__init__.py
@@ -1,15 +1,16 @@
-from os import listdir
 from os.path import basename, getctime, join, dirname
-from glob import glob
 import datetime
+from glob import glob
+from os.path import basename, getctime, join, dirname
 
 import luigi
-from flask import Flask, send_file, render_template, url_for, request, abort
 import pandas as pd
+from flask import Flask, send_file, render_template, abort
 
 from rnaseq_pipeline.config import rnaseq_pipeline
-from rnaseq_pipeline.tasks import GenerateReportForExperiment, CountExperiment, ExtractGeoSeriesBatchInfo, SubmitExperimentDataToGemma, SubmitExperimentBatchInfoToGemma
 from rnaseq_pipeline.gemma import GemmaTaskMixin
+from rnaseq_pipeline.tasks import GenerateReportForExperiment, CountExperiment, SubmitExperimentDataToGemma, \
+    SubmitExperimentBatchInfoToGemma
 
 app = Flask('rnaseq_pipeline.webviewer')
 
@@ -31,7 +32,9 @@ def not_found(e):
 @app.route('/')
 def home():
     report_dir = join(cfg.OUTPUT_DIR, 'report')
-    latest_experiments = [(basename(path), basename(dirname(path)), datetime.datetime.now() - datetime.datetime.fromtimestamp(getctime(path))) for path in sorted(glob(join(report_dir, '*', '*')), key=lambda path: -getctime(path))]
+    latest_experiments = [(basename(path), basename(dirname(path)),
+                           datetime.datetime.now() - datetime.datetime.fromtimestamp(getctime(path))) for path in
+                          sorted(glob(join(report_dir, '*', '*')), key=lambda path: -getctime(path))]
     return render_template('index.html', latest_experiments=latest_experiments[:10])
 
 @app.route('/experiment/<experiment_id>')
@@ -44,14 +47,16 @@ def experiment_summary(experiment_id):
     submit_batch_info_task = SubmitExperimentBatchInfoToGemma(experiment_id)
     ebi_task = submit_batch_info_task.requires()
     if ebi_task.complete():
-        batch_info = pd.read_csv(ebi_task.output().path, sep='\t', names=['geo_sample_id', 'sra_run_id', 'geo_platform_id', 'sra_experiment_url', 'fastq_header'])
+        batch_info = pd.read_csv(ebi_task.output().path, sep='\t',
+                                 names=['geo_sample_id', 'sra_run_id', 'geo_platform_id', 'sra_experiment_url',
+                                        'fastq_header'])
     else:
         batch_info = None
 
     return render_template('experiment-summary.html',
-            experiment_id=experiment_id, batch_info=batch_info,
-            submit_data_task=submit_data_task,
-            submit_batch_info_task=submit_batch_info_task)
+                           experiment_id=experiment_id, batch_info=batch_info,
+                           submit_data_task=submit_data_task,
+                           submit_batch_info_task=submit_batch_info_task)
 
 @app.route('/experiment/<experiment_id>/batch-info')
 def experiment_batch_info(experiment_id):
@@ -94,7 +99,8 @@ def experiment_report(experiment_id, reference_id=None):
     else:
         taxon = 'human'
         source = 'local'
-    generate_report_task = GenerateReportForExperiment(experiment_id, reference_id=reference_id, taxon=taxon, source=source)
+    generate_report_task = GenerateReportForExperiment(experiment_id, reference_id=reference_id, taxon=taxon,
+                                                       source=source)
     if not generate_report_task.complete():
         abort(404, f'No report available for {experiment_id} in {reference_id}.')
     return send_file(generate_report_task.output().path)
diff --git a/tests/test_geo.py b/tests/test_geo.py
@@ -1,9 +1,10 @@
-from rnaseq_pipeline.sources.geo import match_geo_platform, retrieve_geo_platform_miniml, DownloadGeoSampleMetadata, DownloadGeoSeriesMetadata
+import luigi
+
 from rnaseq_pipeline.platforms import IlluminaPlatform
+from rnaseq_pipeline.sources.geo import match_geo_platform, DownloadGeoSampleMetadata, \
+    DownloadGeoSeriesMetadata
 from rnaseq_pipeline.utils import remove_task_output
 
-import luigi
-
 def test_parse_illumina_platform():
     platform = match_geo_platform('GPL30172')
     assert isinstance(platform, IlluminaPlatform)
diff --git a/tests/test_platforms.py b/tests/test_platforms.py
@@ -1,4 +1,4 @@
-from rnaseq_pipeline.platforms import Platform, BgiPlatform, IlluminaPlatform, IlluminaNexteraPlatform
+from rnaseq_pipeline.platforms import BgiPlatform, IlluminaPlatform, IlluminaNexteraPlatform
 
 def test_bgi_platform_trim_single_end_reads():
     task = BgiPlatform('BGISEQ-500').get_trim_single_end_reads_task('r1', 'r1_dest')
diff --git a/tests/test_sra.py b/tests/test_sra.py
diff --git a/tests/test_targets.py b/tests/test_targets.py
diff --git a/tests/test_tasks.py b/tests/test_tasks.py

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-from rnaseq_pipeline.platforms import Platform, BgiPlatform, IlluminaPlatform, IlluminaNexteraPlatform`
	`1`	`+from rnaseq_pipeline.platforms import BgiPlatform, IlluminaPlatform, IlluminaNexteraPlatform`
`2`	`2`
`3`	`3`	`def test_bgi_platform_trim_single_end_reads():`
`4`	`4`	`task = BgiPlatform('BGISEQ-500').get_trim_single_end_reads_task('r1', 'r1_dest')`