refactor: create antismash downloader module and move there inherent code

src/nplinker/genomics/__init__.py

@@ -1,25 +1,10 @@
 import logging
 
-from .abc import BGCLoaderBase
+logging.getLogger(__name__).addHandler(logging.NullHandler())
+
 from .bgc import BGC
 from .gcf import GCF
-from .genomics import filter_mibig_only_gcf
-from .genomics import get_bgcs_from_gcfs
-from .genomics import get_strains_from_bgcs
 from .genomics import load_gcfs
-from .genomics import map_bgc_to_gcf
-from .genomics import map_strain_to_bgc
-
-logging.getLogger(__name__).addHandler(logging.NullHandler())
+from .abc import BGCLoaderBase
 
-__all__ = [
-    "BGCLoaderBase",
-    "BGC",
-    "GCF",
-    "filter_mibig_only_gcf",
-    "get_bgcs_from_gcfs",
-    "get_strains_from_bgcs",
-    "load_gcfs",
-    "map_bgc_to_gcf",
-    "map_strain_to_bgc"
-]
+__all__ = ["bgc", "gcf", "load_gcfs", "BGCLoaderBase"]

src/nplinker/genomics/abc.py

@@ -1,8 +1,6 @@
 from abc import ABC
 from abc import abstractmethod
-from collections.abc import Sequence
 from .bgc import BGC
-from .gcf import GCF
 
 
 class BGCLoaderBase(ABC):
@@ -32,15 +30,3 @@ def get_bgcs(self) -> dict[str, BGC]:
             dict[str, BGC]: key is BGC name and value is
                 :class:`~nplinker.genomic.BGC` objects
         """
-
-
-
-class GCFLoaderBase(ABC):
-
-    @abstractmethod
-    def get_gcfs(self) -> Sequence[GCF]:
-        """Get GCF objects
-
-        Returns:
-            Sequence[GCF]: a list of :class:`~nplinker.genomic.GCF` objects
-        """

src/nplinker/genomics/antismash/__init__.py

@@ -1,7 +1,9 @@
 import logging
+from .antismash_downloader import download_and_extract_antismash_metadata
 from .antismash_loader import AntismashBGCLoader
 from .antismash_loader import parse_bgc_genbank
 
+
 logging.getLogger(__name__).addHandler(logging.NullHandler())
 
-__all__ = ["AntismashBGCLoader", "parse_bgc_genbank"]
+__all__ = ["AntismashBGCLoader", "parse_bgc_genbank", "download_and_extract_antismash_metadata"]

src/nplinker/genomics/antismash/antismash_downloader.py

@@ -0,0 +1,85 @@
+import os
+import shutil
+from nplinker.logconfig import LogConfig
+from nplinker.utils import download_and_extract_archive
+from nplinker.utils import list_dirs
+from nplinker.utils import list_files
+
+
+logger = LogConfig.getLogger(__name__)
+
+# urls to be given to download antismash data
+ANTISMASH_DB_PAGE_URL = 'https://antismash-db.secondarymetabolites.org/output/{}/'
+ANTISMASH_DB_DOWNLOAD_URL = 'https://antismash-db.secondarymetabolites.org/output/{}/{}'
+
+# The antiSMASH DBV2 is for the availability of the old version, better to keep it.
+ANTISMASH_DBV2_PAGE_URL = 'https://antismash-dbv2.secondarymetabolites.org/output/{}/'
+ANTISMASH_DBV2_DOWNLOAD_URL = 'https://antismash-dbv2.secondarymetabolites.org/output/{}/{}'
+
+
+def _check_roots(download_root: str, extract_root: str):
+    if download_root == extract_root:
+        raise ValueError(
+            "Identical path of download directory and extract directory")
+
+
+def _check_extract_path(extract_path: str):
+    if os.path.exists(extract_path):
+        # check if extract_path is empty
+        files = list(os.listdir(extract_path))
+        if len(files) != 0:
+            raise ValueError(f'Nonempty directory: "{extract_path}"')
+    else:
+        os.makedirs(extract_path, exist_ok=True)
+
+
+def download_and_extract_antismash_metadata(refseq_assembly_id: str,
+                                            download_root: str,
+                                            extract_root: str):
+    """Download and extract Antismash files for a specified refseq_assembly_id.
+
+    Args:
+        refseq_assembly_id(str): Assembled genomic RefSeq (reference sequence) id.
+            If the id is versioned (e.g., "GCF_004339725.1") please be sure to 
+            specify the version as well. 
+        download_root(str): Path to the directory to place downloaded archive in.
+        extract_root(str): Path to the directory data files will be extracted to.
+            Note that if it will create an antismash/ directory in the specified extract_root, if
+            it doesn't already exist.
+            The files will be extracted to <extract_root>/antismash/<refseq_assembly_id>/ dir.
+
+    Raises:
+        ValueError: if download_root and extract_root dirs are the same.
+        ValueError: if <extract_root>/antismash/<refseq_assembly_id> dir is not empty.
+
+    Examples:
+        >>> download_and_extract_antismash_metadata("GCF_004339725.1", "/data/download", "/data/extracted")
+        """
+    extract_path = os.path.join(extract_root, "antismash", refseq_assembly_id)
+
+    _check_roots(download_root, extract_root)
+    _check_extract_path(extract_path)
+
+    for base_url in [ANTISMASH_DB_DOWNLOAD_URL, ANTISMASH_DBV2_DOWNLOAD_URL]:
+        url = base_url.format(refseq_assembly_id, refseq_assembly_id + '.zip')
+
+        download_and_extract_archive(url, download_root, extract_path,
+                                     refseq_assembly_id + '.zip')
+        logger.info(
+            'Genome data successfully extracted for %s', refseq_assembly_id)
+        break
+
+    # delete subdirs
+    logger.info('Deleting unnecessary extracted subdirs and files')
+    subdirs = list_dirs(extract_path)
+    for subdir_path in subdirs:
+        shutil.rmtree(subdir_path)
+
+    files_to_keep = list_files(extract_path, suffix=(".json", ".gbk"))
+
+    for file in list_files(extract_path):
+        if file not in files_to_keep:
+            os.remove(file)
+    logger.info(
+        'download_and_extract_antismash_metadata process for %s is over', refseq_assembly_id
+    )

src/nplinker/genomics/antismash/antismash_loader.py

@@ -126,8 +126,7 @@ def parse_bgc_genbank(file: str) -> BGC:
     product_prediction = features.get("product")
     if product_prediction is None:
         raise ValueError(
-            "Not found product prediction in antiSMASH Genbank file {}".format(
-                file))
+            f"Not found product prediction in antiSMASH Genbank file {file}")
 
     # init BGC
     bgc = BGC(bgc_id=fname, product_prediction=product_prediction)

src/nplinker/genomics/bgc.py

@@ -105,11 +105,7 @@ def add_parent(self, gcf: GCF) -> None:
         Args:
             gcf(GCF): gene cluster family
         """
