python-ring-doorbell · sdb9696 · Sep 19, 2024 · Sep 13, 2024 · Sep 19, 2024
diff --git 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
@@ -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,40 @@ 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:
+        dev_dict: dict[int, RingGeneric] = {}
+        devices = ring.devices()
+        for device_family in device_families:
+            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:
+            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 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:
+        error(f"Cannot find {' or '.join(device_families)} with name {device_name}")
+
+
 @click.group(
     invoke_without_command=True,
     cls=CatchAllExceptions(click.Group),
@@ -554,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
@@ -748,5 +792,21 @@ 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", "other"], RingOther, device_name)
+    await device.async_open_door()
+    echo(f"{device.name} opened")
+
+
 if __name__ == "__main__":
     cli()  # pylint: disable=no-value-for-parameter
diff --git 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)
 

diff --git 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