black

fusion_report/app.py

@@ -89,7 +89,9 @@ def generate_report(self, params: Namespace) -> None:
         """Generate fusion report with all pages."""
         report = Report(params.config, params.output)
         fusions = [
-            fusion for fusion in self.manager.fusions if len(fusion.tools) >= params.tool_cutoff
+            fusion
+            for fusion in self.manager.fusions
+            if len(fusion.tools) >= params.tool_cutoff
         ]
 
         index_page = report.create_page(
@@ -124,15 +126,31 @@ def enrich(self, params) -> None:
         local_fusions: Dict[str, List[str]] = {}
 
         if not params.no_cosmic:
-            local_fusions.update({CosmicDB(params.db_path).name: CosmicDB(params.db_path).get_all_fusions()})
+            local_fusions.update(
+                {
+                    CosmicDB(params.db_path)
+                    .name: CosmicDB(params.db_path)
+                    .get_all_fusions()
+                }
+            )
 
         if not params.no_fusiongdb2:
-            local_fusions.update({MitelmanDB(params.db_path).name: MitelmanDB(params.db_path).get_all_fusions()})
+            local_fusions.update(
+                {
+                    MitelmanDB(params.db_path)
+                    .name: MitelmanDB(params.db_path)
+                    .get_all_fusions()
+                }
+            )
 
         if not params.no_mitelman:
-            local_fusions.update({FusionGDB2(params.db_path).name: FusionGDB2(params.db_path).get_all_fusions()})
-
-
+            local_fusions.update(
+                {
+                    FusionGDB2(params.db_path)
+                    .name: FusionGDB2(params.db_path)
+                    .get_all_fusions()
+                }
+            )
 
         for fusion in self.manager.fusions:
             for db_name, db_list in local_fusions.items():
@@ -168,7 +186,10 @@ def export_results(self, path: str, extension: str) -> None:
                         if tool in fusion.tools.keys():
                             row.append(
                                 ",".join(
-                                    [f"{key}: {value}" for key, value in fusion.tools[tool].items()]
+                                    [
+                                        f"{key}: {value}"
+                                        for key, value in fusion.tools[tool].items()
+                                    ]
                                 )
                             )
                         else:
@@ -188,12 +209,16 @@ def generate_fusion_list(self, path: str, cutoff: int):
             - fusions_list_filtered.tsv
         """
         # unfiltered list
-        with open(os.path.join(path, "fusion_list.tsv"), "w", encoding="utf-8") as output:
+        with open(
+            os.path.join(path, "fusion_list.tsv"), "w", encoding="utf-8"
+        ) as output:
             for fusion in self.manager.fusions:
                 output.write(f"{fusion.name}\n")
 
         # filtered list
-        with open(os.path.join(path, "fusion_list_filtered.tsv"), "w", encoding="utf-8") as output:
+        with open(
+            os.path.join(path, "fusion_list_filtered.tsv"), "w", encoding="utf-8"
+        ) as output:
             for fusion in self.manager.fusions:
                 if len(fusion.tools) >= cutoff:
                     output.write(f"{fusion.name}\n")
@@ -207,7 +232,10 @@ def score(self, params: Dict[str, Any]) -> None:
         for fusion in self.manager.fusions:
             # tool estimation
             tool_score: float = sum(
-                [params[f"{tool.lower()}_weight"] / 100.0 for tool, _ in fusion.tools.items()]
+                [
+                    params[f"{tool.lower()}_weight"] / 100.0
+                    for tool, _ in fusion.tools.items()
+                ]
             )
             tool_score_expl: List[str] = [
                 format((params[f"{tool}_weight"] / 100.0), ".3f")
@@ -216,16 +244,16 @@ def score(self, params: Dict[str, Any]) -> None:
 
             # database estimation
             db_score: float = sum(
-                float(Settings.FUSION_WEIGHTS[db_name.lower()]) for db_name in fusion.dbs
+                float(Settings.FUSION_WEIGHTS[db_name.lower()])
+                for db_name in fusion.dbs
             )
             db_score_expl: List[str] = [
-                format(Settings.FUSION_WEIGHTS[db_name.lower()], ".3f") for db_name in fusion.dbs
+                format(Settings.FUSION_WEIGHTS[db_name.lower()], ".3f")
+                for db_name in fusion.dbs
             ]
 
             score: float = float("%0.3f" % (0.5 * tool_score + 0.5 * db_score))
-            score_explained = (
-                f'0.5 * ({" + ".join(tool_score_expl)}) + 0.5 * ({" + ".join(db_score_expl)})'
-            )
+            score_explained = f'0.5 * ({" + ".join(tool_score_expl)}) + 0.5 * ({" + ".join(db_score_expl)})'
             fusion.score, fusion.score_explained = score, score_explained
 
     @staticmethod

fusion_report/args_builder.py

@@ -32,12 +32,17 @@ def __init__(self):
             action="version",
             version=f"fusion-report {Settings.VERSION}",
         )
-        self.command_parser: _SubParsersAction = self.parser.add_subparsers(dest="command")
+        self.command_parser: _SubParsersAction = self.parser.add_subparsers(
+            dest="command"
+        )
 
     @property
     def supported_tools(self):
         """Return all supported fusion detection tools."""
-        return [tool["key"].replace("--", "") for tool in self.arguments["args"]["run"]["tools"]]
+        return [
+            tool["key"].replace("--", "")
+            for tool in self.arguments["args"]["run"]["tools"]
+        ]
 
     def build(self) -> None:
         """Build command-line arguments."""
@@ -53,7 +58,9 @@ def run_args(self, args, weight) -> None:
             "Mandatory arguments", "Required arguments to run app."
         )
         for mandatory in args["mandatory"]:
-            run_mandatory.add_argument(mandatory["key"], help=mandatory["help"], type=str)
+            run_mandatory.add_argument(
+                mandatory["key"], help=mandatory["help"], type=str
+            )
         # fusion tools
         run_tools = run_parser.add_argument_group(
             "Tools", "List of all supported tools with their weights."
@@ -113,7 +120,9 @@ def download_args(self, args: Dict[str, Any]) -> None:
             "download", help="Download required databases"
         )
         for mandatory in args["mandatory"]:
-            download_parser.add_argument(mandatory["key"], help=mandatory["help"], type=str)
+            download_parser.add_argument(
+                mandatory["key"], help=mandatory["help"], type=str
+            )
 
         for optional in args["optionals"]:
             download_parser.add_argument(
@@ -126,9 +135,13 @@ def download_args(self, args: Dict[str, Any]) -> None:
 
     def sync_args(self, args: Dict[str, Any]) -> None:
         """Build sync command-line arguments."""
-        download_parser = self.command_parser.add_parser("sync", help="Synchronize databases")
+        download_parser = self.command_parser.add_parser(
+            "sync", help="Synchronize databases"
+        )
         for mandatory in args["mandatory"]:
-            download_parser.add_argument(mandatory["key"], help=mandatory["help"], type=str)
+            download_parser.add_argument(
+                mandatory["key"], help=mandatory["help"], type=str
+            )
 
         self._cosmic(args, download_parser)
 
@@ -141,7 +154,9 @@ def _cosmic(self, args: Dict[str, Any], parser) -> None:
         )
         for cosmic in args["cosmic"]:
             if not cosmic.get("action"):
-                download_cosmic.add_argument(cosmic["key"], help=cosmic.get("help"), type=str)
+                download_cosmic.add_argument(
+                    cosmic["key"], help=cosmic.get("help"), type=str
+                )
             else:
                 download_cosmic.add_argument(
                     cosmic["key"], help=cosmic.get("help"), action=cosmic.get("action")

fusion_report/common/db.py

@@ -68,7 +68,9 @@ def setup(
                     rows: List[List[str]] = [first_line]
                     for line in resource:
                         row = line.split(delimiter)
-                        rows.append(row + ["" for _ in range(len(row), len(first_line))])
+                        rows.append(
+                            row + ["" for _ in range(len(row), len(first_line))]
+                        )
                     self.connection.executemany(
                         f"""INSERT INTO {file.split('/')[-1].split('.')[0].lower()}
                             VALUES ({','.join(['?' for _ in range(0, len(first_line))])})""",

fusion_report/common/fusion_manager.py

@@ -35,7 +35,9 @@ def parse(self, tool: str, file: str, allow_multiple_genes: bool) -> None:
                     factory_parser.set_header(fusion_output.readline().replace('"', ""))
                     for line in fusion_output:
                         line = line.replace('"', "").strip()
-                        fusion_list: List[Tuple[str, Dict[str, Any]]] = factory_parser.parse(line)
+                        fusion_list: List[
+                            Tuple[str, Dict[str, Any]]
+                        ] = factory_parser.parse(line)
                         if allow_multiple_genes is None and len(fusion_list) > 1:
                             fusion_list = [fusion_list[0]]
                         for fusion_name, details in fusion_list:

fusion_report/common/net.py

@@ -30,18 +30,22 @@ def get_cosmic_token(params: Namespace):
             return params.cosmic_token
 
         if params.cosmic_usr is not None and params.cosmic_passwd is not None:
-            return base64.b64encode(f"{params.cosmic_usr}:{params.cosmic_passwd}".encode()).decode(
-                "utf-8"
-            )
+            return base64.b64encode(
+                f"{params.cosmic_usr}:{params.cosmic_passwd}".encode()
+            ).decode("utf-8")
         else:
-            raise DownloadException("COSMIC credentials have not been provided correctly")
+            raise DownloadException(
+                "COSMIC credentials have not been provided correctly"
+            )
 
     @staticmethod
     def run_qiagen_cmd(cmd, return_output=False, silent=False):
         if not silent:
             print(cmd)
         if return_output:
-            output = subprocess.check_output(cmd, shell=True, executable="/bin/bash").strip()
+            output = subprocess.check_output(
+                cmd, shell=True, executable="/bin/bash"
+            ).strip()
             return output
         else:
             subprocess.check_call(cmd, shell=True, executable="/bin/bash")
@@ -78,7 +82,8 @@ def fetch_fusion_file_id(output_path: str):
             sep="\t",
         )
         file_id = df.loc[
-            (df["file_name"] == Settings.COSMIC["FILE"]) & (df["genome_draft"] == "cosmic/GRCh38"),
+            (df["file_name"] == Settings.COSMIC["FILE"])
+            & (df["genome_draft"] == "cosmic/GRCh38"),
             "file_id",
         ].values[0]
         return file_id
@@ -107,7 +112,8 @@ def get_large_file(url: str) -> None:
 
             if (
                 not os.path.exists(file)
-                or (response.headers.get("Content-Length") or 0) != os.stat(file).st_size
+                or (response.headers.get("Content-Length") or 0)
+                != os.stat(file).st_size
             ):
                 with open(file, "wb") as out_file:
                     for chunk in response.iter_content(chunk_size=8192):
@@ -146,7 +152,9 @@ def get_cosmic_from_sanger(token: str, return_err: List[str]) -> None:
             return_err.append(f'{Settings.COSMIC["NAME"]}: {ex}')
 
     @staticmethod
-    def get_cosmic_from_qiagen(token: str, return_err: List[str], outputpath: str) -> None:
+    def get_cosmic_from_qiagen(
+        token: str, return_err: List[str], outputpath: str
+    ) -> None:
         """Method for download COSMIC database from QIAGEN."""
         try:
             Net.get_qiagen_files(token, outputpath)
@@ -173,13 +181,19 @@ def get_cosmic_from_qiagen(token: str, return_err: List[str], outputpath: str) -
     def get_fusiongdb2(self, return_err: List[str]) -> None:
         """Method for download FusionGDB2 database."""
         try:
-            url: str = f'{Settings.FUSIONGDB2["HOSTNAME"]}/{Settings.FUSIONGDB2["FILE"]}'
+            url: str = (
+                f'{Settings.FUSIONGDB2["HOSTNAME"]}/{Settings.FUSIONGDB2["FILE"]}'
+            )
             Net.get_large_file(url)
             file: str = f'{Settings.FUSIONGDB2["FILE"]}'
             df = pd.read_excel(file, engine="openpyxl")
-            df["fusion"] = df["5'-gene (text format)"] + "--" + df["3'-gene (text format)"]
+            df["fusion"] = (
+                df["5'-gene (text format)"] + "--" + df["3'-gene (text format)"]
+            )
             file_csv = "fusionGDB2.csv"
-            df["fusion"].to_csv(file_csv, header=False, index=False, sep=",", encoding="utf-8")
+            df["fusion"].to_csv(
+                file_csv, header=False, index=False, sep=",", encoding="utf-8"
+            )
 
             db = FusionGDB2(".")
             db.setup([file_csv], delimiter=",", skip_header=False)
@@ -195,7 +209,9 @@ def get_mitelman(self, return_err: List[str]) -> None:
             Net.get_large_file(url)
             with ZipFile(Settings.MITELMAN["FILE"], "r") as archive:
                 files = [
-                    x for x in archive.namelist() if "MBCA.TXT.DATA" in x and not "MACOSX" in x
+                    x
+                    for x in archive.namelist()
+                    if "MBCA.TXT.DATA" in x and not "MACOSX" in x
                 ]
                 archive.extractall()
 

fusion_report/common/template.py

@@ -46,7 +46,9 @@ def render(self, page: Page, extra_variables: Dict[str, Any]) -> None:
         """Renders page"""
         merged_variables = {**self.j2_variables.json_serialize(), **extra_variables}
         view = self.j2_env.get_template(page.view).render(merged_variables)
-        with open(os.path.join(self.output_dir, page.filename), "w", encoding="utf-8") as file_out:
+        with open(
+            os.path.join(self.output_dir, page.filename), "w", encoding="utf-8"
+        ) as file_out:
             file_out.write(view)
 
     def include_raw(self, filename: str) -> Markup:

fusion_report/config.py

@@ -64,7 +64,9 @@ def institution(self, institution: Dict[str, str]) -> None:
 
         if "img" in institution.keys() and os.path.exists(institution["img"]):
             image = os.path.join(Settings.ROOT_DIR, institution["img"])
-            self._institution["img"] = base64.b64encode(open(image, "rb").read()).decode("utf-8")
+            self._institution["img"] = base64.b64encode(
+                open(image, "rb").read()
+            ).decode("utf-8")
 
         if "url" in institution.keys():
             self._institution["url"] = institution["url"]

fusion_report/data/fusiongdb2.py

@@ -10,7 +10,9 @@ class FusionGDB2(Db, metaclass=Singleton):
     """Implementation of FusionGDB2 Database. All core functionality is handled by parent class."""
 
     def __init__(self, path: str) -> None:
-        super().__init__(path, Settings.FUSIONGDB2["NAME"], Settings.FUSIONGDB2["SCHEMA"])
+        super().__init__(
+            path, Settings.FUSIONGDB2["NAME"], Settings.FUSIONGDB2["SCHEMA"]
+        )
 
     def get_all_fusions(self) -> List[str]:
         """Returns all fusions from database."""

fusion_report/data/mitelman.py

@@ -14,7 +14,9 @@ def __init__(self, path: str) -> None:
 
     def get_all_fusions(self) -> List[str]:
         """Returns all fusions from database."""
-        query: str = '''SELECT DISTINCT geneshort FROM mbca WHERE geneshort LIKE "%::%"'''
+        query: str = (
+            '''SELECT DISTINCT geneshort FROM mbca WHERE geneshort LIKE "%::%"'''
+        )
         res = self.select(query)
 
         return [fusion["geneshort"].strip().replace("::", "--") for fusion in res]

fusion_report/download.py

@@ -28,9 +28,7 @@ def validate(self, params: Namespace) -> None:
         else:
             self.cosmic_token = Net.get_cosmic_token(params)
 
-
     def download_all(self, params: Namespace) -> None:
-
         # making sure output directory exists
         if not os.path.exists(params.output):
             os.makedirs(params.output, 0o755)

fusion_report/modules/loader.py

@@ -24,7 +24,9 @@ def exec(self, name: str) -> Dict[str, Any]:
         """
         try:
             variables = self.__build_factory(name, self.manager, self.params).load()
-            variables["partial"] = os.path.join(f'{name.replace(".", "/")}', "partial.html")
+            variables["partial"] = os.path.join(
+                f'{name.replace(".", "/")}', "partial.html"
+            )
             return variables
         except AttributeError as ex:
             raise ModuleException(ex)

fusion_report/parsers/abstract_fusion.py

@@ -12,5 +12,7 @@ def set_header(self, header: str, delimiter: Optional[str] = None):
         """Set header."""
 
     @abc.abstractmethod
-    def parse(self, line: str, delimiter: Optional[str] = None) -> List[Tuple[str, Dict[str, Any]]]:
+    def parse(
+        self, line: str, delimiter: Optional[str] = None
+    ) -> List[Tuple[str, Dict[str, Any]]]:
         """Parsing method required to be implemented per fusion tool."""

fusion_report/parsers/arriba.py

@@ -10,7 +10,9 @@ class Arriba(AbstractFusionTool):
     def set_header(self, header: str, delimiter: Optional[str] = "\t"):
         self.header: List[str] = header.strip().split(delimiter)
 
-    def parse_multiple(self, left_fusion: str, right_fusion: str, delimiter: str) -> List[str]:
+    def parse_multiple(
+        self, left_fusion: str, right_fusion: str, delimiter: str
+    ) -> List[str]:
         if delimiter not in left_fusion and delimiter not in right_fusion:
             return [f"{left_fusion}--{right_fusion}"]
 
@@ -20,7 +22,9 @@ def parse_multiple(self, left_fusion: str, right_fusion: str, delimiter: str) ->
 
         return fusions
 
-    def parse(self, line: str, delimiter: Optional[str] = "\t") -> List[Tuple[str, Dict[str, Any]]]:
+    def parse(
+        self, line: str, delimiter: Optional[str] = "\t"
+    ) -> List[Tuple[str, Dict[str, Any]]]:
         col: List[str] = [x.strip() for x in line.split(delimiter)]
         fusions = self.parse_multiple(
             col[self.header.index("#gene1")], col[self.header.index("gene2")], ","