Add CLI for controlling the amplifier from the command line

.env.example

@@ -0,0 +1,7 @@
+# SPDX-FileCopyrightText: 2026 Renaud Bruyeron <bruyeron@gmail.com>
+#
+# SPDX-License-Identifier: CC0-1.0
+
+# Optional: set to the amplifier's IP (or hostname) to verify you're talking to the right device.
+# If unset, the first UDP broadcast received on port 45454 is used (auto-discovery).
+DEVIALET_IP=192.168.1.100

.github/workflows/pylint.yml

@@ -20,7 +20,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install -e ".[dev,gui]"
+        pip install -e ".[dev,cli,gui]"
     - name: Test with pytest
       run: pytest
     - name: Analysing the code with pylint

README.md

@@ -15,11 +15,12 @@ The amplifier is controlled over UDP on the local network. There is no official
 the protocol was reverse-engineered with Wireshark. "Non-Pro" means hardware from before
 the Core Infinity board.
 
-This repository provides two things:
+This repository provides three things:
 
 - **`pydevialet_expert_nonpro`** — a small, dependency-free Python library implementing the
   UDP protocol. This is the publishable artifact, installable from PyPI and reusable by any
   application (CLIs, GUIs, Home Assistant, …).
+- **A CLI** (`devialet`) — a command-line remote, installed as an optional extra.
 - **A Kivy GUI** (`gui/devimote.py`) — the original graphical remote, now a thin consumer of
   the library. It is a standalone script and is *not* part of the published package.
 
@@ -48,6 +49,61 @@ of `DeviMoteBackEnd`.
 
 All calls are blocking; run them on a worker thread if you need async behaviour.
 
+## CLI
+
+Install the `cli` extra to get the `devialet` command:
+
+```bash
+pip install "pydevialet-expert-nonpro[cli]"
+```
+
+Or run without installing via `uv`:
+
+```bash
+uvx --from "pydevialet-expert-nonpro[cli]" devialet --help
+```
+
+```
+Usage: devialet [OPTIONS] COMMAND [ARGS]...
+
+Commands:
+  status  Show current amplifier status
+  volume  Set volume in dB
+  mute    Toggle mute on/off
+  power   Toggle power (on/standby)
+  source  Select input source by name (case-insensitive partial match)
+```
+
+The CLI auto-discovers the amplifier by listening for its UDP broadcast. You can optionally
+pin it to a specific device by setting `DEVIALET_IP` in a `.env` file (copy `.env.example`):
+
+```bash
+cp .env.example .env   # then edit DEVIALET_IP=<your-amp-ip>
+```
+
+```bash
+devialet status
+devialet volume -- -20.5   # use -- before negative values
+devialet mute
+devialet power
+devialet source analog
+```
+
+### Session state
+
+Each `devialet` invocation is a fresh process, but the amplifier tracks a command
+sequence counter across commands. To keep that counter continuous between
+invocations (and avoid commands being silently ignored as stale duplicates), the CLI
+persists it to a small per-amplifier-IP JSON file:
+
+- **macOS**: `~/Library/Application Support/pydevialet-expert-nonpro/state.json`
+- **Linux**: `$XDG_CONFIG_HOME/pydevialet-expert-nonpro/state.json` (defaults to
+  `~/.config/pydevialet-expert-nonpro/state.json`)
+- **Windows**: `%APPDATA%\pydevialet-expert-nonpro\state.json`
+
+The path (and current counter) is also shown by `devialet status`. It's safe to delete
+this file at any time; it will be recreated starting from 0 on the next command.
+
 ## GUI
 
 The Kivy GUI is a standalone script. Install the project with the `gui` extra and run it:
@@ -65,7 +121,7 @@ python gui/devimote.py
 ## Development
 
 ```bash
-pip install -e ".[dev,gui]"
+pip install -e ".[dev,cli,gui]"
 pytest                                         # unit tests (no amplifier needed)
 pylint src/pydevialet_expert_nonpro tests gui  # must stay at 10.00/10
 reuse lint                                     # licensing compliance

pyproject.toml

@@ -32,7 +32,13 @@ Issues = "https://github.com/gnulabis/devimote/issues"
 
 [project.optional-dependencies]
 dev = ["pylint", "pytest"]
-gui = ["kivy[base]>=2.3.1"]
+cli = ["click", "python-dotenv"]
+gui = [
+    "kivy[base]>=2.3.1",
+]
+
+[project.scripts]
+devialet = "pydevialet_expert_nonpro.cli:main"
 
 [build-system]
 requires = ["setuptools>=64"]

src/pydevialet_expert_nonpro/backend.py

@@ -69,7 +69,7 @@ def _send_command(self, data: bytearray) -> None:
         for _ in range(4):
             data[3] = self.packet_cnt
             data[5] = self.packet_cnt >> 1
-            self.packet_cnt += 1
+            self.packet_cnt = (self.packet_cnt + 1) % 256
             crc = _crc16(data[0:12])
             data[12] = (crc & 0xff00) >> 8
             data[13] = crc & 0x00ff

src/pydevialet_expert_nonpro/cli.py

@@ -0,0 +1,149 @@
+# SPDX-FileCopyrightText: 2026 Renaud Bruyeron <bruyeron@gmail.com>
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+'''CLI remote control for Devialet Expert amplifiers'''
+
+import contextlib
+import json
+import os
+import socket
+from pathlib import Path
+
+import click
+from dotenv import load_dotenv
+
+from pydevialet_expert_nonpro import DeviMoteBackEnd
+
+load_dotenv()
+
+
+def _volume_db(raw: int) -> float:
+    return (raw - 195) / 2.0
+
+
+def _state_path() -> Path:
+    return Path(click.get_app_dir('pydevialet-expert-nonpro')) / 'state.json'
+
+
+def _load_packet_cnt(ip: str) -> int:
+    try:
+        state = json.loads(_state_path().read_text())
+    except (FileNotFoundError, json.JSONDecodeError):
+        return 0
+    return state.get(ip, 0)
+
+
+def _save_packet_cnt(ip: str, packet_cnt: int) -> None:
+    path = _state_path()
+    try:
+        state = json.loads(path.read_text())
+    except (FileNotFoundError, json.JSONDecodeError):
+        state = {}
+    state[ip] = packet_cnt
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(state))
+
+
+def _connect() -> DeviMoteBackEnd:
+    backend = DeviMoteBackEnd()
+    expected_ip = os.getenv('DEVIALET_IP')
+    s = backend.update()
+    if not s['connected']:
+        raise click.ClickException(
+            'Amplifier not found (timeout waiting for UDP broadcast on port 45454)'
+        )
+    if expected_ip:
+        resolved = socket.gethostbyname(expected_ip)
+        if s['ip'] != resolved:
+            click.echo(
+                f"Warning: expected {expected_ip} ({resolved}), connected to {s['ip']}",
+                err=True,
+            )
+    backend.packet_cnt = _load_packet_cnt(s['ip'])
+    return backend
+
+
+@contextlib.contextmanager
+def _session():
+    '''Connect and persist the packet counter afterwards.
+
+    The amplifier tracks the command sequence counter across invocations; since each
+    CLI run creates a fresh backend, a command sent shortly after a previous run gets
+    the same counter values again and is treated as a stale duplicate (this is why
+    source selection could silently fail / leave the amp in a bad state).
+    '''
+    backend = _connect()
+    try:
+        yield backend
+    finally:
+        _save_packet_cnt(backend.status['ip'], backend.packet_cnt)
+
+
+@click.group()
+def main():
+    '''CLI remote control for Devialet Expert amplifiers'''
+
+
+@main.command()
+def status():
+    '''Show current amplifier status'''
+    backend = _connect()
+    s = backend.status
+    ch_name = s['ch_list'].get(s['channel'], f"#{s['channel']}").strip()
+    sources = ', '.join(f"{n.strip()} ({i})" for i, n in sorted(s['ch_list'].items()))
+    click.echo(f"Device:  {s['dev_name'].strip()}")
+    click.echo(f"IP:      {s['ip']}")
+    click.echo(f"Power:   {'ON' if s['power'] else 'STANDBY'}")
+    click.echo(f"Volume:  {_volume_db(s['volume']):+.1f} dB")
+    click.echo(f"Muted:   {'Yes' if s['muted'] else 'No'}")
+    click.echo(f"Source:  {ch_name} ({s['channel']})")
+    click.echo(f"Sources: {sources}")
+    click.echo(f"State:   {_state_path()} (packet_cnt={backend.packet_cnt})")
+
+
+@main.command()
+@click.argument('db', type=float)
+def volume(db):
+    '''Set volume in dB. Use -- before negative values: volume -- -15'''
+    if db > DeviMoteBackEnd.VOLUME_LIMIT:
+        raise click.ClickException(
+            f"Volume {db:+.1f} dB exceeds limit {DeviMoteBackEnd.VOLUME_LIMIT:+.1f} dB"
+        )
+    with _session() as backend:
+        backend.set_volume(db)
+    click.echo(f"Volume set to {db:+.1f} dB")
+
+
+@main.command()
+def mute():
+    '''Toggle mute on/off'''
+    with _session() as backend:
+        backend.toggle_mute()
+    click.echo('Mute toggled')
+
+
+@main.command()
+def power():
+    '''Toggle power (on/standby)'''
+    with _session() as backend:
+        backend.toggle_power()
+    click.echo('Power toggled')
+
+
+@main.command()
+@click.argument('name')
+def source(name):
+    '''Select input source by name (case-insensitive partial match)'''
+    with _session() as backend:
+        ch_list = backend.status['ch_list']
+        matches = {idx: n for idx, n in ch_list.items() if name.lower() in n.lower()}
+        if not matches:
+            available = ', '.join(n.strip() for n in ch_list.values())
+            raise click.ClickException(f"No source matching '{name}'. Available: {available}")
+        if len(matches) > 1:
+            found = ', '.join(f"{n.strip()} ({i})" for i, n in sorted(matches.items()))
+            raise click.ClickException(f"Ambiguous source '{name}', matches: {found}")
+        idx, n = next(iter(matches.items()))
+        backend.set_output(idx)
+    click.echo(f"Source set to {n.strip()}")