diff --git a/pyControl4/advanced_lighting.py b/pyControl4/advanced_lighting.py new file mode 100644 index 0000000..1209958 --- /dev/null +++ b/pyControl4/advanced_lighting.py @@ -0,0 +1,110 @@ +"""Controls Control4 Advanced Lighting scenes.""" + +from __future__ import annotations + +import json +from typing import Any + +from pyControl4.director import C4Director + +ADVANCED_LIGHTING_PATH = "/api/v1/agents/advanced_lighting" +ADVANCED_LIGHTING_COMMANDS_PATH = "/api/v1/agents/advanced_lighting/commands" + + +class C4AdvancedLighting: + """Provides access to Control4 Advanced Lighting scenes. + + The Advanced Lighting agent manages named lighting scenes that can + activate, deactivate, or toggle groups of lights simultaneously. + + Use ``C4AdvancedLighting.create(director)`` to construct an instance — + it fetches the agent's internal device ID from the Director automatically. + """ + + def __init__(self, director: C4Director, agent_device_id: int) -> None: + """Creates a C4AdvancedLighting object. + + Parameters: + `director` - A `pyControl4.director.C4Director` object. + + `agent_device_id` - The Control4 item ID of the Advanced Lighting + agent device. Obtain this via `C4AdvancedLighting.create()`. + """ + self.director = director + self.agent_device_id = int(agent_device_id) + + @classmethod + async def create(cls, director: C4Director) -> C4AdvancedLighting: + """Creates a C4AdvancedLighting instance by fetching the agent device + ID from the Director. + + Parameters: + `director` - A `pyControl4.director.C4Director` object. + + Raises: + `ValueError` if the Advanced Lighting agent is not present or + returns no commands. + """ + data = await director.send_get_request(ADVANCED_LIGHTING_COMMANDS_PATH) + commands: list[dict[str, Any]] = json.loads(data) + if not commands: + raise ValueError( + "Advanced Lighting agent returned no commands — " + "is the Advanced Lighting agent enabled in Control4?" + ) + agent_device_id: int = commands[0]["deviceId"] + return cls(director, agent_device_id) + + async def get_scenes(self) -> list[dict[str, Any]]: + """Returns a list of Advanced Lighting scenes from the Director. + + Each scene dict contains: + + - ``scene_id`` (int): Unique scene identifier + - ``name`` (str): Scene display name + - ``is_active`` (bool): Whether the scene is currently active + - ``ramp_capable`` (bool): Whether the scene supports ramping + - ``full_on`` (bool): Whether the scene sets all loads to full on + - ``full_off`` (bool): Whether the scene sets all loads to full off + - ``user_defined`` (bool): Whether the scene was defined by a user + - ``lock_loads`` (bool): Whether the scene locks loads from other changes + """ + data = await self.director.send_get_request(ADVANCED_LIGHTING_PATH) + result: list[dict[str, Any]] = json.loads(data) + return result + + async def activate_scene(self, scene_id: int) -> None: + """Activates a lighting scene. + + Parameters: + `scene_id` - The Control4 scene ID (from ``get_scenes()``). + """ + await self.director.send_post_request( + f"/api/v1/items/{self.agent_device_id}/commands", + "ACTIVATE_SCENE", + {"SCENE_ID": scene_id}, + ) + + async def deactivate_scene(self, scene_id: int) -> None: + """Deactivates a lighting scene. + + Parameters: + `scene_id` - The Control4 scene ID (from ``get_scenes()``). + """ + await self.director.send_post_request( + f"/api/v1/items/{self.agent_device_id}/commands", + "DEACTIVATE_SCENE", + {"SCENE_ID": scene_id}, + ) + + async def toggle_scene(self, scene_id: int) -> None: + """Toggles a lighting scene between active and inactive. + + Parameters: + `scene_id` - The Control4 scene ID (from ``get_scenes()``). + """ + await self.director.send_post_request( + f"/api/v1/items/{self.agent_device_id}/commands", + "TOGGLE_SCENE", + {"SCENE_ID": scene_id}, + ) diff --git a/tests/test_advanced_lighting.py b/tests/test_advanced_lighting.py new file mode 100644 index 0000000..1fba8a1 --- /dev/null +++ b/tests/test_advanced_lighting.py @@ -0,0 +1,118 @@ +"""Tests for C4AdvancedLighting.""" + +import json +from unittest.mock import AsyncMock, call, patch + +import pytest + +from pyControl4.advanced_lighting import C4AdvancedLighting + +SCENE_LIST = [ + { + "scene_id": 1, + "name": "Morning", + "is_active": False, + "ramp_capable": True, + "full_on": False, + "full_off": False, + "user_defined": True, + "lock_loads": False, + }, + { + "scene_id": 2, + "name": "All Off", + "is_active": False, + "ramp_capable": False, + "full_on": False, + "full_off": True, + "user_defined": False, + "lock_loads": False, + }, +] + +COMMANDS_RESPONSE = json.dumps([{"deviceId": 421, "command": "ACTIVATE_SCENE"}]) +SCENES_RESPONSE = json.dumps(SCENE_LIST) + + +@pytest.mark.asyncio +async def test_create_fetches_agent_device_id(director): + """create() fetches the agent device ID from the commands endpoint.""" + with patch.object( + director, "send_get_request", new=AsyncMock(return_value=COMMANDS_RESPONSE) + ): + adv = await C4AdvancedLighting.create(director) + assert adv.agent_device_id == 421 + + +@pytest.mark.asyncio +async def test_create_raises_on_empty_commands(director): + """create() raises ValueError when the Advanced Lighting agent is absent.""" + with patch.object(director, "send_get_request", new=AsyncMock(return_value="[]")): + with pytest.raises( + ValueError, match="Advanced Lighting agent returned no commands" + ): + await C4AdvancedLighting.create(director) + + +@pytest.mark.asyncio +async def test_get_scenes_returns_list(director): + """get_scenes() returns the parsed list of scene dicts.""" + adv = C4AdvancedLighting(director, 421) + with patch.object( + director, "send_get_request", new=AsyncMock(return_value=SCENES_RESPONSE) + ): + scenes = await adv.get_scenes() + assert len(scenes) == 2 + assert scenes[0]["scene_id"] == 1 + assert scenes[0]["name"] == "Morning" + assert scenes[1]["full_off"] is True + + +@pytest.mark.asyncio +async def test_activate_scene_sends_correct_command(director): + """activate_scene() POSTs ACTIVATE_SCENE with the correct scene ID.""" + adv = C4AdvancedLighting(director, 421) + mock_post = AsyncMock(return_value="{}") + with patch.object(director, "send_post_request", new=mock_post): + await adv.activate_scene(1) + mock_post.assert_called_once_with( + "/api/v1/items/421/commands", + "ACTIVATE_SCENE", + {"SCENE_ID": 1}, + ) + + +@pytest.mark.asyncio +async def test_deactivate_scene_sends_correct_command(director): + """deactivate_scene() POSTs DEACTIVATE_SCENE with the correct scene ID.""" + adv = C4AdvancedLighting(director, 421) + mock_post = AsyncMock(return_value="{}") + with patch.object(director, "send_post_request", new=mock_post): + await adv.deactivate_scene(2) + mock_post.assert_called_once_with( + "/api/v1/items/421/commands", + "DEACTIVATE_SCENE", + {"SCENE_ID": 2}, + ) + + +@pytest.mark.asyncio +async def test_toggle_scene_sends_correct_command(director): + """toggle_scene() POSTs TOGGLE_SCENE with the correct scene ID.""" + adv = C4AdvancedLighting(director, 421) + mock_post = AsyncMock(return_value="{}") + with patch.object(director, "send_post_request", new=mock_post): + await adv.toggle_scene(1) + mock_post.assert_called_once_with( + "/api/v1/items/421/commands", + "TOGGLE_SCENE", + {"SCENE_ID": 1}, + ) + + +@pytest.mark.asyncio +async def test_direct_constructor_agent_device_id(director): + """Direct constructor stores agent_device_id as int.""" + adv = C4AdvancedLighting(director, "421") + assert adv.agent_device_id == 421 + assert isinstance(adv.agent_device_id, int)