From 00e56ede08f6d9c0addb434aada0cff064457c73 Mon Sep 17 00:00:00 2001 From: Steven B <51370195+sdb9696@users.noreply.github.com> Date: Fri, 13 Sep 2024 12:57:51 +0100 Subject: [PATCH 1/2] Add cli command to open door on intercom --- ring_doorbell/cli.py | 58 ++++++++++++++++++++++++++++++++++++++++++- ring_doorbell/ring.py | 2 ++ 2 files changed, 59 insertions(+), 1 deletion(-) diff --git a/ring_doorbell/cli.py b/ring_doorbell/cli.py index cb56eb5..70be02e 100644 --- a/ring_doorbell/cli.py +++ b/ring_doorbell/cli.py @@ -14,7 +14,7 @@ from contextlib import asynccontextmanager from datetime import datetime from pathlib import Path, PurePath -from typing import Sequence, cast +from typing import Sequence, cast, TypeVar, NoReturn import asyncclick as click @@ -26,6 +26,7 @@ RingDoorBell, RingEvent, RingGeneric, + RingOther, ) from ring_doorbell.const import CLI_TOKEN_FILE, GCM_TOKEN_FILE, PACKAGE_NAME, USER_AGENT from ring_doorbell.listen import can_listen @@ -58,6 +59,9 @@ def CatchAllExceptions(cls): def _handle_exception(debug, exc): if isinstance(exc, click.ClickException): raise + # Handle exit request from click. + if isinstance(exc, click.exceptions.Exit): + sys.exit(exc.exit_code) echo(f"Raised error: {exc}") if debug: raise @@ -119,6 +123,12 @@ async def handle_parse_result(self, ctx, opts, args): echo = click.echo +def error(msg: str) -> NoReturn: + """Print an error and exit.""" + echo(msg) + sys.exit(1) + + def token_updated(token) -> None: """Writes token to file.""" cache_file.write_text(json.dumps(token), encoding="utf-8") @@ -189,6 +199,37 @@ async def _get_ring(username, password, do_update_data, user_agent=USER_AGENT): return ring +_T = TypeVar("_T") + + +def _get_device( + ring: Ring, + device_families: list[str], + device_type: type[_T], + device_name: str | None = None, +) -> _T: + if not device_name: + devs: list[RingGeneric] = [] + for device_family in device_families: + devs.extend(ring.devices()[device_family]) + found = len(devs) + if found == 1: + return cast(_T, devs[0]) + elif found == 0: + error(f"No {' or '.join(device_families)} found") + else: + error( + f"There are {found} {' or '.join(device_families)}, you need to pass the --device-name option." + ) + elif dev := ring.get_device_by_name(device_name): + if dev.family in device_families and isinstance(dev, device_type): + return dev + else: + error(f"{device_name} is not a {' or '.join(device_families)}") + else: + error(f"Cannot find {' or '.join(device_families)} with name {device_name}") + + @click.group( invoke_without_command=True, cls=CatchAllExceptions(click.Group), @@ -748,5 +789,20 @@ def credentials_updated_callback(credentials) -> None: await event_listener.stop() +@cli.command +@click.option( + "--device-name", + required=False, + default=None, + help=("Name of the intercom if there are more than one."), +) +@pass_ring +@click.pass_context +async def open_door(ctx, ring: Ring, device_name: str | None) -> None: + """Open the door of a intercom device.""" + device = _get_device(ring, ["intercoms"], RingOther, device_name) + await device.async_open_door() + + if __name__ == "__main__": cli() # pylint: disable=no-value-for-parameter diff --git a/ring_doorbell/ring.py b/ring_doorbell/ring.py index 6ac2d5d..f070101 100644 --- a/ring_doorbell/ring.py +++ b/ring_doorbell/ring.py @@ -355,6 +355,8 @@ def __getitem__(self, device_type: str) -> Sequence[RingGeneric]: return self._authorized_doorbots if device_type == "other": return self._other + if device_type == "intercoms": + return self._other msg = f"Invalid device_type {device_type}" raise RingError(msg) From 7dc542c98cc168187f5f29f0731ce78bffb00f24 Mon Sep 17 00:00:00 2001 From: Steven B <51370195+sdb9696@users.noreply.github.com> Date: Thu, 19 Sep 2024 08:18:26 +0100 Subject: [PATCH 2/2] Add tests --- .pre-commit-config.yaml | 2 +- ring_doorbell/cli.py | 20 +++++++++------- tests/test_cli.py | 53 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 66 insertions(+), 9 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 050fb27..7bfae72 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -36,7 +36,7 @@ repos: # for more accurate checking than using the pre-commit mypy mirror - id: mypy name: mypy - entry: uv run mypy --verbose + entry: uv run mypy language: system types_or: [python, pyi] require_serial: true diff --git a/ring_doorbell/cli.py b/ring_doorbell/cli.py index 70be02e..31de5e0 100644 --- a/ring_doorbell/cli.py +++ b/ring_doorbell/cli.py @@ -209,10 +209,13 @@ def _get_device( device_name: str | None = None, ) -> _T: if not device_name: - devs: list[RingGeneric] = [] + dev_dict: dict[int, RingGeneric] = {} + devices = ring.devices() for device_family in device_families: - devs.extend(ring.devices()[device_family]) - found = len(devs) + for dev in devices[device_family]: + dev_dict[dev.device_api_id] = dev + devs = list(dev_dict.values()) + found = len(dev_dict) if found == 1: return cast(_T, devs[0]) elif found == 0: @@ -221,9 +224,9 @@ def _get_device( error( f"There are {found} {' or '.join(device_families)}, you need to pass the --device-name option." ) - elif dev := ring.get_device_by_name(device_name): - if dev.family in device_families and isinstance(dev, device_type): - return dev + elif device := ring.get_device_by_name(device_name): + if device.family in device_families and isinstance(device, device_type): + return device else: error(f"{device_name} is not a {' or '.join(device_families)}") else: @@ -595,7 +598,7 @@ async def history_command(ctx, ring: Ring, device_name, kind, limit, json_flag): "-dn", default=None, required=False, - help="Name of the ring device, if ommited uses the first device returned", + help="Name of the ring device, if omitted uses the first device returned", ) @pass_ring @click.pass_context @@ -800,8 +803,9 @@ def credentials_updated_callback(credentials) -> None: @click.pass_context async def open_door(ctx, ring: Ring, device_name: str | None) -> None: """Open the door of a intercom device.""" - device = _get_device(ring, ["intercoms"], RingOther, device_name) + device = _get_device(ring, ["intercoms", "other"], RingOther, device_name) await device.async_open_door() + echo(f"{device.name} opened") if __name__ == "__main__": diff --git a/tests/test_cli.py b/tests/test_cli.py index 5cea069..881421b 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -18,6 +18,7 @@ list_command, listen, motion_detection, + open_door, show, videos, ) @@ -272,3 +273,55 @@ async def test_listen_event_handler(mocker, auth): "Currently active count = 1" ) echomock.assert_called_with(exp) + + +async def test_open_door(ring, aioresponses_mock, devices_fixture): + runner = CliRunner() + + res = await runner.invoke( + open_door, + ["--device-name", "Ingress"], + obj=ring, + ) + assert res.exit_code == 0 + assert res.output == "Ingress opened\n" + + +async def test_get_device(ring, aioresponses_mock, devices_fixture): + runner = CliRunner() + + # Get device by name + res = await runner.invoke( + open_door, + ["--device-name", "Ingress"], + obj=ring, + ) + assert res.exit_code == 0 + assert res.output == "Ingress opened\n" + + # Get device by single type + res = await runner.invoke( + open_door, + [], + obj=ring, + ) + assert res.exit_code == 0 + assert res.output == "Ingress opened\n" + + # Get wrong device type + res = await runner.invoke( + open_door, + ["--device-name", "Front"], + obj=ring, + ) + assert res.exit_code == 1 + assert "Front is not a intercoms" in res.output + + # Wrong name + res = await runner.invoke( + open_door, + ["--device-name", "Frontx"], + obj=ring, + ) + assert res.exit_code == 1 + assert "Cannot find intercoms or other with name Frontx" in res.output