Updated module imports

src/autobigs/cli/info.py

@@ -0,0 +1,46 @@
+from argparse import ArgumentParser, Namespace
+import asyncio
+from autobigs.engine.data.remote.databases.bigsdb import BIGSdbIndex
+
+def setup_parser(parser: ArgumentParser):
+    parser.description = "Fetches the latest BIGSdb MLST database definitions."
+    parser.add_argument(
+        "--retrieve-bigsdbs", "-l",
+        action="store_true",
+        dest="list_dbs",
+        required=False,
+        default=False,
+        help="Lists all known BIGSdb MLST databases (fetched from known APIs and cached)."
+    )
+
+    parser.add_argument(
+        "--retrieve-bigsdb-schemas", "-lschemas",
+        nargs="+",
+        action="extend",
+        dest="list_bigsdb_schemas",
+        required=False,
+        default=[],
+        type=str,
+        help="Lists the known schema IDs for a given BIGSdb sequence definition database name. The name, and then the ID of the schema is given."
+    )
+
+    parser.set_defaults(run=run_asynchronously)
+    return parser
+
+async def run(args: Namespace):
+    async with BIGSdbIndex() as bigsdb_index:
+        if args.list_dbs:
+            known_seqdef_dbs = await bigsdb_index.get_known_seqdef_dbs(force=False)
+            print("\n".join(known_seqdef_dbs.keys()))
+
+        for bigsdb_schema_name in args.list_bigsdb_schemas:
+            schemas = await bigsdb_index.get_schemas_for_seqdefdb(bigsdb_schema_name)
+            for schema_desc, schema_id in schemas.items():
+                print(f"{schema_desc}: {schema_id}")
+
+        if not (args.list_dbs or len(args.list_bigsdb_schemas) > 0):
+            print("Nothing to do. Try specifying \"-l\".")
+
+def run_asynchronously(args: Namespace):
+    asyncio.run(run(args))
+

src/autobigs/cli/meta.py

@@ -0,0 +1,2 @@
+def get_module_base_name(name):
+    return name.split(".")[-1]

src/autobigs/cli/program.py

@@ -0,0 +1,39 @@
+import argparse
+from importlib import metadata
+from os import path
+import os
+
+from autobigs.cli import info, st
+from autobigs.cli.meta import get_module_base_name
+import importlib
+
+root_parser = argparse.ArgumentParser(epilog='Use "%(prog)s info -h" to learn how to get available MLST databases, and their available schemas.'
+                                      + ' Once that is done, use "%(prog)s st -h" to learn how to retrieve MLST profiles.'
+                                      )
+subparsers = root_parser.add_subparsers(required=False)
+
+info.setup_parser(subparsers.add_parser(get_module_base_name(info.__name__)))
+st.setup_parser(subparsers.add_parser(get_module_base_name(st.__name__)))
+
+root_parser.add_argument(
+    "--version",
+    action="store_true",
+    default=False,
+    required=False,
+    help="Displays the autoBIGS.CLI version, and the autoBIGS.Engine version."
+)
+
+
+def run():
+    args = root_parser.parse_args()
+    if args.version:
+        print(f'autoBIGS.CLI is running version {
+              metadata.version("autoBIGS-cli")}.')
+        print(f'autoBIGS.Engine is running version {
+              metadata.version("autoBIGS-engine")}.')
+    if hasattr(args, "run"):
+        args.run(args)
+
+
+if __name__ == "__main__":
+    run()

src/autobigs/cli/st.py

@@ -0,0 +1,70 @@
+
+from argparse import ArgumentParser, Namespace
+import asyncio
+import datetime
+from autobigs.engine.data.local.csv import write_mlst_profiles_as_csv
+from autobigs.engine.data.local.fasta import read_multiple_fastas
+from autobigs.engine.data.remote.databases.bigsdb import BIGSdbIndex
+
+
+def setup_parser(parser: ArgumentParser):
+    parser.description = "Returns MLST exact profile matches."
+    parser.add_argument(
+        "fastas",
+        nargs="+",
+        action='extend',
+        default=[],
+        type=str,
+        help="The FASTA files to process. Multiple can be listed."
+    )
+
+    parser.add_argument(
+        "seqdefdb",
+        help="The BIGSdb seqdef database to use for typing."
+    )
+
+    parser.add_argument(
+        "schema",
+        type=int,
+        help="The BIGSdb seqdef database schema ID (integer) to use for typing."
+    )
+
+    parser.add_argument(
+        "out",
+        default=f'./{datetime.datetime.now().strftime(r"%Y%m%d%H%M%S")}',
+        help="The output CSV name (.csv will be appended)."
+    )
+
+    parser.add_argument(
+        "--exact", "-ex",
+        action="store_true",
+        dest="exact",
+        required=False,
+        default=False,
+        help="Should run exact matching rather than returning all similar ones"
+    )
+
+    parser.add_argument(
+        "--stop-on-fail", "-sof",
+        action="store_true",
+        dest="stop_on_fail",
+        required=False,
+        default=False,
+        help="Should the algorithm stop in the case there are no matches (or partial matches when expecting exact matches)."
+    )
+    parser.set_defaults(run=run_asynchronously)
+    return parser
+
+async def run(args: Namespace):
+    async with BIGSdbIndex() as bigsdb_index:
+        gen_strings = read_multiple_fastas(args.fastas)
+        async with await bigsdb_index.build_profiler_from_seqdefdb(args.seqdefdb, args.schema) as mlst_profiler:
+            mlst_profiles = mlst_profiler.profile_multiple_strings(gen_strings, exact=args.exact)
+            failed = await write_mlst_profiles_as_csv(mlst_profiles, args.out)
+            if len(failed) > 0:
+                print(f"A total of {len(failed)} IDs failed:\n{"\n".join(failed)}")
+            print(f"Completed fetching MLSTs for {len(args.fastas)} sequences.")
+
+def run_asynchronously(args):
+    asyncio.run(run(args))
+