Add support for managing multiple TVs (#86)

* Add support for managing multiple TVs

This adds the following commands for managing the TVs in the configuration:

* `alga tv add` (replacing `alga setup`)
* `alga tv list`
* `alga tv remove`
* `alga tv rename`
* `alga tv set-default`

A new global option has also been added, which allows specifying which TV to
send commands to on a command-by-command basis. It has to be used like `alga
--tv <identifer> ...`.

The WebSocket client code has been refactored as part of this, to make it
easier to split up the use cases of adding a new TV (where a handshake
shouldn't be performed) and sending regular commands to manage a TV. Tests has
also been added for this part of the code.

A new version of the configuration file format has been added and Alga will
automatically migrate to it if an old configuration file format is detected.
When this is done, the existing TV in the configuration will be assigned an
identifier for "default".

* Bump Python version used for non-tests to Python 3.13

* Stop referencing `alga setup` in README file

.python-version

@@ -1 +1 @@
-3.12
+3.13

README.md

@@ -25,17 +25,14 @@ Setup
 -----
 
 The first time you use the utility, you will need to setup a connection to the TV.
-With the TV on, run `alga setup [hostname/IP]`.
+With the TV on, run `alga tv add <identifier> [hostname/IP]`.
 This will bring up a prompt on the TV asking if you want to accept the pairing.
 When accepted, Alga will be ready to use.
 
-If no hostname or IP address is provided to `alga setup`, it will be default try to connect to "lgwebostv" which should work.
+If no hostname or IP address is provided to `alga tv add`, it will be default try to connect to "lgwebostv" which should work.
 
 The hostname, a key and MAC address will be written to `~/.config/alga/config.json` for future use.
 
-Note: It's currently only possible to pair Alga with one TV at a time.
-Let me know if this is a deal breaker for you.
-
 
 Usage
 -----

src/alga/__init__.py

@@ -1 +1,6 @@
+from alga.types import State
+
+
 __version__ = "1.4.0"
+
+state = State()

src/alga/__main__.py

@@ -1,4 +1,6 @@
-from typer import Typer
+from typing import Annotated, Optional
+
+from typer import Option, Typer
 
 from alga import (
     cli_adhoc,
@@ -8,23 +10,32 @@
     cli_media,
     cli_power,
     cli_remote,
-    cli_setup,
     cli_sound_output,
+    cli_tv,
     cli_version,
     cli_volume,
+    state,
 )
 
 
-app = Typer(no_args_is_help=True)
+def global_options(
+    tv: Annotated[
+        Optional[str], Option(help="Specify which TV the command should be sent to")
+    ] = None,
+) -> None:
+    state.tv_id = tv
+
+
+app = Typer(no_args_is_help=True, callback=global_options)
 app.add_typer(cli_adhoc.app)
 app.add_typer(cli_app.app, name="app")
 app.add_typer(cli_channel.app, name="channel")
 app.add_typer(cli_input.app, name="input")
 app.add_typer(cli_media.app, name="media")
 app.add_typer(cli_power.app, name="power")
 app.add_typer(cli_remote.app, name="remote")
-app.add_typer(cli_setup.app)
 app.add_typer(cli_sound_output.app, name="sound-output")
+app.add_typer(cli_tv.app, name="tv")
 app.add_typer(cli_version.app)
 app.add_typer(cli_volume.app, name="volume")
 

src/alga/cli_power.py

@@ -1,7 +1,7 @@
 from typer import Typer
 from wakeonlan import send_magic_packet
 
-from alga import client, config
+from alga import client, config, state
 
 
 app = Typer(no_args_is_help=True, help="Turn TV (or screen) on and off")
@@ -19,7 +19,11 @@ def on() -> None:
     """Turn TV on via Wake-on-LAN"""
 
     cfg = config.get()
-    send_magic_packet(cfg["mac"])
+
+    tv_id = state.tv_id or cfg["default_tv"]
+    tv = cfg["tvs"][tv_id]
+
+    send_magic_packet(tv["mac"])
 
 
 @app.command()

src/alga/cli_tv.py

@@ -0,0 +1,155 @@
+import json
+from ipaddress import ip_address
+from socket import gaierror, getaddrinfo
+from typing import Annotated, Optional
+
+from getmac import get_mac_address
+from rich import print
+from rich.console import Console
+from rich.table import Table
+from typer import Argument, Exit, Typer
+
+from alga import client, config
+from alga.payloads import get_hello_data
+
+
+def _ip_from_hostname(hostname: str) -> Optional[str]:
+    try:
+        results = getaddrinfo(host=hostname, port=None)
+        # TODO: Do we want to handle receiving multiple IP addresses?
+        return results[0][4][0]
+    except gaierror:
+        return None
+
+
+app = Typer(no_args_is_help=True, help="Set up TVs to manage via Alga")
+
+
+@app.command()
+def add(
+    name: Annotated[str, Argument()], hostname: Annotated[str, Argument()] = "lgwebostv"
+) -> None:  # pragma: no cover
+    """Pair a new TV"""
+
+    # Check if we have been passed an IP address
+    ip: Optional[str]
+    try:
+        ip = ip_address(hostname).compressed
+    except ValueError:
+        ip = _ip_from_hostname(hostname)
+        if not ip:
+            print(
+                f"[red]Could not find any host by the name '{hostname}'.[/red] Is the TV on and connected to the network?"
+            )
+            raise Exit(code=1)
+
+    with client.connect(hostname=hostname, timeout=60) as connection:
+        connection.send(json.dumps(get_hello_data()))
+        response = json.loads(connection.recv())
+        assert response == {
+            "id": "register_0",
+            "payload": {"pairingType": "PROMPT", "returnValue": True},
+            "type": "response",
+        }, "Unexpected response received"
+
+        console = Console()
+        with console.status("Please approve the connection request on the TV now..."):
+            response = json.loads(connection.recv())
+
+    if "client-key" not in response["payload"]:
+        print("[red]Setup failed![/red]")
+        raise Exit(code=1)
+
+    mac_address = get_mac_address(ip=ip)
+
+    cfg = config.get()
+    cfg["tvs"][name] = {
+        "hostname": hostname,
+        "key": response["payload"]["client-key"],
+        "mac": mac_address,
+    }
+
+    if not cfg.get("default_tv"):
+        cfg["default_tv"] = name
+
+    config.write(cfg)
+
+    print("TV configured, Alga is ready to use")
+
+
+@app.command()
+def list() -> None:
+    """List current TVs"""
+
+    cfg = config.get()
+
+    table = Table()
+    table.add_column("Default")
+    table.add_column("Name")
+    table.add_column("Hostname/IP")
+    table.add_column("MAC address")
+
+    for name, tv in cfg["tvs"].items():
+        default = "*" if cfg["default_tv"] == name else ""
+        table.add_row(default, name, tv["hostname"], tv["mac"])
+
+    console = Console()
+    console.print(table)
+
+
+@app.command()
+def remove(name: Annotated[str, Argument()]) -> None:
+    """Remove a TV"""
+
+    cfg = config.get()
+
+    try:
+        cfg["tvs"].pop(name)
+
+        if cfg["default_tv"] == name:
+            cfg["default_tv"] = ""
+
+        config.write(cfg)
+    except KeyError:
+        print(
+            f"[red]A TV with the name '{name}' was not found in the configuration[/red]"
+        )
+        raise Exit(code=1)
+
+
+@app.command()
+def rename(
+    old_name: Annotated[str, Argument()], new_name: Annotated[str, Argument()]
+) -> None:
+    """Change the identifier for a TV"""
+
+    cfg = config.get()
+
+    try:
+        cfg["tvs"][new_name] = cfg["tvs"].pop(old_name)
+
+        if cfg["default_tv"] == old_name:
+            cfg["default_tv"] = new_name
+
+        config.write(cfg)
+    except KeyError:
+        print(
+            f"[red]A TV with the name '{old_name}' was not found in the configuration[/red]"
+        )
+        raise Exit(code=1)
+
+
+@app.command()
+def set_default(name: Annotated[str, Argument()]) -> None:
+    """Set the default TV"""
+
+    cfg = config.get()
+
+    if name in cfg["tvs"]:
+        cfg["default_tv"] = name
+        config.write(cfg)
+    else:
+        print(
+            f"[red]A TV with the name '{name}' was not found in the configuration[/red]"
+        )
+        raise Exit(code=1)