feat: fetch-feeds command

.gitignore

@@ -1,12 +1,9 @@
 .idea
 gtfs/__version__.py
+
 # Generated / downloaded files
 *.zip
-*.p
-*.csv
-*.html
-patco-gtfs/
-transitfeedcrash.txt
+*.pkl
 
 # virtualenv
 .venv/
@@ -18,51 +15,3 @@ transitfeedcrash.txt
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]
-
-# C extensions
-*.so
-
-# Distribution / packaging
-.Python
-env/
-build/
-develop-eggs/
-dist/
-downloads/
-eggs/
-.eggs/
-lib/
-lib64/
-parts/
-sdist/
-var/
-*.egg-info/
-.installed.cfg
-*.egg
-
-# PyInstaller
-#  Usually these files are written by a python script from a template
-#  before PyInstaller builds the exe, so as to inject date/other infos into it.
-*.manifest
-*.spec
-
-# Installer logs
-pip-log.txt
-pip-delete-this-directory.txt
-
-# Unit test / coverage reports
-htmlcov/
-.tox/
-.coverage
-.coverage.*
-.cache
-nosetests.xml
-coverage.xml
-*,cover
-
-# Translations
-*.mo
-*.pot
-
-# PyBuilder
-target/

gtfs/__main__.py

@@ -1,42 +1,18 @@
 #!/usr/bin/env python
 """Command line interface for fetching GTFS."""
-import logging
 from typing import Optional
 
 import typer
 from prettytable.colortable import ColorTable, Themes
 from typing_extensions import Annotated
 
-from .feed_source import FeedSource
 from .feed_sources import feed_sources
-from .utils.constants import Predicate, spinner
+from .utils.check_params import check_bbox, check_output_dir, check_sources
+from .utils.constants import LOG, Predicate, spinner
 from .utils.geom import Bbox, bbox_contains_bbox, bbox_intersects_bbox
+from .utils.multithreading import multi_fetch
 
-logging.basicConfig()
-LOG = logging.getLogger()
-app = typer.Typer()
-
-
-def check_bbox(bbox: str) -> Optional[Bbox]:
-    if bbox is None:
-        return
-    try:
-        min_x, min_y, max_x, max_y = [float(coord) for coord in bbox.split(",")]
-    except ValueError as e:
-        err_message = e.args[0]
-        if "could not convert" in err_message:
-            raise typer.BadParameter("Please pass only numbers as bbox values!")
-        elif "not enough values to unpack" in err_message:
-            raise typer.BadParameter(
-                "Please pass bbox as a string separated by commas like this: min_x,min_y,max_x,max_y"
-            )
-        else:
-            raise typer.BadParameter(f"Unhandled exception: {e}")
-
-    if min_x == max_x or min_y == max_y:
-        raise typer.BadParameter("Area cannot be zero! Please pass a valid bbox.")
-
-    return Bbox(min_x, min_y, max_x, max_y)
+app = typer.Typer(help="Fetch GTFS feeds from various transit agencies.")
 
 
 @app.command()
@@ -58,6 +34,14 @@ def list_feeds(
             help="the gtfs feed should intersect or should be contained inside the user's bbox",
         ),
     ] = None,
+    search: Annotated[
+        Optional[str],
+        typer.Option(
+            "--search",
+            "-s",
+            help="search for feeds based on a string",
+        ),
+    ] = None,
     pretty: Annotated[
         bool,
         typer.Option(
@@ -67,109 +51,112 @@ def list_feeds(
         ),
     ] = False,
 ) -> None:
-    """Filter feeds spatially based on bounding box."""
+    """Filter feeds spatially based on bounding box or search string.
+
+    :param bbox: set of coordinates to filter feeds spatially
+    :param predicate: the gtfs feed should intersect or should be contained inside the user's bbox
+    :param search: Search for feeds based on a string.
+    :param pretty: display feeds inside a pretty table
+    """
+    sources: list = feed_sources
+
+    if search is not None:
+        if bbox is not None or predicate is not None:
+            raise typer.BadParameter(
+                "Please pass either bbox or search text, not both at the same time!"
+            )
+        else:
+            sources = [src for src in feed_sources if search.lower() in src.__name__.lower()]
+
     if bbox is None and predicate is not None:
         raise typer.BadParameter(
             f"Please pass a bbox if you want to filter feeds spatially based on predicate = {predicate}!"
         )
-    elif bbox is not None and predicate is None:
+
+    if bbox is not None and predicate is None:
         raise typer.BadParameter(
             f"Please pass a predicate if you want to filter feeds spatially based on bbox = {bbox}!"
         )
-    else:
-        spinner("Fetching feeds...", 1)
-        if pretty is True:
-            pretty_output = ColorTable(
-                ["Feed Source", "Transit URL", "Bounding Box"], theme=Themes.OCEAN, hrules=1
-            )
 
