Refactor API to use typer as CLI interface (#70)

bany/__main__.py

@@ -1,27 +1,29 @@
+"""
+Command line scripts for dealing with budgets.
+"""
 import logging
-from argparse import ArgumentParser
-from argparse import RawTextHelpFormatter
+from functools import partial
 from pathlib import Path
+from typing import Annotated
+from typing import cast
 
 import pandas
 import pandas as pd
 from rich.logging import RichHandler
+from typer import Option
+from typer import Typer
 
-import bany.cmd.extract
-import bany.cmd.solve
-import bany.cmd.split
 from bany.core.logger import console
-from bany.core.logger import logger
-from bany.core.settings import Settings
 
+app = Typer(add_completion=False, help=__doc__, rich_markup_mode="rich")
 
-def main() -> int:
-    parent = ArgumentParser(add_help=False)
-    parent.add_argument("--verbose", action="store_true", help="use verbose logging?")
-    opts, remaining = parent.parse_known_args()
 
+@app.callback()
+def setup(
+    verbose: Annotated[bool, Option()] = False,
+) -> None:
     logging.basicConfig(
-        level=logging.DEBUG if opts.verbose else logging.INFO,
+        level=logging.DEBUG,
         format="%(message)s",
         datefmt="[%X]",
         handlers=[RichHandler(show_path=False, rich_tracebacks=True, console=console, tracebacks_suppress=(pandas,))],
@@ -32,36 +34,104 @@ def main() -> int:
     pd.set_option("display.max_rows", 512)
     pd.set_option("display.max_columns", 512)
 
-    env = Settings()
-    parser = ArgumentParser(
-        usage=f"{Path(__file__).parent.name}",
-        description=__doc__,
-        parents=[parent],
-        formatter_class=RawTextHelpFormatter,
-        epilog=f"environment:\n{env.model_dump_json(indent=2)}",
+
+app.add_typer(sub := Typer(), name="solve", help="Solve the bucket partitioning problem.")
+
+
+@sub.command()
+def montecarlo(
+    config: Annotated[Path, Option(help="The config file to use.")] = Path.cwd().joinpath("config.yml"),
+    step_size: Annotated[float, Option(help="The Monte Carlo step size to use.")] = 0.01,
+) -> None:
+    """
+    Solve the partitioning problem using Monte Carlo techniques.
+    """
+    from bany.cmd.solve.solvers.montecarlo import BucketSolverConstrainedMonteCarlo
+    from bany.cmd.solve.solvers.basesolver import BucketSolver
+    from bany.cmd.solve.app import main
+
+    main(
+        config=config,
+        solver=cast(
+            type[BucketSolver],
+            partial(
+                BucketSolverConstrainedMonteCarlo.solve,
+                step_size=step_size,
+            ),
+        ),
+    )
+
+
+@sub.command()
+def constrained(
+    config: Annotated[Path, Option(help="The config file to use.")] = Path.cwd().joinpath("config.yml"),
+) -> None:
+    """
+    Solve the partitioning problem using constrained optimization.
+    """
+    from bany.cmd.solve.solvers.constrained import BucketSolverConstrained
+    from bany.cmd.solve.solvers.basesolver import BucketSolver
+    from bany.cmd.solve.app import main
+
+    main(
+        config=config,
+        solver=cast(
+            type[BucketSolver],
+            partial(
+                BucketSolverConstrained.solve,
+            ),
+        ),
     )
 
-    subparsers = parser.add_subparsers(title="commands")
 
-    for controller in (bany.cmd.extract.Controller, bany.cmd.solve.Controller, bany.cmd.split.Controller):
-        subparser = subparsers.add_parser(controller.__module__.split(".")[-2])
-        subparser.set_defaults(controller=controller)
-        controller.add_args(subparser)
+@sub.command()
+def unconstrained(
+    config: Annotated[Path, Option(help="The config file to use.")] = Path.cwd().joinpath("config.yml"),
+) -> None:
+    """
+    Solve the partitioning problem using unconstrained optimization.
+    """
+    from bany.cmd.solve.solvers.unconstrained import BucketSolverSimple
+    from bany.cmd.solve.solvers.basesolver import BucketSolver
+    from bany.cmd.solve.app import main
+
+    main(
+        config=config,
+        solver=cast(
+            type[BucketSolver],
+            partial(
+                BucketSolverSimple.solve,
+            ),
+        ),
+    )
+
+
+@app.command()
+def split() -> None:
+    """
+    Itemize and split a receipt between people.
+    """
+    from .cmd.split.app import App
+
+    raise SystemExit(App().cmdloop())
+
+
+app.add_typer(sub := Typer(), name="extract", help="Parse an input file and create transactions in YNAB.")
+
 
-    opts = parser.parse_args(args=remaining)
+@sub.command()
+def pdf(
+    inp: Annotated[Path, Option(help="The input file to parse.")],
+    config: Annotated[Path, Option(help="The config file to use.")] = Path.cwd().joinpath("config.yml"),
+    upload: Annotated[bool, Option(help="Upload transactions to YNAB budget?")] = False,
+) -> None:
+    """
+    Parse a PDF file and create transactions in YNAB.
+    """
+    from bany.cmd.extract.app import main
 
-    if hasattr(opts, "controller"):
-        # noinspection PyBroadException
-        try:
-            opts.controller(environ=env, options=opts)()
-            return 0
-        except Exception:
-            logger.exception("caught unhandled exception!")
-            return 1
-    else:
-        parser.print_help()
-        return 1
+    main(extractor="", inp=inp, config=config, upload=upload)
 
 
 if __name__ == "__main__":
-    raise SystemExit(main())
+    app()

bany/cmd/extract/app.py

@@ -0,0 +1,53 @@
+"""
+Parse an input file and create transactions in YNAB.
+"""
+import pathlib
+import re
+from pathlib import Path
+
+import pandas as pd
+from moneyed import Money
+from moneyed import USD
+
+from bany.cmd.extract.extractors import EXTRACTORS
+from bany.cmd.extract.extractors.base import Extractor
+from bany.core.logger import logger
+from bany.core.logger import logline
+from bany.ynab.api import YNAB
+
+
+def main(extractor: str, inp: Path, config: Path, upload: bool) -> None:
+    ynab = YNAB()
+
+    if inp.is_dir():
+        inp = _get_latest_pdf(root=inp)
+
+    logger.info("inp: %s", inp)
+
+    extractor: Extractor = EXTRACTORS[extractor].create(ynab=ynab, config=config)
+    extracted = pd.DataFrame(
+        extract.model_dump() | dict(transaction=extract) for extract in extractor.extract(path=inp)
+    )
+
+    if extracted.empty:
+        logger.error("no extracted found in PDF")
+    else:
+        logline()
+        excluded = {"transaction", "account_id", "budget_id", "category_id", "import_id"}
+        logger.info("extracted:\n%s", extracted.loc[:, ~extracted.columns.isin(excluded)])
+
+        logline()
+        logger.info("%-9s : %12s", "TOTAL [+]", Money(extracted[extracted.amount > 0].amount.sum() / 1000, USD))
+        logger.info("%-9s : %12s", "TOTAL [-]", Money(extracted[extracted.amount < 0].amount.sum() / 1000, USD))
+        logger.info("%-9s : %12s", "TOTAL", Money(extracted.amount.sum() / 1000, USD))
+
+        if upload:
+            for i, extract in extracted.iterrows():
+                transaction = extract.transaction
+                ynab.transact(transaction.budget_id, transaction)
+
+
+def _get_latest_pdf(root: pathlib.Path) -> pathlib.Path:
+    for path in sorted(root.glob("*.pdf"), key=lambda p: [int(v) for v in re.findall(r"\d+", p.name)], reverse=True):
+        return path
+    raise FileNotFoundError(root.joinpath("*.pdf"))

bany/cmd/solve/__init__.py

@@ -1 +0,0 @@
-from .controller import Controller  # noqa: F401

bany/cmd/solve/app.py

@@ -0,0 +1,53 @@
+"""
+
+Parse an input file and create transactions in YNAB.
+"""
+from pathlib import Path
+
+import networkx as nx
+import pandas as pd
+
+import bany.cmd.solve.network.loader
+import bany.cmd.solve.network.visualize as visualize
+import bany.cmd.solve.solvers.constrained
+import bany.cmd.solve.solvers.graphsolver
+import bany.cmd.solve.solvers.montecarlo
+import bany.cmd.solve.solvers.unconstrained
+from bany.cmd.solve.network.algo import aggregate_quantity
+from bany.cmd.solve.network.attrs import node_attrs
+from bany.cmd.solve.solvers.basesolver import BucketSolver
+from bany.cmd.solve.solvers.graphsolver import solve
+from bany.core.logger import logger
+from bany.core.logger import logline
+from bany.core.money import moneyfmt
+
+
+def main(solver: type[BucketSolver], config: Path) -> None:
+    frame: pd.DataFrame = bany.cmd.solve.network.loader.load(path=config)
+    logger.info("frame:\n%s\n", frame)
+
+    graph: nx.DiGraph = bany.cmd.solve.network.algo.create(frame)
+    visualize.log("graph", graph, **visualize.FORMATS_INP)
+
+    solved: nx.DiGraph = solve(graph, solver=solver, inplace=False)
+    visualize.log("solved", solved, **visualize.FORMATS_OUT)
+
+    _display_results(solved, fmt=f"%-{max(15, max(len(n) for n in solved.nodes))}s: %s")
+
+
+def _display_results(graph: nx.DiGraph, fmt: str):
+    amount_to_add: float = aggregate_quantity(graph, key=node_attrs.amount_to_add.column)
+    logger.info(fmt, "amount_to_add", moneyfmt(amount_to_add))
+
+    results_value: float = aggregate_quantity(graph, key=node_attrs.results_value.column, leaves=True)
+    logger.info(fmt, "results_value", moneyfmt(results_value))
+
+    results_ratio: float = aggregate_quantity(graph, key=node_attrs.results_ratio.column, leaves=True)
+    logger.info(fmt, "results_ratio", moneyfmt(results_ratio, decimals=10))
+
+    logline()
+    for node in graph:
+        # noinspection PyCallingNonCallable
+        if graph.out_degree(node) == 0 and graph.in_degree(node) == 1:
+            amount_to_add: float = graph.nodes[node][node_attrs.amount_to_add.column]
+            logger.info(fmt, node, moneyfmt(amount_to_add))