add line lengh black

fusion_report/app.py

@@ -88,9 +88,7 @@ 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(
@@ -126,29 +124,17 @@ def enrich(self, params) -> None:
 
         if not params.no_cosmic:
             local_fusions.update(
-                {
-                    CosmicDB(params.db_path)
-                    .name: CosmicDB(params.db_path)
-                    .get_all_fusions()
-                }
+                {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()
-                }
+                {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()
-                }
+                {FusionGDB2(params.db_path).name: FusionGDB2(params.db_path).get_all_fusions()}
             )
 
         for fusion in self.manager.fusions:
@@ -185,10 +171,7 @@ 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:
@@ -208,16 +191,12 @@ 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")
@@ -231,10 +210,7 @@ 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")
@@ -243,16 +219,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,17 +32,12 @@ 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."""
@@ -58,9 +53,7 @@ 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."
@@ -120,9 +113,7 @@ 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(
@@ -135,13 +126,9 @@ 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)
 
@@ -154,9 +141,7 @@ 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,9 +68,7 @@ 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,9 +35,7 @@ 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,22 +30,18 @@ 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")
-        else:
-            raise DownloadException(
-                "COSMIC credentials have not been provided correctly"
+            return base64.b64encode(f"{params.cosmic_usr}:{params.cosmic_passwd}".encode()).decode(
+                "utf-8"
             )
+        else:
+            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")
@@ -82,8 +78,7 @@ 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
@@ -112,8 +107,7 @@ 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):
@@ -152,9 +146,7 @@ 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)
@@ -181,19 +173,13 @@ def get_cosmic_from_qiagen(
     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)
@@ -209,9 +195,7 @@ 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,9 +46,7 @@ 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,9 +64,7 @@ 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,9 +10,7 @@ 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,9 +14,7 @@ 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/modules/loader.py

@@ -24,9 +24,7 @@ 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,7 +12,5 @@ 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,9 +10,7 @@ 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}"]
 
@@ -22,9 +20,7 @@ def parse_multiple(
 
         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")], ","

fusion_report/parsers/dragen.py

@@ -10,9 +10,7 @@ class Dragen(AbstractFusionTool):
     def set_header(self, header: str, delimiter: Optional[str] = "\t"):
         self.header: List[str] = header.strip().split(delimiter)
 
-    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)]
         fusion: str = col[self.header.index("#FusionGene")]
         details: Dict[str, Any] = {