piket/piket_client/cli.py

import click
from piket_client.model import (
    ServerStatus,
    NetworkError,
    Consumption,
    Person,
    Settlement,
    ConsumptionType,
)
from prettytable import PrettyTable


@click.group()
def cli():
    """Poke coco from the command line."""
    pass


@cli.command()
def status():
    """Show the current status of the server."""

    status = ServerStatus.is_server_running()

    if isinstance(status, NetworkError):
        print_error(f"Failed to get data from server, error {status.value}")
        return

    print_ok("Server is available.")

    open_consumptions = ServerStatus.unsettled_consumptions()

    if isinstance(open_consumptions, NetworkError):
        print_error(
            f"Failed to get unsettled consumptions, error {open_consumptions.value}"
        )
        return

    click.echo(f"There are {open_consumptions.amount} unsettled consumptions.")

    if open_consumptions.amount > 0:
        click.echo(f"First at: {open_consumptions.first_timestamp.strftime('%c')}")
        click.echo(f"Most recent at: {open_consumptions.last_timestamp.strftime('%c')}")


@cli.group()
def people():
    pass


@people.command("list")
@click.option("--active/--inactive", default=None)
def list_people(active: bool) -> None:
    people = Person.get_all(active=active)

    if isinstance(people, NetworkError):
        print_error(f"Could not get people: {people.value}")
        return

    table = PrettyTable()
    table.field_names = ["ID", "Full name", "Display name", "Active"]
    table.align["ID"] = "r"
    table.align["Full name"] = "l"
    table.align["Display name"] = "l"
    table.sortby = "Full name"

    for p in people:
        table.add_row([p.person_id, p.full_name, p.display_name, p.active])

    print(table)


@cli.group()
def settlements():
    pass


@settlements.command("show")
@click.argument("settlement_id", type=click.INT)
def show_settlement(settlement_id: int) -> None:
    """Get and view the contents of a Settlement."""
    s = Settlement.get(settlement_id)

    if isinstance(s, NetworkError):
        print_error(f"Could not get Settlement: {s.value}")
        return

    output_settlement_info(s)


@settlements.command("create")
@click.argument("name")
def create_settlement(name: str) -> None:
    """Create a new Settlement."""
    s = Settlement.create(name)

    if isinstance(s, NetworkError):
        print_error(f"Could not create Settlement: {s.value}")
        return

    output_settlement_info(s)


def output_settlement_info(s: Settlement) -> None:
    click.echo(f'Settlement {s.settlement_id}, "{s.name}"')

    click.echo(f"Summary:")
    for key, value in s.consumption_summary.items():
        click.echo(f" - {value['count']} {value['name']} ({key})")

    ct_name_by_id = {key: value["name"] for key, value in s.consumption_summary.items()}

    table = PrettyTable()
    table.field_names = ["Name", *ct_name_by_id.values()]
    table.sortby = "Name"
    table.align = "r"
    table.align["Name"] = "l"  # type: ignore

    zero_fields = {k: "" for k in ct_name_by_id.values()}

    for item in s.per_person_counts.values():
        r = {"Name": item["full_name"], **zero_fields}
        for key, value in item["counts"].items():
            r[ct_name_by_id[key]] = value

        table.add_row(r.values())

    print(table)


@cli.group()
def consumption_types():
    pass


@consumption_types.command("list")
def list_consumption_types() -> None:
    active = ConsumptionType.get_all(active=True)
    inactive = ConsumptionType.get_all(active=False)

    if isinstance(active, NetworkError) or isinstance(inactive, NetworkError):
        print_error("Could not get consumption types!")
        return

    table = PrettyTable()
    table.field_names = ["ID", "Name", "Active"]
    table.sortby = "ID"

    for ct in active + inactive:
        table.add_row([ct.consumption_type_id, ct.name, ct.active])

    print(table)


@consumption_types.command("create")
@click.argument("name")
def create_consumption_type(name: str) -> None:
    ct = ConsumptionType(name=name).create()

    if not isinstance(ct, NetworkError):
        print_ok(f'Created consumption type "{name}" with ID {ct.consumption_type_id}.')


@consumption_types.command("activate")
@click.argument("consumption_type_id", type=click.INT)
def activate_consumption_type(consumption_type_id: int) -> None:
    ct = ConsumptionType.get(consumption_type_id)

    if isinstance(ct, NetworkError):
        print_error(f"Could not get ConsumptionType: {ct.value}")
        return

    result = ct.set_active(True)

    if not isinstance(result, NetworkError):
        print_ok(
            f"Consumption type {ct.consumption_type_id} ({ct.name}) is now active."
        )


@consumption_types.command("deactivate")
@click.argument("consumption_type_id", type=click.INT)
def deactivate_consumption_type(consumption_type_id: int) -> None:
    ct = ConsumptionType.get(consumption_type_id)

    if isinstance(ct, NetworkError):
        print_error(f"Could not get ConsumptionType: {ct.value}")
        return

    result = ct.set_active(False)

    if not isinstance(result, NetworkError):
        print_ok(
            f"Consumption type {ct.consumption_type_id} ({ct.name}) is now inactive."
        )


def print_ok(msg: str) -> None:
    click.echo(click.style(msg, fg="green"))


def print_error(msg: str) -> None:
    click.echo(click.style(msg, fg="red", bold=True), err=True)


if __name__ == "__main__":
    cli()