-        for src in feed_sources:
-            feed_bbox: Bbox = src.bbox
-            if bbox is not None and predicate == "contains":
-                if not bbox_contains_bbox(feed_bbox, bbox):
-                    continue
-            elif bbox is not None and predicate == "intersects":
-                if (not bbox_intersects_bbox(feed_bbox, bbox)) and (
-                    not bbox_intersects_bbox(bbox, feed_bbox)
-                ):
-                    continue
-
-            if pretty is True:
-                pretty_output.add_row(
-                    [
-                        src.__name__,
-                        src.url,
-                        [feed_bbox.min_x, feed_bbox.min_y, feed_bbox.max_x, feed_bbox.max_y],
-                    ]
-                )
-                continue
+    spinner("Filtering feeds...", 1)
 
-            print(src.url)
+    if pretty is True:
+        pretty_output = ColorTable(
+            ["Feed Source", "Transit URL", "Bounding Box"], theme=Themes.OCEAN, hrules=1
+        )
+
+    for src in sources:
+        feed_bbox: Bbox = src.bbox
+        if bbox is not None and predicate == "contains":
+            if not bbox_contains_bbox(feed_bbox, bbox):
+                continue
+        elif bbox is not None and predicate == "intersects":
+            if (not bbox_intersects_bbox(feed_bbox, bbox)) and (
+                not bbox_intersects_bbox(bbox, feed_bbox)
+            ):
+                continue
 
         if pretty is True:
-            print("\n" + pretty_output.get_string())
+            pretty_output.add_row(
+                [
+                    src.__name__,
+                    src.url,
+                    [feed_bbox.min_x, feed_bbox.min_y, feed_bbox.max_x, feed_bbox.max_y],
+                ]
+            )
+            continue
+
+        print(src.url)
+
+    if pretty is True:
+        print("\n" + pretty_output.get_string())
 
 
 @app.command()
-def fetch_feeds(sources=None):
-    """
+def fetch_feeds(
+    sources: Annotated[
+        Optional[str],
+        typer.Option(
+            "--sources",
+            "-src",
+            help="pass value as a string separated by commas like this: Berlin,AlbanyNy,...",
+            callback=check_sources,
+        ),
+    ] = None,
+    output_dir: Annotated[
+        str,
+        typer.Option(
+            "--output-dir",
+            "-o",
+            help="the directory where the downloaded feeds will be saved, default is feeds",
+            callback=check_output_dir,
+        ),
+    ] = "feeds",
+    concurrency: Annotated[
+        Optional[int],
+        typer.Option(
+            "--concurrency",
+            "-c",
+            help="the number of concurrent downloads, default is 4",
+        ),
+    ] = 4,
+) -> None:
+    """Fetch feeds from sources.
+
     :param sources: List of :FeedSource: modules to fetch; if not set, will fetch all available.
+    :param output_dir: The directory where the downloaded feeds will be saved; default is "feeds"
+     in current working directory.
+    :param concurrency: The number of concurrent downloads; default is 4.
     """
-    statuses = {}  # collect the statuses for all the files
 
-    # default to use all of them
     if not sources:
         sources = feed_sources
+    else:
+        sources = [src for src in feed_sources if src.__name__.lower() in sources.lower()]
 
-    LOG.info("Going to fetch feeds from sources: %s", sources)
-    for src in sources:
-        LOG.debug("Going to start fetch for %s...", src)
-        try:
-            if issubclass(src, FeedSource):
-                inst = src()
-                inst.fetch()
-                statuses.update(inst.status)
-            else:
-                LOG.warning(
-                    "Skipping class %s, which does not subclass FeedSource.",
-                    src.__name__,
-                )
-        except AttributeError:
-            LOG.error("Skipping feed %s, which could not be found.", src)
-
-    # remove last check key set at top level of each status dictionary
-    if "last_check" in statuses:
-        del statuses["last_check"]
-
-    ptable = ColorTable(
-        [
-            "file",
-            "new?",
-            "valid?",
-            "current?",
-            "newly effective?",
-            "error",
-        ],
-        theme=Themes.OCEAN,
-        hrules=1,
-    )
-
-    for file_name in statuses:
-        stat = statuses[file_name]
-        msg = []
-        msg.append(file_name)
-        msg.append("x" if "is_new" in stat and stat["is_new"] else "")
-        msg.append("x" if "is_valid" in stat and stat["is_valid"] else "")
-        msg.append("x" if "is_current" in stat and stat["is_current"] else "")
-        msg.append("x" if "newly_effective" in stat and stat.get("newly_effective") else "")
-        if "error" in stat:
-            msg.append(stat["error"])
-        else:
-            msg.append("")
-        ptable.add_row(msg)
+    LOG.info(f"Going to fetch feeds from sources: {sources}")
 
-    LOG.info("Results:\n%s", ptable.get_string())
-    LOG.info("All done!")
+    multi_fetch(sources, output_dir, concurrency)
 
 
 if __name__ == "__main__":