Add alarm zones

pyControl4/alarm.py

@@ -1,13 +1,65 @@
-"""Controls Control4 security panel and contact sensor (door, window, motion)
-devices.
+"""Controls Control4 security panel, security zones, and contact sensor
+(door, window, motion) devices.
 """
 
 from __future__ import annotations
 
+import json
+from enum import IntEnum
+
 from pyControl4 import C4Entity
 from pyControl4.director import C4Director
 
 
+class C4ZoneType(IntEnum):
+    """Control4 security zone types.
+
+    These correspond to the type_id values returned by GET_ZONE_LIST.
+    """
+
+    UNKNOWN = 0
+    CONTACT_SENSOR = 1
+    EXTERIOR_DOOR = 2
+    EXTERIOR_WINDOW = 3
+    INTERIOR_DOOR = 4
+    MOTION_SENSOR = 5
+    FIRE = 6
+    GAS = 7
+    CO = 8
+    HEAT = 9
+    WATER = 10
+    SMOKE = 11
+    PRESSURE = 12
+    GLASS_BREAK = 13
+    GATE = 14
+    GARAGE = 15
+    COLD = 16
+
+    @classmethod
+    def get_name(cls, type_id: int) -> str:
+        """Get human-readable name for a zone type ID."""
+        names = {
+            cls.UNKNOWN: "Unknown",
+            cls.CONTACT_SENSOR: "Contact Sensor",
+            cls.EXTERIOR_DOOR: "Exterior Door",
+            cls.EXTERIOR_WINDOW: "Exterior Window",
+            cls.INTERIOR_DOOR: "Interior Door",
+            cls.MOTION_SENSOR: "Motion Sensor",
+            cls.FIRE: "Fire Sensor",
+            cls.GAS: "Gas Detector",
+            cls.CO: "CO Detector",
+            cls.HEAT: "Heat Detector",
+            cls.WATER: "Water Sensor",
+            cls.SMOKE: "Smoke Detector",
+            cls.PRESSURE: "Pressure Sensor",
+            cls.GLASS_BREAK: "Glass Break Sensor",
+            cls.GATE: "Gate Sensor",
+            cls.GARAGE: "Garage Door Sensor",
+            cls.COLD: "Cold Sensor",
+        }
+        return names.get(type_id, "Security Zone")
+
+
 class C4SecurityPanel(C4Entity):
     async def get_arm_state(self) -> str | None:
         """
@@ -214,6 +266,64 @@ async def send_key_press(self, key: str) -> None:
             {"KeyName": key},
         )
 
+    async def get_zones(self) -> list[dict] | None:
+        """Returns a list of all security zones for this partition.
+
+        Each zone is a dictionary with the following keys:
+            - `id` (int): Zone ID
+            - `name` (str): Zone name
+            - `room_id` (int): Room ID where the zone is located
+            - `room_name` (str): Room name where the zone is located
+            - `type_id` (int): Zone type ID (see C4ZoneType enum)
+            - `is_open` (bool): True if zone is open/triggered
+            - `is_bypassed` (bool): True if zone is bypassed
+            - `is_chimeable` (bool): True if zone can chime
+            - `can_bypass` (bool): True if zone can be bypassed
+            - `can_control` (bool): True if zone can be controlled
+        """
+        result = await self.director.send_post_request(
+            f"/api/v1/items/{self.item_id}/commands",
+            "GET_ZONE_LIST",
+            {},
+            is_async=False,
+        )
+        if result:
+            try:
+                data = json.loads(result)
+                zones = data.get("zones", {})
+                # Handle both list and single zone response
+                zone_list = zones.get("zone", [])
+                if isinstance(zone_list, dict):
+                    zone_list = [zone_list]
+                return zone_list
+            except (json.JSONDecodeError, AttributeError):
+                return None
+        return None
+
+    async def get_open_zones(self) -> list[dict] | None:
+        """Returns a list of only open (unsecured) zones for this partition.
+
+        Returns the same zone structure as `get_zones()`, but filtered to only
+        include zones that are currently open/triggered.
+        """
+        result = await self.director.send_post_request(
+            f"/api/v1/items/{self.item_id}/commands",
+            "GET_OPEN_ZONE_LIST",
+            {},
+            is_async=False,
+        )
+        if result:
+            try:
+                data = json.loads(result)
+                zones = data.get("zones", {})
+                zone_list = zones.get("zone", [])
+                if isinstance(zone_list, dict):
+                    zone_list = [zone_list]
+                return zone_list
+            except (json.JSONDecodeError, AttributeError):
+                return None
+        return None
+
 
 class C4ContactSensor:
     def __init__(self, director: C4Director, item_id: int) -> None:

tests/test_alarm_zones.py

@@ -0,0 +1,226 @@
+"""Tests for C4SecurityPanel zone methods and C4ZoneType."""
+
+import json
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from pyControl4.alarm import C4SecurityPanel, C4ZoneType
+
+
+@pytest.mark.asyncio
+async def test_get_zones_returns_zone_list(director):
+    """Test that get_zones returns a list of zones."""
+    zones_response = json.dumps(
+        {
+            "zones": {
+                "zone": [
+                    {
+                        "id": 1,
+                        "name": "Front Door",
+                        "room_id": 100,
+                        "room_name": "Living Room",
+                        "type_id": 2,
+                        "is_open": False,
+                        "is_bypassed": False,
+                        "is_chimeable": True,
+                        "can_bypass": True,
+                        "can_control": True,
+                    },
+                    {
+                        "id": 2,
+                        "name": "Back Window",
+                        "room_id": 101,
+                        "room_name": "Kitchen",
+                        "type_id": 3,
+                        "is_open": True,
+                        "is_bypassed": False,
+                        "is_chimeable": False,
+                        "can_bypass": True,
+                        "can_control": False,
+                    },
+                ]
+            }
+        }
+    )
+
+    with patch.object(
+        director, "send_post_request", new=AsyncMock(return_value=zones_response)
+    ):
+        panel = C4SecurityPanel(director, 500)
+        zones = await panel.get_zones()
+
+    assert zones is not None
+    assert len(zones) == 2
+    assert zones[0]["name"] == "Front Door"
+    assert zones[0]["type_id"] == 2
+    assert zones[0]["is_open"] is False
+    assert zones[1]["name"] == "Back Window"
+    assert zones[1]["is_open"] is True
+
+
+@pytest.mark.asyncio
+async def test_get_zones_single_zone(director):
+    """Test that get_zones handles a single zone response (returned as dict not list)."""
+    zones_response = json.dumps(
+        {
+            "zones": {
+                "zone": {
+                    "id": 1,
+                    "name": "Front Door",
+                    "type_id": 2,
+                    "is_open": False,
+                }
+            }
+        }
+    )
+
+    with patch.object(
+        director, "send_post_request", new=AsyncMock(return_value=zones_response)
+    ):
+        panel = C4SecurityPanel(director, 500)
+        zones = await panel.get_zones()
+
+    assert zones is not None
+    assert len(zones) == 1
+    assert zones[0]["name"] == "Front Door"
+
+
+@pytest.mark.asyncio
+async def test_get_zones_empty(director):
+    """Test that get_zones handles empty zone list."""
+    zones_response = json.dumps({"zones": {"zone": []}})
+
+    with patch.object(
+        director, "send_post_request", new=AsyncMock(return_value=zones_response)
+    ):
+        panel = C4SecurityPanel(director, 500)
+        zones = await panel.get_zones()
+
+    assert zones is not None
+    assert len(zones) == 0
+
+
+@pytest.mark.asyncio
+async def test_get_zones_no_zones_key(director):
+    """Test that get_zones handles response without zones key."""
+    zones_response = json.dumps({})
+
+    with patch.object(
+        director, "send_post_request", new=AsyncMock(return_value=zones_response)
+    ):
+        panel = C4SecurityPanel(director, 500)
+        zones = await panel.get_zones()
+
+    assert zones is not None
+    assert len(zones) == 0
+
+
+@pytest.mark.asyncio
+async def test_get_zones_invalid_json(director):
+    """Test that get_zones handles invalid JSON response."""
+    with patch.object(
+        director, "send_post_request", new=AsyncMock(return_value="not json")
+    ):
+        panel = C4SecurityPanel(director, 500)
+        zones = await panel.get_zones()
+
+    assert zones is None
+
+
+@pytest.mark.asyncio
+async def test_get_zones_sends_correct_command(director):
+    """Test that get_zones sends the correct command."""
+    zones_response = json.dumps({"zones": {"zone": []}})
+    mock = AsyncMock(return_value=zones_response)
+
+    with patch.object(director, "send_post_request", new=mock):
+        panel = C4SecurityPanel(director, 500)
+        await panel.get_zones()
+
+    mock.assert_called_once_with(
+        "/api/v1/items/500/commands",
+        "GET_ZONE_LIST",
+        {},
+        is_async=False,
+    )
+
+
+@pytest.mark.asyncio
+async def test_get_open_zones(director):
+    """Test that get_open_zones returns only open zones."""
+    zones_response = json.dumps(
+        {
+            "zones": {
+                "zone": [
+                    {"id": 2, "name": "Back Window", "is_open": True},
+                ]
+            }
+        }
+    )
+
+    with patch.object(
+        director, "send_post_request", new=AsyncMock(return_value=zones_response)
+    ):
+        panel = C4SecurityPanel(director, 500)
+        zones = await panel.get_open_zones()
+
+    assert zones is not None
+    assert len(zones) == 1
+    assert zones[0]["name"] == "Back Window"
+
+
+@pytest.mark.asyncio
+async def test_get_open_zones_sends_correct_command(director):
+    """Test that get_open_zones sends the correct command."""
+    zones_response = json.dumps({"zones": {"zone": []}})
+    mock = AsyncMock(return_value=zones_response)
+
+    with patch.object(director, "send_post_request", new=mock):
+        panel = C4SecurityPanel(director, 500)
+        await panel.get_open_zones()
+
+    mock.assert_called_once_with(
+        "/api/v1/items/500/commands",
+        "GET_OPEN_ZONE_LIST",
+        {},
+        is_async=False,
+    )
+
+
+class TestC4ZoneType:
+    """Tests for C4ZoneType enum."""
+
+    def test_zone_type_values(self):
+        """Test that zone type enum has correct values."""
+        assert C4ZoneType.UNKNOWN == 0
+        assert C4ZoneType.CONTACT_SENSOR == 1
+        assert C4ZoneType.EXTERIOR_DOOR == 2
+        assert C4ZoneType.EXTERIOR_WINDOW == 3
+        assert C4ZoneType.INTERIOR_DOOR == 4
+        assert C4ZoneType.MOTION_SENSOR == 5
+        assert C4ZoneType.FIRE == 6
+        assert C4ZoneType.GAS == 7
+        assert C4ZoneType.CO == 8
+        assert C4ZoneType.HEAT == 9
+        assert C4ZoneType.WATER == 10
+        assert C4ZoneType.SMOKE == 11
+        assert C4ZoneType.PRESSURE == 12
+        assert C4ZoneType.GLASS_BREAK == 13
+        assert C4ZoneType.GATE == 14
+        assert C4ZoneType.GARAGE == 15
+        assert C4ZoneType.COLD == 16
+
+    def test_get_name_known_types(self):
+        """Test get_name returns correct names for known types."""
+        assert C4ZoneType.get_name(2) == "Exterior Door"
+        assert C4ZoneType.get_name(3) == "Exterior Window"
+        assert C4ZoneType.get_name(5) == "Motion Sensor"
+        assert C4ZoneType.get_name(6) == "Fire Sensor"
+        assert C4ZoneType.get_name(10) == "Water Sensor"
+        assert C4ZoneType.get_name(15) == "Garage Door Sensor"
+
+    def test_get_name_unknown_type(self):
+        """Test get_name returns default for unknown types."""
+        assert C4ZoneType.get_name(99) == "Security Zone"
+        assert C4ZoneType.get_name(-1) == "Security Zone"