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) @people.command("create") @click.option("--display-name", type=click.STRING) @click.argument("name", type=click.STRING) def create_person(name: str, display_name: str) -> None: """Create a person.""" person = Person(full_name=name, display_name=display_name).create() if isinstance(person, NetworkError): print_error(f"Could not create Person: {person.value}") return print_ok(f'Created person "{name}" with ID {person.person_id}.') @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()