shell.py

#!/usr/bin/env python3
"""Interactive shell for testing Keba KeContact framework.

Usage:
    python shell.py <charger_ip> [<charger_ip2> ...]

Examples:
    python shell.py 192.168.1.100
    python shell.py 192.168.1.100 192.168.1.101 192.168.1.102
"""

import asyncio
import sys
import logging
from typing import Dict, List
from keba_kecontact import KebaClient
from keba_kecontact.udp_handler import KebaUdpHandler

logging.basicConfig(level=logging.INFO)
_LOGGER = logging.getLogger(__name__)


class MultiChargerShell:
    """Interactive shell for managing multiple Keba KeContact chargers."""

    def __init__(self, ip_addresses: List[str]):
        self.ip_addresses = ip_addresses
        self.clients: Dict[str, KebaClient] = {}
        self.handler: KebaUdpHandler = None
        self.current_charger: str = ip_addresses[0]
        self.running = False

    async def start(self):
        """Start the shell."""
        print(f"\n=== Keba KeContact Multi-Charger Shell ===")
        print(f"Connecting to {len(self.ip_addresses)} charger(s)...")

        self.handler = KebaUdpHandler()
        await self.handler.start()

        for ip in self.ip_addresses:
            try:
                client = KebaClient(ip, self.handler)
                await client.connect()
                self.clients[ip] = client
                print(f"  Connected to {ip}")
            except Exception as e:
                print(f"  Failed to connect to {ip}: {e}")

        if not self.clients:
            print("\nNo chargers connected. Exiting.")
            return

        print(f"\nActive charger: {self.current_charger}")
        print()
        self.running = True

        try:
            await self.show_help()
            await self.run_loop()
        finally:
            await self.stop()

    async def stop(self):
        """Stop the shell."""
        for ip, client in self.clients.items():
            await client.disconnect()

        if self.handler:
            await self.handler.stop()

        print("\nDisconnected. Goodbye!")

    async def show_help(self):
        """Show available commands."""
        print("Available commands:")
        if len(self.clients) > 1:
            print("  list            - List all connected chargers")
            print("  use <ip>        - Switch to specified charger")
            print("  use <index>     - Switch to charger by index (0, 1, 2...)")
            print("  all <command>   - Execute command on all chargers")
            print()
        print("  r1              - Get Report 1 (product info)")
        print("  r2              - Get Report 2 (current state)")
        print("  r3              - Get Report 3 (power/energy)")
        print("  r100            - Get Report 100 (session info)")
        print("  enable          - Enable charging")
        print("  disable         - Disable charging")
        print("  start           - Start charging")
        print("  stop            - Stop charging")
        print("  curr <mA>       - Set current (e.g., 'curr 16000' for 16A)")
        print("  energy <value>  - Set energy limit")
        print("  display <text>  - Display text on charger")
        print("  info            - Show all basic info")
        print("  help            - Show this help")
        print("  quit/exit       - Exit shell")
        print()

    def get_prompt(self) -> str:
        """Get the command prompt."""
        if len(self.clients) == 1:
            return f"keba@{self.current_charger}> "

        index = list(self.clients.keys()).index(self.current_charger)
        return f"keba[{index}]@{self.current_charger}> "

    async def run_loop(self):
        """Main command loop."""
        while self.running:
            try:
                cmd = await asyncio.get_event_loop().run_in_executor(
                    None, input, self.get_prompt()
                )
                cmd = cmd.strip()

                if not cmd:
                    continue

                await self.execute_command(cmd)

            except KeyboardInterrupt:
                print("\nUse 'quit' or 'exit' to exit")
            except EOFError:
                break
            except Exception as e:
                print(f"Error: {e}")

    async def execute_command(self, cmd: str):
        """Execute a command."""
        parts = cmd.split()
        if not parts:
            return

        command = parts[0].lower()

        try:
            if command in ["quit", "exit", "q"]:
                self.running = False

            elif command == "help":
                await self.show_help()

            elif command == "list" and len(self.clients) > 1:
                await self.cmd_list()

            elif command == "use" and len(self.clients) > 1:
                await self.cmd_use(parts)

            elif command == "all" and len(self.clients) > 1:
                if len(parts) < 2:
                    print("Usage: all <command>")
                    print("Example: all r3")
                    return
                await self.cmd_all(" ".join(parts[1:]))

            elif command == "r1":
                await self.cmd_report_1(self.current_charger)

            elif command == "r2":
                await self.cmd_report_2(self.current_charger)

            elif command == "r3":
                await self.cmd_report_3(self.current_charger)

            elif command == "r100":
                await self.cmd_report_100(self.current_charger)

            elif command == "enable":
                await self.clients[self.current_charger].enable()
                print("OK Charging enabled")

            elif command == "disable":
                await self.clients[self.current_charger].disable()
                print("OK Charging disabled")

            elif command == "start":
                await self.clients[self.current_charger].start_charging()
                print("OK Charging started")

            elif command == "stop":
                await self.clients[self.current_charger].stop_charging()
                print("OK Charging stopped")

            elif command == "curr":
                if len(parts) < 2:
                    print("Usage: curr <milliamps>")
                    print("Example: curr 16000  (16A)")
                    return
                milliamps = int(parts[1])
                await self.clients[self.current_charger].set_current(milliamps)
                print(f"OK Current set to {milliamps}mA ({milliamps/1000}A)")

            elif command == "energy":
                if len(parts) < 2:
                    print("Usage: energy <value>")
                    return
                energy = int(parts[1])
                await self.clients[self.current_charger].set_energy(energy)
                print(f"OK Energy limit set to {energy}")

            elif command == "display":
                if len(parts) < 2:
                    print("Usage: display <text>")
                    return
                text = " ".join(parts[1:])
                await self.clients[self.current_charger].display_text(text)
                print(f"OK Display text set to: {text}")

            elif command == "info":
                await self.cmd_info(self.current_charger)

            else:
                print(f"Unknown command: {command}")
                print("Type 'help' for available commands")

        except TimeoutError:
            print(f"ERROR Timeout - no response from {self.current_charger}")
        except ValueError as e:
            print(f"ERROR Invalid value: {e}")
        except Exception as e:
            print(f"ERROR {e}")

    async def cmd_list(self):
        """List all connected chargers."""
        print("\nConnected chargers:")
        for idx, ip in enumerate(self.clients.keys()):
            marker = "*" if ip == self.current_charger else " "
            print(f"  {marker} [{idx}] {ip}")
        print()

    async def cmd_use(self, parts: List[str]):
        """Switch to a different charger."""
        if len(parts) < 2:
            print("Usage: use <ip> or use <index>")
            return

        target = parts[1]

        if target.isdigit():
            idx = int(target)
            ips = list(self.clients.keys())
            if 0 <= idx < len(ips):
                self.current_charger = ips[idx]
                print(f"Switched to [{idx}] {self.current_charger}")
            else:
                print(f"ERROR Invalid index: {idx}")
        elif target in self.clients:
            self.current_charger = target
            print(f"Switched to {self.current_charger}")
        else:
            print(f"ERROR Charger not found: {target}")

    async def cmd_all(self, cmd: str):
        """Execute command on all chargers."""
        print(f"\nExecuting '{cmd}' on all chargers:")
        for ip in self.clients.keys():
            print(f"\n--- {ip} ---")
            old_charger = self.current_charger
            self.current_charger = ip
            await self.execute_command(cmd)
            self.current_charger = old_charger

    async def cmd_report_1(self, ip: str):
        """Get and display Report 1."""
        client = self.clients[ip]
        report = await client.get_report_1()
        print(f"\n=== Report 1: Product Information ({ip}) ===")
        print(f"  Product:     {report.product}")
        print(f"  Serial:      {report.serial}")
        print(f"  Firmware:    {report.firmware}")
        print(f"  COM-module:  {report.com_module}")
        print(f"  Backend:     {report.backend}")
        print(f"  DIP-Sw1:     {report.dip_switch_1}")
        print(f"  DIP-Sw2:     {report.dip_switch_2}")
        print()

    async def cmd_report_2(self, ip: str):
        """Get and display Report 2."""
        client = self.clients[ip]
        report = await client.get_report_2()
        print(f"\n=== Report 2: Current State ({ip}) ===")
        print(f"  State:         {report.state}")
        print(f"  Error1:        {report.error_1}")
        print(f"  Error2:        {report.error_2}")
        print(f"  Plug:          {report.plug}")
        print(f"  Enable sys:    {report.enable_sys}")
        print(f"  Enable user:   {report.enable_user}")
        print(f"  Max curr:      {report.max_curr} mA ({report.max_curr/1000}A)")
        print(f"  Max curr %:    {report.max_curr_percent}%")
        print(f"  Curr HW:       {report.curr_hw} mA")
        print(f"  Curr user:     {report.curr_user} mA")
        print(f"  Setenergy:     {report.setenergy}")
        print(f"  Output:        {report.output}")
        print(f"  Input:         {report.input}")
        print()

    async def cmd_report_3(self, ip: str):
        """Get and display Report 3."""
        client = self.clients[ip]
        report = await client.get_report_3()
        print(f"\n=== Report 3: Power & Energy ({ip}) ===")
        print(f"  Voltage L1:    {report.u1} V")
        print(f"  Voltage L2:    {report.u2} V")
        print(f"  Voltage L3:    {report.u3} V")
        print(f"  Current L1:    {report.i1} mA")
        print(f"  Current L2:    {report.i2} mA")
        print(f"  Current L3:    {report.i3} mA")
        print(f"  Power:         {report.power_kw} kW")
        print(f"  Power Factor:  {report.pf}%")
        print(f"  Session Energy: {report.energy_present_kwh} kWh")
        print(f"  Total Energy:   {report.energy_total_kwh} kWh")
        print()

    async def cmd_report_100(self, ip: str):
        """Get and display Report 100."""
        client = self.clients[ip]
        report = await client.get_report_100()
        print(f"\n=== Report 100: Session Information ({ip}) ===")
        print(f"  Session ID:    {report.session_id}")
        print(f"  Curr HW:       {report.curr_hw} mA")
        print(f"  E start:       {report.e_start}")
        print(f"  E pres:        {report.e_pres}")
        print(f"  Started:       {report.started}")
        print(f"  Ended:         {report.ended}")
        print(f"  Reason:        {report.reason}")
        print(f"  RFID tag:      {report.rfid_tag}")
        print(f"  RFID class:    {report.rfid_class}")
        print()

    async def cmd_info(self, ip: str):
        """Get and display all basic information."""
        print(f"\nFetching information from {ip}...")

        await self.cmd_report_1(ip)
        await self.cmd_report_2(ip)
        await self.cmd_report_3(ip)


async def main():
    """Main entry point."""
    if len(sys.argv) < 2:
        print("Usage: python shell.py <charger_ip> [<charger_ip2> ...]")
        print("Examples:")
        print("  python shell.py 192.168.1.100")
        print("  python shell.py 192.168.1.100 192.168.1.101")
        sys.exit(1)

    ip_addresses = sys.argv[1:]

    shell = MultiChargerShell(ip_addresses)
    await shell.start()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nExiting...")