-        gcf.add_bgc(self)
-
-    def detach_parent(self, gcf: GCF) -> None:
-        """Remove a parent GCF"""
-        gcf.detach_bgc(self)
+        self.parents.add(gcf)
 
     @property
     def strain(self) -> Strain | None:

src/nplinker/genomics/gcf.py

@@ -33,12 +33,11 @@ def __init__(self, gcf_id: str) -> None:
         """
         self.gcf_id = gcf_id
         self._bgcs: set[BGC] = set()
+        self.strains: StrainCollection = StrainCollection()
         self.bigscape_class: str | None = None
         # CG TODO: remove attribute id, see issue 103
         #    https://github.com/NPLinker/nplinker/issues/103
         self.id: int | None = None
-        self.bgc_ids: set[str] = set()
-        self.strains: StrainCollection = StrainCollection()
 
     def __str__(self):
         return f"GCF(id={self.gcf_id}, #bgcs={len(self.bgcs)}, #strains={len(self.strains)})."
@@ -59,25 +58,13 @@ def bgcs(self) -> set[BGC]:
 
     def add_bgc(self, bgc: BGC) -> None:
         """Add a BGC object to the GCF."""
-        bgc.parents.add(self)
         self._bgcs.add(bgc)
-        self.bgc_ids.add(bgc.bgc_id)
+        bgc.add_parent(self)
         if bgc.strain is not None:
             self.strains.add(bgc.strain)
         else:
             logger.warning("No strain specified for the BGC %s", bgc.bgc_id)
 
-    def detach_bgc(self, bgc: BGC) -> None:
-        """Remove a child BGC object."""
-        bgc.parents.remove(self)
-        self._bgcs.remove(bgc)
-        self.bgc_ids.remove(bgc.bgc_id)
-        if bgc.strain is not None:
-            for other_bgc in self._bgcs:
-                if other_bgc.strain == bgc.strain:
-                    return
-            self.strains.remove(bgc.strain)
-
     def has_strain(self, strain: str | Strain) -> bool:
         """Check if the given strain exists.
 
@@ -88,11 +75,3 @@ def has_strain(self, strain: str | Strain) -> bool:
             bool: True when the given strain exist.
         """
         return strain in self.strains
-
-    def has_mibig_only(self) -> bool:
-        """Check if the GCF's children are only BGC objects from MIBiG.
-
-        Returns:
-            bool: True if `GCF.bgcs` are only MIBiG BGC objects
-        """
-        return all(map(lambda bgc: bgc.is_mibig(), self.bgcs))