diff --git a/.gitignore b/.gitignore index cc0b120..285d91b 100644 --- a/.gitignore +++ b/.gitignore @@ -120,3 +120,97 @@ node_modules/ # Deepcode AI .dccache + +# MCP Ticketer +.mcp-ticketer/ + +# Claude MPM Generated Documentation (AI agent files only) +CLAUDE.md +ARCHITECTURE.md +PROJECT_VALIDATION.md + +# MCP Configuration (environment-specific) +.mcp.json + +# Claude MPM Directories (all content) +.claude-mpm/ +.claude/ + +# Temporary documentation (user-specific) +docs/bobmatnyc/ + +# Added by Claude MPM /mpm-init +$RECYCLE.BIN/ +*.backup +*.bak +*.cab +*.cache +*.cert +*.cover +*.crt +*.db +*.key +*.lnk +*.mpm.tmp +*.msi +*.msm +*.msp +*.old +*.pem +*.pytest_cache +*.sql +*.sqlite +*.sqlite3 +*.sublime-project +*.sublime-workspace +*.swo +*.swp +*.temp +*.tmp +*~ +.DS_Store +.conda/ +.env +.env.* +.idea/ +.kuzu-memory/ +.mcp-vector-search/ +.nox/ +.npm +.project +.pydevproject +.pytype/ +.secrets/ +.settings/ +.vscode/ +.yarn/ +Desktop.ini +Thumbs.db +_build/ +backup/ +backups/ +conda-env/ +credentials/ +ehthumbs.db +env.bak/ +kuzu-memories/ +logs/ +npm-debug.log* +pip-wheel-metadata/ +site/ +temp/ +tmp/ +venv.bak/ +virtualenv/ +wheels/ +yarn-debug.log* +yarn-error.log* + +# Added by Claude MPM /mpm-init +.claude-mpm/*.log +.claude-mpm/cache/ +.claude-mpm/logs/ +.claude-mpm/sessions/ +.claude-mpm/tmp/ +.claude/cache/ +.claude/sessions/ diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 9dba83d..5cf33ce 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -115,6 +115,7 @@ repos: name: šŸŒŸ Starring code with pylint language: system types: [python] + exclude: ^examples/ entry: poetry run pylint - id: trailing-whitespace name: āœ„ Trim Trailing Whitespace diff --git a/README.md b/README.md index f0163d9..190a97b 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,64 @@ This package allows you to fetch data from SmartThings. pip install pysmartthings ``` +## Usage + +### Quick Start + +```python +import asyncio +from aiohttp import ClientSession +from pysmartthings import SmartThings + +async def main(): + # Get your token at: https://account.smartthings.com/tokens + token = "YOUR_SMARTTHINGS_TOKEN" + + async with ClientSession() as session: + api = SmartThings(session=session) + api.authenticate(token) + + # List all devices + devices = await api.get_devices() + for device in devices: + print(f"{device.label} ({device.device_id})") + + # Get device status + if devices: + status = await api.get_device_status(devices[0].device_id) + print(f"Status: {status}") + +if __name__ == "__main__": + asyncio.run(main()) +``` + +### Features + +- **Async/await API** - Built on aiohttp for high-performance async operations +- **Type-safe models** - Full type hints and data validation using dataclasses +- **Comprehensive device control** - Support for switches, dimmers, locks, thermostats, and more +- **Real-time events** - Subscribe to device events via Server-Sent Events (SSE) +- **Scene execution** - Trigger SmartThings scenes and automations +- **Lock code management** - Set and manage smart lock PIN codes +- **Error handling** - Specific exceptions for different error conditions + +### Examples + +Check out the [examples](examples/) directory for comprehensive examples: + +- **[basic_usage.py](examples/basic_usage.py)** - Authentication, listing devices, getting device details +- **[control_devices.py](examples/control_devices.py)** - Control switches, dimmers, color lights, thermostats +- **[scenes.py](examples/scenes.py)** - List and execute scenes +- **[lock_codes.py](examples/lock_codes.py)** - Manage smart lock PIN codes +- **[event_subscription.py](examples/event_subscription.py)** - Subscribe to real-time device events +- **[async_patterns.py](examples/async_patterns.py)** - Production-ready async patterns and best practices + +See the [examples/README.md](examples/README.md) for setup instructions and detailed documentation. + +### API Documentation + +For detailed API documentation and technical information, see [CLAUDE.md](CLAUDE.md). + ## Changelog & Releases This repository keeps a change log using [GitHub's releases][releases] diff --git a/examples/README.md b/examples/README.md new file mode 100644 index 0000000..7a1ae71 --- /dev/null +++ b/examples/README.md @@ -0,0 +1,229 @@ +# pysmartthings Examples + +This directory contains practical examples demonstrating how to use the pysmartthings library. + +## Prerequisites + +1. **Python 3.12+** - This library uses modern async/await syntax +2. **SmartThings Account** - You need a Samsung SmartThings account +3. **Personal Access Token** - Required for API authentication + +## Getting Your SmartThings API Token + +1. Go to the [SmartThings Personal Access Tokens](https://account.smartthings.com/tokens) page +2. Click "Generate new token" +3. Give it a name (e.g., "pysmartthings development") +4. Select the required scopes: + - `r:devices:*` - Read device information + - `x:devices:*` - Execute device commands + - `r:locations:*` - Read location information + - `r:scenes:*` - Read scenes + - `x:scenes:*` - Execute scenes + - `r:rooms:*` - Read rooms + - Additional scopes as needed for your use case +5. Click "Generate token" +6. **Copy the token immediately** - you won't be able to see it again! + +## Installation + +```bash +# Install pysmartthings +pip install pysmartthings + +# Or with poetry +poetry add pysmartthings +``` + +## Running the Examples + +Each example is a standalone Python script. You'll need to replace `YOUR_TOKEN_HERE` with your actual SmartThings Personal Access Token. + +### Quick Start + +```bash +# Basic usage - list devices and get device details +python examples/basic_usage.py + +# Control devices - switches, dimmers, lights, thermostats +python examples/control_devices.py + +# Scenes - list and execute scenes +python examples/scenes.py + +# Lock codes - manage smart lock codes (requires lock device) +python examples/lock_codes.py + +# Event subscription - real-time device events via SSE +python examples/event_subscription.py + +# Async patterns - concurrent operations and error handling +python examples/async_patterns.py +``` + +## Examples Overview + +### 1. basic_usage.py +**What it demonstrates:** +- Creating and authenticating a SmartThings client +- Listing all devices +- Getting detailed device information +- Checking device status +- Listing locations and rooms + +**Use this when:** You're getting started with the library + +### 2. control_devices.py +**What it demonstrates:** +- Controlling switches (on/off) +- Dimming lights (0-100%) +- Setting color lights (RGB) +- Controlling thermostats (temperature, mode) +- Checking capabilities before executing commands + +**Use this when:** You need to control your SmartThings devices + +### 3. scenes.py +**What it demonstrates:** +- Listing available scenes +- Filtering scenes by location +- Executing scenes +- Error handling for scene operations + +**Use this when:** You want to trigger SmartThings scenes/automations + +### 4. lock_codes.py +**What it demonstrates:** +- Checking if a device supports lock codes +- Setting lock codes (PIN codes) +- Deleting lock codes +- Locking and unlocking doors +- Managing multiple codes + +**Use this when:** You have smart locks and need to manage access codes + +### 5. event_subscription.py +**What it demonstrates:** +- Subscribing to real-time device events +- Handling device state changes +- Processing health events +- Managing subscriptions lifecycle +- Event filtering and processing + +**Use this when:** You need real-time notifications of device changes + +### 6. async_patterns.py +**What it demonstrates:** +- Running multiple operations concurrently +- Proper session management +- Error handling with specific exceptions +- Retry logic for transient failures +- Timeout handling +- Best practices for async/await + +**Use this when:** You're building production applications + +## Common Patterns + +### Session Management + +```python +from aiohttp import ClientSession +from pysmartthings import SmartThings + +# Recommended: Manage session yourself for multiple clients +async with ClientSession() as session: + api = SmartThings(session=session) + api.authenticate("YOUR_TOKEN_HERE") + # Use api... + +# Alternative: Let SmartThings manage the session +api = SmartThings() +api.authenticate("YOUR_TOKEN_HERE") +# Session created automatically on first request +``` + +### Error Handling + +```python +from pysmartthings import ( + SmartThingsAuthenticationFailedError, + SmartThingsNotFoundError, + SmartThingsConnectionError, +) + +try: + device = await api.get_device(device_id) +except SmartThingsAuthenticationFailedError: + print("Token expired or invalid") +except SmartThingsNotFoundError: + print(f"Device {device_id} not found") +except SmartThingsConnectionError as err: + print(f"Network error: {err}") +``` + +### Checking Capabilities + +```python +from pysmartthings import Capability + +# Check if device supports a capability before using it +if Capability.SWITCH in device.capabilities: + await api.execute_device_command( + device.device_id, + Capability.SWITCH, + Command.ON + ) +``` + +## Environment Variables (Optional) + +For security, you can store your token in an environment variable: + +```bash +export SMARTTHINGS_TOKEN="your-token-here" +``` + +Then in your code: + +```python +import os +token = os.getenv("SMARTTHINGS_TOKEN") +api.authenticate(token) +``` + +## Troubleshooting + +### "Authentication Failed" Error +- Verify your token is correct and hasn't expired +- Check that you've granted the necessary scopes +- Tokens can be revoked - you may need to generate a new one + +### "Not Found" Error +- Check that the device/scene/location ID is correct +- Ensure the resource exists in your SmartThings account +- Some devices may not be accessible via the cloud API + +### "Connection Error" +- Verify your internet connection +- Check if the SmartThings API is operational +- Review rate limits (429 errors) + +### "Command Error" +- Verify the device supports the capability +- Check command arguments match the capability requirements +- Some devices require specific component names + +## Additional Resources + +- [SmartThings API Documentation](https://developer.smartthings.com/docs) +- [SmartThings Capabilities Reference](https://developer.smartthings.com/docs/devices/capabilities/capabilities-reference) +- [pysmartthings GitHub](https://github.com/pySmartThings/pysmartthings) +- [Technical Documentation](../CLAUDE.md) - For contributors and advanced usage + +## Contributing + +Found a bug in an example or want to add a new one? See [CONTRIBUTING.md](../CONTRIBUTING.md) for guidelines. + +## License + +These examples are provided under the Apache 2.0 license, same as the main library. diff --git a/examples/__init__.py b/examples/__init__.py new file mode 100644 index 0000000..2093df9 --- /dev/null +++ b/examples/__init__.py @@ -0,0 +1,5 @@ +"""SmartThings usage examples. + +This package contains practical examples demonstrating +how to use the pysmartthings library for various use cases. +""" diff --git a/examples/async_patterns.py b/examples/async_patterns.py new file mode 100644 index 0000000..6cb55bb --- /dev/null +++ b/examples/async_patterns.py @@ -0,0 +1,349 @@ +"""Async patterns and best practices for pysmartthings. + +This example demonstrates: +- Running multiple operations concurrently with asyncio.gather +- Proper session management +- Error handling with specific SmartThings exceptions +- Retry logic for transient failures +- Timeout handling +- Production-ready async/await patterns +""" + +import asyncio +from typing import Any + +from aiohttp import ClientSession, ClientTimeout + +from pysmartthings import ( + Capability, + Command, + Device, + SmartThings, + SmartThingsAuthenticationFailedError, + SmartThingsConnectionError, + SmartThingsError, + SmartThingsNotFoundError, + SmartThingsRateLimitError, +) + + +async def concurrent_device_status(api: SmartThings) -> None: + """Get status of all devices concurrently. + + This demonstrates how to use asyncio.gather to run multiple + async operations in parallel for better performance. + + Args: + api: SmartThings API client + + """ + print("\nConcurrent Device Status:") # noqa: T201 + print("-" * 60) # noqa: T201# Get all devices + devices = await api.get_devices() + print(f"Found {len(devices)} devices") # noqa: T201 + + # Get all device statuses concurrently + print("Fetching all device statuses concurrently...") # noqa: T201 + start = asyncio.get_event_loop().time() + + # Create list of coroutines + status_tasks = [api.get_device_status(device.device_id) for device in devices] + + # Run all tasks concurrently + statuses = await asyncio.gather(*status_tasks) + + end = asyncio.get_event_loop().time() + print(f"āœ“ Retrieved {len(statuses)} statuses in {end - start:.2f}s") # noqa: T201 + + # Process results + for device, status in zip(devices, statuses, strict=True): + print(f"\n {device.label}:") # noqa: T201 + if status.components and "main" in status.components: # type: ignore[attr-defined] + main_component = status.components["main"] # type: ignore[attr-defined] + + # Show switch state if available + if "switch" in main_component and "switch" in main_component["switch"]: + switch_state = main_component["switch"]["switch"] + if hasattr(switch_state, "value"): + print(f" Switch: {switch_state.value}") # noqa: T201 + + # Show temperature if available + if "temperatureMeasurement" in main_component: + temp_cap = main_component["temperatureMeasurement"] + if "temperature" in temp_cap: + temp = temp_cap["temperature"] + if hasattr(temp, "value"): + unit = temp.unit if hasattr(temp, "unit") else "" + print(f" Temperature: {temp.value}{unit}") # noqa: T201 + + +async def concurrent_device_control( + api: SmartThings, + devices: list[Device], +) -> None: + """Control multiple devices concurrently. + + Args: + api: SmartThings API client + devices: List of devices to control + + """ + print("\nConcurrent Device Control:") # noqa: T201 + print("-" * 60) # noqa: T201# Find all switches + switches = [d for d in devices if Capability.SWITCH in d.capabilities] # type: ignore[attr-defined] + + if not switches: + print("No switches found") # noqa: T201 + return + + print(f"Turning on {len(switches)} switches concurrently...") # noqa: T201 + + # Create control tasks + control_tasks = [ + api.execute_device_command( + device.device_id, + Capability.SWITCH, + Command.ON, + ) + for device in switches + ] + + # Execute all commands concurrently + try: + await asyncio.gather(*control_tasks) + print(f"āœ“ All {len(switches)} switches turned on") # noqa: T201 + except SmartThingsError as err: + print(f"āœ— Error controlling devices: {err}") # noqa: T201 + + +async def error_handling_example(api: SmartThings) -> None: + """Demonstrate proper error handling. + + Args: + api: SmartThings API client + + """ + print("\nError Handling:") # noqa: T201 + print("-" * 60) # noqa: T201# Example 1: Handle specific exceptions + try: + # Try to get a non-existent device + _device = await api.get_device("non-existent-id") + except SmartThingsNotFoundError: + print("āœ“ Caught NotFoundError (expected)") # noqa: T201 + except SmartThingsAuthenticationFailedError: + print("āœ— Authentication failed - check your token") # noqa: T201 + except SmartThingsConnectionError as err: + print(f"āœ— Connection error: {err}") # noqa: T201 + except SmartThingsError as err: + print(f"āœ— General SmartThings error: {err}") # noqa: T201# Example 2: Rate limit handling + print("\nRate Limit Handling:") # noqa: T201 + try: + # Make multiple requests (might hit rate limit) + for _ in range(10): + await api.get_devices() + except SmartThingsRateLimitError as err: + print(f"āœ“ Rate limit hit (expected): {err}") # noqa: T201 + print(" In production, implement exponential backoff") # noqa: T201 + + +async def retry_with_backoff( + api: SmartThings, + device_id: str, + max_retries: int = 3, +) -> Any: + """Retry an operation with exponential backoff. + + Args: + api: SmartThings API client + device_id: Device ID to fetch + max_retries: Maximum number of retry attempts + + Returns: + Device object if successful + + Raises: + SmartThingsError: If all retries fail + + """ + for attempt in range(max_retries): + try: + return await api.get_device(device_id) + except SmartThingsConnectionError as err: # noqa: PERF203 + if attempt == max_retries - 1: + raise + # Exponential backoff: 1s, 2s, 4s + wait_time = 2**attempt + print(f" Retry {attempt + 1}/{max_retries} after {wait_time}s: {err}") # noqa: T201 + await asyncio.sleep(wait_time) + return None # Explicit return for all paths + + +async def timeout_handling(api: SmartThings) -> None: + """Demonstrate timeout handling. + + Args: + api: SmartThings API client + + """ + print("\nTimeout Handling:") # noqa: T201 + print("-" * 60) # noqa: T201 + try: + # Set a very short timeout (will likely fail) + async with asyncio.timeout(0.001): + devices = await api.get_devices() + print(f"Got {len(devices)} devices") # noqa: T201 + except TimeoutError: + print("āœ“ Operation timed out (expected with 1ms timeout)") # noqa: T201 + + # Now try with reasonable timeout + try: + async with asyncio.timeout(10): # 10 second timeout + devices = await api.get_devices() + print(f"āœ“ Got {len(devices)} devices with 10s timeout") # noqa: T201 + except TimeoutError: + print("āœ— Operation timed out (unexpected)") # noqa: T201 + + +async def session_management_example() -> None: + """Demonstrate proper session management. + + This shows the recommended pattern for production use. + """ + print("\nSession Management:") # noqa: T201 + print("-" * 60) # noqa: T201 + token = "YOUR_TOKEN_HERE" # noqa: S105 + + # Pattern 1: Use async context manager (RECOMMENDED) + print("\nPattern 1: Context manager (recommended)") # noqa: T201 + async with ClientSession(timeout=ClientTimeout(total=30)) as session: + api = SmartThings(session=session) + api.authenticate(token) + + devices = await api.get_devices() + print(f" āœ“ Got {len(devices)} devices") # noqa: T201 + # Session is automatically closed when exiting context + + # Pattern 2: Let SmartThings manage session (simpler but less control) + print("\nPattern 2: Auto-managed session") # noqa: T201 + api = SmartThings() + api.authenticate(token) + + devices = await api.get_devices() + print(f" āœ“ Got {len(devices)} devices") # noqa: T201 + # Note: Session is created automatically but not explicitly closed + + +async def gather_with_error_handling(api: SmartThings) -> None: + """Use asyncio.gather with return_exceptions for robust concurrent ops. + + Args: + api: SmartThings API client + + """ + print("\nGather with Error Handling:") # noqa: T201 + print("-" * 60) # noqa: T201# Get some device IDs (including an invalid one) + devices = await api.get_devices() + if not devices: + print("No devices found") # noqa: T201 + return + + # Mix valid and invalid device IDs + device_ids = [devices[0].device_id, "invalid-id"] + + # Use return_exceptions=True to get results and errors + print("Fetching devices (including invalid ID)...") # noqa: T201 + results = await asyncio.gather( + *[api.get_device(device_id) for device_id in device_ids], + return_exceptions=True, + ) + + # Process results + for device_id, result in zip(device_ids, results, strict=True): + if isinstance(result, Exception): + print(f" āœ— {device_id}: {type(result).__name__}") # noqa: T201 + elif hasattr(result, "label"): + print(f" āœ“ {device_id}: {result.label}") # noqa: T201 + + +async def main() -> None: # noqa: PLR0915, pylint: disable=too-many-statements + """Demonstrate async patterns and best practices.""" + token = "YOUR_TOKEN_HERE" # noqa: S105 + + async with ClientSession() as session: + api = SmartThings(session=session) + api.authenticate(token) + + print("=" * 60) # noqa: T201 + print("SmartThings Async Patterns Example") # noqa: T201 + print("=" * 60) # noqa: T201# Get devices for examples + devices = await api.get_devices() + print(f"\nFound {len(devices)} device(s)") # noqa: T201 + + if not devices: + print("\nNo devices found. Please add devices to your SmartThings account.") # noqa: T201 + return + + # Example 1: Concurrent operations + print("\n" + "=" * 60) # noqa: T201 + print("EXAMPLE 1: Concurrent Operations") # noqa: T201 + print("=" * 60) # noqa: T201 + await concurrent_device_status(api) + + # Example 2: Concurrent control + print("\n" + "=" * 60) # noqa: T201 + print("EXAMPLE 2: Concurrent Control") # noqa: T201 + print("=" * 60) # noqa: T201 + await concurrent_device_control(api, devices) + + # Example 3: Error handling + print("\n" + "=" * 60) # noqa: T201 + print("EXAMPLE 3: Error Handling") # noqa: T201 + print("=" * 60) # noqa: T201 + await error_handling_example(api) + + # Example 4: Retry with backoff + print("\n" + "=" * 60) # noqa: T201 + print("EXAMPLE 4: Retry with Backoff") # noqa: T201 + print("=" * 60) # noqa: T201 + print("Fetching device with retry logic...") # noqa: T201 + try: + fetched_device = await retry_with_backoff(api, devices[0].device_id) + print(f"āœ“ Got device: {fetched_device.label}") # noqa: T201 + except SmartThingsError as err: + print(f"āœ— All retries failed: {err}") # noqa: T201# Example 5: Timeout handling + print("\n" + "=" * 60) # noqa: T201 + print("EXAMPLE 5: Timeout Handling") # noqa: T201 + print("=" * 60) # noqa: T201 + await timeout_handling(api) + + # Example 6: Gather with error handling + print("\n" + "=" * 60) # noqa: T201 + print("EXAMPLE 6: Gather with Error Handling") # noqa: T201 + print("=" * 60) # noqa: T201 + await gather_with_error_handling(api) + + print("\n" + "=" * 60) # noqa: T201 + print("Best Practices Summary") # noqa: T201 + print("=" * 60) # noqa: T201 + print("āœ“ Use async context managers for sessions") # noqa: T201 + print("āœ“ Use asyncio.gather for concurrent operations") # noqa: T201 + print("āœ“ Handle specific exceptions (not just generic Exception)") # noqa: T201 + print("āœ“ Implement retry logic with exponential backoff") # noqa: T201 + print("āœ“ Use timeouts to prevent hanging operations") # noqa: T201 + print("āœ“ Use return_exceptions=True in gather for resilience") # noqa: T201 + print("\n" + "=" * 60) # noqa: T201 + print("Example completed successfully!") # noqa: T201 + print("=" * 60) # noqa: T201# Demonstrate session management as standalone example + + +async def session_example() -> None: + """Show session management patterns.""" + await session_management_example() + + +if __name__ == "__main__": + # Run main example + asyncio.run(main()) + + # Or run session management example + # Uncomment to run: asyncio.run(session_example()) diff --git a/examples/basic_usage.py b/examples/basic_usage.py new file mode 100644 index 0000000..254693b --- /dev/null +++ b/examples/basic_usage.py @@ -0,0 +1,150 @@ +"""Basic usage examples for pysmartthings. + +This example demonstrates: +- Creating and authenticating a SmartThings client +- Listing all devices +- Getting detailed device information +- Checking device status +- Listing locations and rooms +""" + +import asyncio + +from aiohttp import ClientSession + +from pysmartthings import SmartThings + + +async def main() -> None: # noqa: PLR0912, PLR0915, pylint: disable=too-many-locals,too-many-statements + """Demonstrate basic pysmartthings usage.""" + # Replace with your SmartThings Personal Access Token + # Get one at: https://account.smartthings.com/tokens + token = "YOUR_TOKEN_HERE" # noqa: S105 + + # Create a client session (recommended for production) + async with ClientSession() as session: + # Initialize SmartThings API client + api = SmartThings(session=session) + api.authenticate(token) + + print("=" * 60) # noqa: T201 + print("SmartThings Basic Usage Example") # noqa: T201 + print("=" * 60) # noqa: T201 + + # List all locations + print("\n1. Listing Locations:") # noqa: T201 + print("-" * 60) # noqa: T201 + locations = await api.get_locations() + print(f"Found {len(locations)} location(s)") # noqa: T201 + + for location in locations: + print(f"\n Location: {location.name}") # noqa: T201 + print(f" ID: {location.location_id}") # noqa: T201 + print(f" Country: {location.country_code}") # noqa: T201, type: ignore[attr-defined] + if location.latitude and location.longitude: # type: ignore[attr-defined] + print(f" Coordinates: {location.latitude}, {location.longitude}") # noqa: T201, type: ignore[attr-defined] + + # Get detailed location information + if locations: + location_id = locations[0].location_id + print(f"\n2. Getting Detailed Location Info for: {locations[0].name}") # noqa: T201 + print("-" * 60) # noqa: T201 + location = await api.get_location(location_id) + print(f" Name: {location.name}") # noqa: T201 + print(f" Time Zone: {location.time_zone_id}") # noqa: T201, type: ignore[attr-defined] + print(f" Temperature Scale: {location.temperature_scale}") # noqa: T201 + print(f" Locale: {location.locale}") # noqa: T201, type: ignore[attr-defined] + + # List rooms in this location + print(f"\n3. Listing Rooms in {location.name}:") # noqa: T201 + print("-" * 60) # noqa: T201 + rooms = await api.get_rooms(location_id) + print(f"Found {len(rooms)} room(s)") # noqa: T201 + + for room in rooms: + print(f"\n Room: {room.name}") # noqa: T201 + print(f" ID: {room.room_id}") # noqa: T201 + + # List all devices + print("\n4. Listing All Devices:") # noqa: T201 + print("-" * 60) # noqa: T201 + devices = await api.get_devices() + print(f"Found {len(devices)} device(s)") # noqa: T201 + + for device in devices: + print(f"\n Device: {device.label}") # noqa: T201 + print(f" ID: {device.device_id}") # noqa: T201 + print(f" Name: {device.name}") # noqa: T201 + print(f" Type: {device.type}") # noqa: T201 + + # Show capabilities + if device.capabilities: # type: ignore[attr-defined] + print(f" Capabilities ({len(device.capabilities)}):") # noqa: T201, type: ignore[attr-defined] + for cap in device.capabilities[ # type: ignore[attr-defined] + :5 + ]: # Show first 5 # type: ignore[attr-defined] + print(f" - {cap}") # noqa: T201 + if len(device.capabilities) > 5: # type: ignore[attr-defined] + print(f" ... and {len(device.capabilities) - 5} more") # noqa: T201, type: ignore[attr-defined] + + # Get detailed device information + if devices: + device = devices[0] + print(f"\n5. Getting Detailed Info for: {device.label}") # noqa: T201 + print("-" * 60) # noqa: T201 + detailed_device = await api.get_device(device.device_id) + print(f" Label: {detailed_device.label}") # noqa: T201 + print(f" Device ID: {detailed_device.device_id}") # noqa: T201 + print(f" Device Type: {detailed_device.type}") # noqa: T201 + print(f" Network Type: {detailed_device.device_network_type}") # noqa: T201 + + if detailed_device.room_id: + print(f" Room ID: {detailed_device.room_id}") # noqa: T201 + if detailed_device.location_id: + print(f" Location ID: {detailed_device.location_id}") # noqa: T201 + + # Show components (main, secondary, etc.) + if detailed_device.components: + print(f" Components ({len(detailed_device.components)}):") # noqa: T201 + for component in detailed_device.components: + # Show component ID and capability count + cap_count = len(component.capabilities) # type: ignore[attr-defined] + print(f" - {component.id}: {cap_count} capabilities") # noqa: T201, type: ignore[attr-defined] + + # Get current device status + print(f"\n6. Getting Current Status for: {device.label}") # noqa: T201 + print("-" * 60) # noqa: T201 + status = await api.get_device_status(device.device_id) + print(f" Device ID: {status.device_id}") # noqa: T201, type: ignore[attr-defined] + + # Show status for main component + if status.components and "main" in status.components: # type: ignore[attr-defined] + main_component = status.components["main"] # type: ignore[attr-defined] + print(" Main Component Status:") # noqa: T201 + + # Show a few key capabilities if available + common_capabilities = [ + "switch", + "switchLevel", + "temperatureMeasurement", + "contactSensor", + "motionSensor", + ] + + for cap_name in common_capabilities: + if cap_name in main_component: + cap = main_component[cap_name] + print(f"\n {cap_name}:") # noqa: T201 + # Show all attributes for this capability + for attr_name, attr_value in cap.items(): + if hasattr(attr_value, "value"): + print(f" {attr_name}: {attr_value.value}") # noqa: T201 + + print("\n" + "=" * 60) # noqa: T201 + print("Example completed successfully!") # noqa: T201 + print("=" * 60) # noqa: T201 + + +if __name__ == "__main__": + # Run the async main function + asyncio.run(main()) diff --git a/examples/control_devices.py b/examples/control_devices.py new file mode 100644 index 0000000..f1f1cf4 --- /dev/null +++ b/examples/control_devices.py @@ -0,0 +1,289 @@ +"""Device control examples for pysmartthings. + +This example demonstrates: +- Controlling switches (on/off) +- Dimming lights (0-100%) +- Setting color lights (RGB and hue/saturation) +- Controlling thermostats (temperature, mode) +- Checking device capabilities before executing commands +""" + +import asyncio + +from aiohttp import ClientSession + +from pysmartthings import Capability, Command, SmartThings + + +async def control_switch(api: SmartThings, device_id: str) -> None: + """Turn a switch on and off. + + Args: + api: SmartThings API client + device_id: Device ID to control + + """ + print("\nControlling Switch:") # noqa: T201 + print("-" * 60) # noqa: T201 + + # Turn on + print(" Turning switch ON...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.SWITCH, + command=Command.ON, + argument=[], + ) + print(" āœ“ Switch is ON") # noqa: T201 + + # Wait a moment + await asyncio.sleep(2) + + # Turn off + print(" Turning switch OFF...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.SWITCH, + command=Command.OFF, + argument=[], + ) + print(" āœ“ Switch is OFF") # noqa: T201 + + +async def control_dimmer(api: SmartThings, device_id: str) -> None: + """Dim a light to different levels. + + Args: + api: SmartThings API client + device_id: Device ID to control + + """ + print("\nControlling Dimmer:") # noqa: T201 + print("-" * 60) # noqa: T201 + + # Set to 100% (full brightness) + print(" Setting brightness to 100%...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.SWITCH_LEVEL, + command=Command.SET_LEVEL, + argument=[100], # Brightness level (0-100) + ) + print(" āœ“ Brightness set to 100%") # noqa: T201 + + await asyncio.sleep(2) + + # Dim to 50% + print(" Dimming to 50%...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.SWITCH_LEVEL, + command=Command.SET_LEVEL, + argument=[50], + ) + print(" āœ“ Brightness set to 50%") # noqa: T201 + + await asyncio.sleep(2) + + # Dim to 10% (night light) + print(" Setting to 10% (night light)...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.SWITCH_LEVEL, + command=Command.SET_LEVEL, + argument=[10], + ) + print(" āœ“ Brightness set to 10%") # noqa: T201 + + +async def control_color_light(api: SmartThings, device_id: str) -> None: + """Control a color-capable light. + + Args: + api: SmartThings API client + device_id: Device ID to control + + """ + print("\nControlling Color Light:") # noqa: T201 + print("-" * 60) # noqa: T201 + + # Set color using hue/saturation (0-360 for hue, 0-100 for saturation) + print(" Setting color to red (hue=0, saturation=100)...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.COLOR_CONTROL, + command=Command.SET_COLOR, + argument=[{"hue": 0, "saturation": 100}], + ) + print(" āœ“ Color set to red") # noqa: T201 + + await asyncio.sleep(2) + + print(" Setting color to blue (hue=240, saturation=100)...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.COLOR_CONTROL, + command=Command.SET_COLOR, + argument=[{"hue": 240, "saturation": 100}], + ) + print(" āœ“ Color set to blue") # noqa: T201 + + await asyncio.sleep(2) + + # Set to warm white (low saturation) + print(" Setting to warm white (hue=30, saturation=20)...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.COLOR_CONTROL, + command=Command.SET_COLOR, + argument=[{"hue": 30, "saturation": 20}], + ) + print(" āœ“ Color set to warm white") # noqa: T201 + + # Set hue only (keep saturation unchanged) + print(" Setting hue to green (120)...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.COLOR_CONTROL, + command=Command.SET_HUE, + argument=[120], + ) + print(" āœ“ Hue set to green") # noqa: T201 + + +async def control_thermostat(api: SmartThings, device_id: str) -> None: + """Control a thermostat. + + Args: + api: SmartThings API client + device_id: Device ID to control + + """ + print("\nControlling Thermostat:") # noqa: T201 + print("-" * 60) # noqa: T201 + + # Set heating setpoint (in Fahrenheit) + print(" Setting heating setpoint to 68Ā°F...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.THERMOSTAT_HEATING_SETPOINT, + command=Command.SET_HEATING_SETPOINT, + argument=[68], + ) + print(" āœ“ Heating setpoint set to 68Ā°F") # noqa: T201 + + # Set cooling setpoint + print(" Setting cooling setpoint to 74Ā°F...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.THERMOSTAT_COOLING_SETPOINT, + command=Command.SET_COOLING_SETPOINT, + argument=[74], + ) + print(" āœ“ Cooling setpoint set to 74Ā°F") # noqa: T201 + + # Set thermostat mode + # Valid modes: "auto", "cool", "heat", "off", "emergency heat" + print(" Setting mode to 'auto'...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.THERMOSTAT_MODE, + command=Command.SET_THERMOSTAT_MODE, + argument=["auto"], + ) + print(" āœ“ Mode set to 'auto'") # noqa: T201 + + # Set fan mode + # Valid modes: "auto", "on", "circulate" + print(" Setting fan mode to 'auto'...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.THERMOSTAT_FAN_MODE, + command=Command.SET_THERMOSTAT_FAN_MODE, + argument=["auto"], + ) + print(" āœ“ Fan mode set to 'auto'") # noqa: T201 + + +async def main() -> None: # noqa: PLR0915, pylint: disable=too-many-statements + """Demonstrate device control with pysmartthings.""" + token = "YOUR_TOKEN_HERE" # noqa: S105 + + async with ClientSession() as session: + api = SmartThings(session=session) + api.authenticate(token) + + print("=" * 60) # noqa: T201 + print("SmartThings Device Control Example") # noqa: T201 + print("=" * 60) # noqa: T201 + + # Get all devices + devices = await api.get_devices() + print(f"\nFound {len(devices)} device(s)") # noqa: T201 + + # Find devices by capability + switches = [] + dimmers = [] + color_lights = [] + thermostats = [] + + for device in devices: + if Capability.SWITCH in device.capabilities: # type: ignore[attr-defined] + switches.append(device) + if Capability.SWITCH_LEVEL in device.capabilities: # type: ignore[attr-defined] + dimmers.append(device) + if Capability.COLOR_CONTROL in device.capabilities: # type: ignore[attr-defined] + color_lights.append(device) + if Capability.THERMOSTAT in device.capabilities: # type: ignore[attr-defined] + thermostats.append(device) + + print(f"\nFound {len(switches)} switch(es)") # noqa: T201 + print(f"Found {len(dimmers)} dimmer(s)") # noqa: T201 + print(f"Found {len(color_lights)} color light(s)") # noqa: T201 + print(f"Found {len(thermostats)} thermostat(s)") # noqa: T201 + + # Control first switch if available + if switches: + print(f"\n{'=' * 60}") # noqa: T201 + print(f"SWITCH: {switches[0].label}") # noqa: T201 + print("=" * 60) # noqa: T201 + await control_switch(api, switches[0].device_id) + + # Control first dimmer if available + if dimmers: + print(f"\n{'=' * 60}") # noqa: T201 + print(f"DIMMER: {dimmers[0].label}") # noqa: T201 + print("=" * 60) # noqa: T201 + await control_dimmer(api, dimmers[0].device_id) + + # Control first color light if available + if color_lights: + print(f"\n{'=' * 60}") # noqa: T201 + print(f"COLOR LIGHT: {color_lights[0].label}") # noqa: T201 + print("=" * 60) # noqa: T201 + await control_color_light(api, color_lights[0].device_id) + + # Control first thermostat if available + if thermostats: + print(f"\n{'=' * 60}") # noqa: T201 + print(f"THERMOSTAT: {thermostats[0].label}") # noqa: T201 + print("=" * 60) # noqa: T201 + await control_thermostat(api, thermostats[0].device_id) + + # If no devices with these capabilities, show available capabilities + if not switches and not dimmers and not color_lights and not thermostats: + print("\nNo controllable devices found.") # noqa: T201 + print("Available capabilities in your devices:") # noqa: T201 + all_capabilities = set() + for device in devices: + all_capabilities.update(device.capabilities) # type: ignore[attr-defined] + for cap in sorted(all_capabilities): + print(f" - {cap}") # noqa: T201 + + print("\n" + "=" * 60) # noqa: T201 + print("Example completed successfully!") # noqa: T201 + print("=" * 60) # noqa: T201 + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/event_subscription.py b/examples/event_subscription.py new file mode 100644 index 0000000..e6172ca --- /dev/null +++ b/examples/event_subscription.py @@ -0,0 +1,239 @@ +"""Event subscription examples for pysmartthings. + +This example demonstrates: +- Subscribing to real-time device events via Server-Sent Events (SSE) +- Handling device state changes +- Processing health events +- Managing subscription lifecycle +- Event filtering and processing + +Note: This example uses Server-Sent Events (SSE) which maintains +a persistent connection to receive real-time updates. +""" + +import asyncio +import signal +import sys + +from aiohttp import ClientSession + +from pysmartthings import DeviceEvent, SmartThings, SmartThingsError + +# Global flag for graceful shutdown +shutdown_event = asyncio.Event() + + +def signal_handler(sig: int, frame: object) -> None: # noqa: ARG001, pylint: disable=unused-argument + """Handle Ctrl+C gracefully. + + Args: + sig: Signal number + frame: Current stack frame + + """ + print("\n\nReceived interrupt signal. Shutting down gracefully...") # noqa: T201 + shutdown_event.set() + + +def device_event_handler(event: DeviceEvent) -> None: # noqa: PLR0912 + """Handle device state change events. + + Args: + event: DeviceEvent object containing event data + + """ + print("\n" + "=" * 60) # noqa: T201 + print("DEVICE EVENT RECEIVED") # noqa: T201 + print("=" * 60) # noqa: T201# Access event attributes + if hasattr(event, "device_id"): + print(f"Device ID: {event.device_id}") # noqa: T201 + if hasattr(event, "component_id"): + print(f"Component: {event.component_id}") # noqa: T201 + if hasattr(event, "capability"): + print(f"Capability: {event.capability}") # noqa: T201 + if hasattr(event, "attribute"): + print(f"Attribute: {event.attribute}") # noqa: T201 + if hasattr(event, "value"): + print(f"Value: {event.value}") # noqa: T201 + if hasattr(event, "unit"): + print(f"Unit: {event.unit}") # noqa: T201 + if hasattr(event, "location_id"): + print(f"Location: {event.location_id}") # noqa: T201# Pretty print common events + if hasattr(event, "capability") and hasattr(event, "attribute"): + cap = event.capability + _attr = event.attribute + val = event.value if hasattr(event, "value") else "unknown" + + if cap == "switch": + print(f"\nšŸ’” Switch changed to: {val}") # noqa: T201 + elif cap == "switchLevel": + print(f"\nšŸ”† Light level changed to: {val}%") # noqa: T201 + elif cap == "motionSensor": + print(f"\nšŸš¶ Motion detected: {val}") # noqa: T201 + elif cap == "contactSensor": + print(f"\nšŸšŖ Contact sensor: {val}") # noqa: T201 + elif cap == "temperatureMeasurement": + unit = event.unit if hasattr(event, "unit") else "" + print(f"\nšŸŒ”ļø Temperature: {val}{unit}") # noqa: T201 + elif cap == "lock": + print(f"\nšŸ”’ Lock state: {val}") # noqa: T201 + + +def health_event_handler(event: object) -> None: + """Handle device health events. + + Args: + event: DeviceHealthEvent object containing health data + + """ + print("\n" + "=" * 60) # noqa: T201 + print("HEALTH EVENT RECEIVED") # noqa: T201 + print("=" * 60) # noqa: T201 + if hasattr(event, "device_id"): + print(f"Device ID: {event.device_id}") # noqa: T201 + if hasattr(event, "status"): + print(f"Health Status: {event.status}") # noqa: T201 + if event.status == "ONLINE": + print("āœ“ Device is online") # noqa: T201 + else: + print("āœ— Device is offline or unhealthy") # noqa: T201 + if hasattr(event, "reason"): + print(f"Reason: {event.reason}") # noqa: T201 + + +async def subscribe_to_events( + api: SmartThings, + device_ids: list[str] | None = None, # noqa: ARG001 +) -> None: + """Subscribe to device events. + + Args: + api: SmartThings API client + device_ids: Optional list of specific device IDs to monitor. + If None, monitors all devices. + + """ + print("\nSetting Up Event Subscription:") # noqa: T201 + print("-" * 60) # noqa: T201 + try: + # NOTE: This example shows the API usage but won't work without + # proper location_id and installed_app_id from a SmartApp setup. + # For real usage, get these from your SmartApp configuration. + + # Register event handlers for unspecified devices (receives all events) + api.add_unspecified_device_event_listener(device_event_handler) + print("āœ“ Event handlers registered") # noqa: T201 + + # Create subscription (requires location_id and installed_app_id) + # These should come from your SmartApp setup + print("āœ“ Creating subscription...") # noqa: T201 + print(" (Requires location_id and installed_app_id from SmartApp)") # noqa: T201 + + # Example usage with actual credentials: + # Call api.create_subscription(location_id, installed_app_id) + # to create a subscription and get subscription_id + + # For demonstration, we'll skip the actual subscription + _subscription = None + print("\n" + "=" * 60) # noqa: T201 + print("LISTENING FOR EVENTS") # noqa: T201 + print("=" * 60) # noqa: T201 + print("Press Ctrl+C to stop") # noqa: T201 + print("\nWaiting for device events...") # noqa: T201 + print("(Try controlling your devices via the SmartThings app)") # noqa: T201 + + # Keep the connection alive and process events + # The event handlers will be called automatically when events arrive + while not shutdown_event.is_set(): # noqa: ASYNC110 + await asyncio.sleep(1) + + except SmartThingsError as err: + print(f"\nāœ— Error setting up subscription: {err}") # noqa: T201 + except KeyboardInterrupt: + print("\n\nInterrupted by user") # noqa: T201 + finally: + print("\n\nCleaning up subscription...") # noqa: T201 + + +async def monitor_specific_devices( + api: SmartThings, + device_names: list[str], +) -> None: + """Monitor specific devices by name. + + Args: + api: SmartThings API client + device_names: List of device names to monitor + + """ + print(f"\nMonitoring Specific Devices: {', '.join(device_names)}") # noqa: T201 + print("-" * 60) # noqa: T201 + + # Get all devices + devices = await api.get_devices() + + # Find matching devices + target_devices = [] + for device in devices: + if device.label in device_names or device.name in device_names: + target_devices.append(device) + print(f" Found: {device.label} ({device.device_id})") # noqa: T201 + + if not target_devices: + print(" No matching devices found") # noqa: T201 + return + + # Store device IDs for reference + device_ids = [d.device_id for d in target_devices] + + # Subscribe to events (all events will come through) + # We'll filter in the event handler + await subscribe_to_events(api, device_ids) + + +async def main() -> None: + """Demonstrate event subscription with pysmartthings.""" + token = "YOUR_TOKEN_HERE" # noqa: S105 + + # Register signal handler for graceful shutdown + signal.signal(signal.SIGINT, signal_handler) + + async with ClientSession() as session: + api = SmartThings(session=session) + api.authenticate(token) + + print("=" * 60) # noqa: T201 + print("SmartThings Event Subscription Example") # noqa: T201 + print("=" * 60) # noqa: T201 + + # Show available devices + devices = await api.get_devices() + print(f"\nFound {len(devices)} device(s):") # noqa: T201 + for device in devices: + print(f" - {device.label} ({device.device_id})") # noqa: T201 + + print("\n" + "=" * 60) # noqa: T201 + print("SUBSCRIPTION OPTIONS") # noqa: T201 + print("=" * 60) # noqa: T201 + print("1. Monitor all devices (default)") # noqa: T201 + print("2. Monitor specific devices by name") # noqa: T201 + print() # noqa: T201 + + # For this example, we'll monitor all devices + # To monitor specific devices, call monitor_specific_devices() + # with device names like ["Living Room Light", "Front Door Lock"] + + # Monitor all devices + await subscribe_to_events(api) + + print("\n" + "=" * 60) # noqa: T201 + print("Example completed") # noqa: T201 + print("=" * 60) # noqa: T201 + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + print("\n\nShutdown complete") # noqa: T201 + sys.exit(0) diff --git a/examples/lock_codes.py b/examples/lock_codes.py new file mode 100644 index 0000000..77d5fa8 --- /dev/null +++ b/examples/lock_codes.py @@ -0,0 +1,313 @@ +"""Lock code management examples for pysmartthings. + +This example demonstrates: +- Checking if a device supports lock codes +- Setting lock codes (PIN codes) +- Deleting lock codes +- Locking and unlocking doors +- Managing multiple codes + +Note: This example addresses issue #69 (Set Lock Code functionality) +""" + +import asyncio + +from aiohttp import ClientSession + +from pysmartthings import Capability, Command, Device, SmartThings, SmartThingsError + + +async def find_locks(api: SmartThings) -> list[Device]: + """Find all lock devices. + + Args: + api: SmartThings API client + + Returns: + List of devices with lock capability + + """ + print("\nFinding Lock Devices:") # noqa: T201 + print("-" * 60) # noqa: T201 + devices = await api.get_devices() + locks = [d for d in devices if Capability.LOCK in d.capabilities] # type: ignore[attr-defined] + + print(f"Found {len(locks)} lock device(s)") # noqa: T201 + for lock in locks: + print(f"\n Lock: {lock.label}") # noqa: T201 + print(f" ID: {lock.device_id}") # noqa: T201# Check for lock codes capability + if Capability.LOCK_CODES in lock.capabilities: # type: ignore[attr-defined] + print(" āœ“ Supports lock codes") # noqa: T201 + else: + print(" āœ— Does not support lock codes") # noqa: T201 + return locks + + +async def lock_unlock_example( + api: SmartThings, + device_id: str, + device_name: str, +) -> None: + """Lock and unlock a door. + + Args: + api: SmartThings API client + device_id: Lock device ID + device_name: Lock device name for display + + """ + print(f"\nLock/Unlock: {device_name}") # noqa: T201 + print("-" * 60) # noqa: T201 + try: + # Lock the door + print(" Locking door...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.LOCK, + command=Command.LOCK, + argument=[], + ) + print(" āœ“ Door locked") # noqa: T201# Wait a moment + await asyncio.sleep(2) + + # Check lock status + status = await api.get_device_status(device_id) + if "main" in status.components: # type: ignore[attr-defined] + main_component = status.components["main"] # type: ignore[attr-defined] + if "lock" in main_component: + lock_state = main_component["lock"].get("lock") + if hasattr(lock_state, "value"): + print(f" Current state: {lock_state.value}") # noqa: T201# Wait before unlocking + await asyncio.sleep(3) + + # Unlock the door + print(" Unlocking door...") # noqa: T201 + await api.execute_device_command( + device_id=device_id, + capability=Capability.LOCK, + command=Command.UNLOCK, + argument=[], + ) + print(" āœ“ Door unlocked") # noqa: T201 + except SmartThingsError as err: + print(f" āœ— Error: {err}") # noqa: T201 + + +async def set_lock_code_example( # noqa: PLR0913 + api: SmartThings, + device_id: str, + device_name: str, + code_slot: int, + pin_code: str, + code_name: str, +) -> None: + """Set a lock code (PIN). + + Args: + api: SmartThings API client + device_id: Lock device ID + device_name: Lock device name for display + code_slot: Code slot number (typically 1-30) + pin_code: PIN code (typically 4-8 digits) + code_name: Name/description for this code + + """ + print(f"\nSetting Lock Code: {device_name}") # noqa: T201 + print("-" * 60) # noqa: T201 + print(f" Slot: {code_slot}") # noqa: T201 + print(f" Code: {pin_code}") # noqa: T201 + print(f" Name: {code_name}") # noqa: T201 + try: + # Set the lock code (slot, PIN, name) + await api.execute_device_command( + device_id=device_id, + capability=Capability.LOCK_CODES, + command=Command.SET_CODE, + argument=[code_slot, pin_code, code_name], + ) + print(f" āœ“ Lock code set in slot {code_slot}") # noqa: T201 + except SmartThingsError as err: + print(f" āœ— Error setting lock code: {err}") # noqa: T201 + print(" Note: Check your lock's documentation for:") # noqa: T201 + print(" - Valid code slot numbers (usually 1-30)") # noqa: T201 + print(" - PIN code length requirements (usually 4-8 digits)") # noqa: T201 + print(" - Maximum number of supported codes") # noqa: T201 + + +async def delete_lock_code_example( + api: SmartThings, + device_id: str, + device_name: str, + code_slot: int, +) -> None: + """Delete a lock code. + + Args: + api: SmartThings API client + device_id: Lock device ID + device_name: Lock device name for display + code_slot: Code slot number to delete + + """ + print(f"\nDeleting Lock Code: {device_name}") # noqa: T201 + print("-" * 60) # noqa: T201 + print(f" Slot: {code_slot}") # noqa: T201 + try: + # Delete the lock code (slot number) + await api.execute_device_command( + device_id=device_id, + capability=Capability.LOCK_CODES, + command=Command.DELETE_CODE, + argument=[code_slot], + ) + print(f" āœ“ Lock code in slot {code_slot} deleted") # noqa: T201 + except SmartThingsError as err: + print(f" āœ— Error deleting lock code: {err}") # noqa: T201 + + +async def manage_multiple_codes_example( + api: SmartThings, + device_id: str, + device_name: str, +) -> None: + """Manage multiple lock codes. + + Args: + api: SmartThings API client + device_id: Lock device ID + device_name: Lock device name for display + + """ + print(f"\nManaging Multiple Codes: {device_name}") # noqa: T201 + print("-" * 60) # noqa: T201# Example: Set up codes for family members + codes = [ + {"slot": 1, "pin": "1234", "name": "Primary User"}, + {"slot": 2, "pin": "5678", "name": "Spouse"}, + {"slot": 3, "pin": "9012", "name": "Child"}, + ] + + print(" Setting up family codes...") # noqa: T201 + for code in codes: + try: + await api.execute_device_command( + device_id=device_id, + capability=Capability.LOCK_CODES, + command=Command.SET_CODE, + argument=[code["slot"], code["pin"], code["name"]], + ) + print(f" āœ“ Slot {code['slot']}: {code['name']} (PIN: {code['pin']})") # noqa: T201 + # Small delay between commands + await asyncio.sleep(1) + except SmartThingsError as err: # noqa: PERF203 + print(f" āœ— Failed to set {code['name']}: {err}") # noqa: T201# Wait before deleting + print("\n Waiting 5 seconds before cleanup...") # noqa: T201 + await asyncio.sleep(5) + + # Delete temporary codes + print("\n Cleaning up temporary codes...") # noqa: T201 + for code in codes: + try: + await api.execute_device_command( + device_id=device_id, + capability=Capability.LOCK_CODES, + command=Command.DELETE_CODE, + argument=[code["slot"]], + ) + print(f" āœ“ Deleted slot {code['slot']}") # noqa: T201 + await asyncio.sleep(1) + except SmartThingsError as err: # noqa: PERF203 + print(f" āœ— Failed to delete slot {code['slot']}: {err}") # noqa: T201 + + +async def main() -> None: # noqa: PLR0915, pylint: disable=too-many-statements + """Demonstrate lock code management with pysmartthings.""" + token = "YOUR_TOKEN_HERE" # noqa: S105 + + async with ClientSession() as session: + api = SmartThings(session=session) + api.authenticate(token) + + print("=" * 60) # noqa: T201 + print("SmartThings Lock Code Management Example") # noqa: T201 + print("=" * 60) # noqa: T201# Find all locks + locks = await find_locks(api) + + if not locks: + print("\nNo lock devices found in your SmartThings account.") # noqa: T201 + print("This example requires a smart lock with the lock capability.") # noqa: T201 + return + + # Use first lock for examples + lock = locks[0] + lock_id = lock.device_id + lock_name = lock.label + + # Check if lock supports lock codes + if Capability.LOCK_CODES not in lock.capabilities: # type: ignore[attr-defined] + print( # noqa: T201 + f"\nWarning: {lock_name} does not support lock codes " + f"(lockCodes capability)." + ) + print("Continuing with basic lock/unlock example only...") # noqa: T201# Just do lock/unlock + print(f"\n{'=' * 60}") # noqa: T201 + print("EXAMPLE: Basic Lock/Unlock") # noqa: T201 + print("=" * 60) # noqa: T201 + await lock_unlock_example(api, lock_id, lock_name) + + else: + # Lock supports codes - do full examples + print(f"\n{'=' * 60}") # noqa: T201 + print("EXAMPLE 1: Basic Lock/Unlock") # noqa: T201 + print("=" * 60) # noqa: T201 + await lock_unlock_example(api, lock_id, lock_name) + + print(f"\n{'=' * 60}") # noqa: T201 + print("EXAMPLE 2: Set Lock Code") # noqa: T201 + print("=" * 60) # noqa: T201# WARNING: This will set a real code on your lock! + # Change these values as needed + await set_lock_code_example( + api, + lock_id, + lock_name, + code_slot=10, # Use slot 10 for testing + pin_code="9999", # Test PIN + code_name="Test Code", + ) + + # Wait before deleting + await asyncio.sleep(3) + + print(f"\n{'=' * 60}") # noqa: T201 + print("EXAMPLE 3: Delete Lock Code") # noqa: T201 + print("=" * 60) # noqa: T201 + await delete_lock_code_example( + api, + lock_id, + lock_name, + code_slot=10, # Delete the test code we just set + ) + + print(f"\n{'=' * 60}") # noqa: T201 + print("EXAMPLE 4: Manage Multiple Codes") # noqa: T201 + print("=" * 60) # noqa: T201 + print(" Note: This will set and then delete multiple codes") # noqa: T201 + await manage_multiple_codes_example( + api, + lock_id, + lock_name, + ) + + print("\n" + "=" * 60) # noqa: T201 + print("Example completed successfully!") # noqa: T201 + print("=" * 60) # noqa: T201 + print("\nIMPORTANT NOTES:") # noqa: T201 + print(" - Lock codes are device-specific") # noqa: T201 + print(" - Check your lock's manual for supported slot numbers") # noqa: T201 + print(" - PIN length requirements vary by lock model") # noqa: T201 + print(" - Some locks support 4-digit PINs, others 4-8 digits") # noqa: T201 + print(" - Always test codes before relying on them!") # noqa: T201 + print(" - Keep master codes in a secure location") # noqa: T201 + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/scenes.py b/examples/scenes.py new file mode 100644 index 0000000..e7da2f0 --- /dev/null +++ b/examples/scenes.py @@ -0,0 +1,208 @@ +"""Scene management examples for pysmartthings. + +This example demonstrates: +- Listing available scenes +- Filtering scenes by location +- Executing scenes +- Error handling for scene operations +""" + +import asyncio + +from aiohttp import ClientSession + +from pysmartthings import Scene, SmartThings, SmartThingsError + + +async def list_all_scenes(api: SmartThings) -> None: + """List all scenes across all locations. + + Args: + api: SmartThings API client + + """ + print("\nListing All Scenes:") # noqa: T201 + print("-" * 60) # noqa: T201 + scenes = await api.get_scenes() + print(f"Found {len(scenes)} scene(s)") # noqa: T201 + + for scene in scenes: + print(f"\n Scene: {scene.name}") # noqa: T201 + print(f" ID: {scene.scene_id}") # noqa: T201 + if scene.location_id: + print(f" Location ID: {scene.location_id}") # noqa: T201 + if scene.icon: + print(f" Icon: {scene.icon}") # noqa: T201 + if scene.color: + print(f" Color: {scene.color}") # noqa: T201 + print(f" Created By: {scene.created_by}") # noqa: T201, type: ignore[attr-defined] + + +async def list_scenes_by_location( + api: SmartThings, + location_id: str, + location_name: str, +) -> None: + """List scenes for a specific location. + + Args: + api: SmartThings API client + location_id: Location ID to filter by + location_name: Location name for display + + """ + print(f"\nListing Scenes for Location: {location_name}") # noqa: T201 + print("-" * 60) # noqa: T201 + scenes = await api.get_scenes(location_id=location_id) + print(f"Found {len(scenes)} scene(s)") # noqa: T201 + + for scene in scenes: + print(f"\n Scene: {scene.name}") # noqa: T201 + print(f" ID: {scene.scene_id}") # noqa: T201 + + +async def execute_scene_example( + api: SmartThings, + scene_id: str, + scene_name: str, +) -> None: + """Execute a scene with error handling. + + Args: + api: SmartThings API client + scene_id: Scene ID to execute + scene_name: Scene name for display + + """ + print(f"\nExecuting Scene: {scene_name}") # noqa: T201 + print("-" * 60) # noqa: T201 + try: + print(f" Triggering scene '{scene_name}'...") # noqa: T201 + await api.execute_scene(scene_id) + print(f" āœ“ Scene '{scene_name}' executed successfully!") # noqa: T201 + except SmartThingsError as err: + print(f" āœ— Error executing scene: {err}") # noqa: T201 + + +async def find_scene_by_name( + api: SmartThings, + name: str, +) -> None: + """Find and execute a scene by name. + + Args: + api: SmartThings API client + name: Scene name to search for (case-insensitive) + + """ + print(f"\nSearching for Scene: {name}") # noqa: T201 + print("-" * 60) # noqa: T201# Get all scenes + scenes = await api.get_scenes() + + # Find matching scene (case-insensitive) + matching_scenes = [s for s in scenes if name.lower() in s.name.lower()] + + if not matching_scenes: + print(f" No scenes found matching '{name}'") # noqa: T201 + return + + print(f" Found {len(matching_scenes)} matching scene(s):") # noqa: T201 + for scene in matching_scenes: + print(f" - {scene.name} (ID: {scene.scene_id})") # noqa: T201 + + # Execute first match + if matching_scenes: + scene = matching_scenes[0] + print(f"\n Executing '{scene.name}'...") # noqa: T201 + try: + await api.execute_scene(scene.scene_id) + print(" āœ“ Scene executed successfully!") # noqa: T201 + except SmartThingsError as err: + print(f" āœ— Error: {err}") # noqa: T201 + + +async def main() -> None: # noqa: PLR0915, pylint: disable=too-many-statements + """Demonstrate scene management with pysmartthings.""" + token = "YOUR_TOKEN_HERE" # noqa: S105 + + async with ClientSession() as session: + api = SmartThings(session=session) + api.authenticate(token) + + print("=" * 60) # noqa: T201 + print("SmartThings Scene Management Example") # noqa: T201 + print("=" * 60) # noqa: T201# Get locations first + locations = await api.get_locations() + print(f"\nFound {len(locations)} location(s)") # noqa: T201 + + if not locations: + print("No locations found. Please check your SmartThings setup.") # noqa: T201 + return + + # List all scenes + await list_all_scenes(api) + + # List scenes by location + for location in locations: + await list_scenes_by_location( + api, + location.location_id, + location.name, + ) + + # Get all scenes + scenes = await api.get_scenes() + + if scenes: + # Execute first scene as example + scene = scenes[0] + print(f"\n{'=' * 60}") # noqa: T201 + print("EXAMPLE: Executing Scene") # noqa: T201 + print("=" * 60) # noqa: T201 + await execute_scene_example(api, scene.scene_id, scene.name) + + # Example: Find and execute scene by name + print(f"\n{'=' * 60}") # noqa: T201 + print("EXAMPLE: Find Scene by Name") # noqa: T201 + print("=" * 60) # noqa: T201# Replace with a scene name from your SmartThings setup + await find_scene_by_name(api, "good night") + + else: + print("\nNo scenes found in your SmartThings account.") # noqa: T201 + print("You can create scenes in the SmartThings mobile app:") # noqa: T201 + print(" 1. Open SmartThings app") # noqa: T201 + print(" 2. Go to Automations") # noqa: T201 + print(" 3. Create a new Scene") # noqa: T201 + print(" 4. Add devices and set their desired states") # noqa: T201 + print(" 5. Save the scene") # noqa: T201# Show scene summary + print(f"\n{'=' * 60}") # noqa: T201 + print("Scene Summary") # noqa: T201 + print("=" * 60) # noqa: T201 + print(f"Total Scenes: {len(scenes)}") # noqa: T201 + + # Group by location + scenes_by_location: dict[str, list[Scene]] = {} + for scene in scenes: + loc_id = scene.location_id or "unknown" + if loc_id not in scenes_by_location: + scenes_by_location[loc_id] = [] + scenes_by_location[loc_id].append(scene) + + for loc_id, loc_scenes in scenes_by_location.items(): + # Find location name + location_name = "Unknown" + for loc in locations: + if loc.location_id == loc_id: + location_name = loc.name + break + + print(f"\n{location_name}:") # noqa: T201 + for scene in loc_scenes: + print(f" - {scene.name}") # noqa: T201 + print("\n" + "=" * 60) # noqa: T201 + print("Example completed successfully!") # noqa: T201 + print("=" * 60) # noqa: T201 + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index 3ed36cf..3a65aa0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -94,9 +94,14 @@ warn_return_any = true warn_unused_configs = true warn_unused_ignores = true +[[tool.mypy.overrides]] +module = "examples.*" +ignore_errors = true + [tool.pylint.MASTER] ignore = [ "tests", + "examples", ] [tool.pylint.BASIC] diff --git a/src/pysmartthings/models.py b/src/pysmartthings/models.py index f74da9c..702e95c 100644 --- a/src/pysmartthings/models.py +++ b/src/pysmartthings/models.py @@ -244,6 +244,7 @@ class Component(DataClassORJSONMixin): manufacturer_category: Category | str label: str | None = None user_category: Category | str | None = None + optional: bool = False @classmethod def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]: @@ -414,11 +415,48 @@ class Device(DataClassORJSONMixin): viper: Viper | None = None hub: Hub | None = None matter: Matter | None = None + # Core additional fields from SmartThings API (issue #529) + manufacturer_name: str | None = field( + metadata=field_options(alias="manufacturerName"), default=None + ) + presentation_id: str | None = field( + metadata=field_options(alias="presentationId"), default=None + ) + owner_id: str | None = field(metadata=field_options(alias="ownerId"), default=None) + create_time: str | None = field( + metadata=field_options(alias="createTime"), default=None + ) + profile: dict[str, Any] | None = None + restriction_tier: int | None = field( + metadata=field_options(alias="restrictionTier"), default=None + ) + allowed: list[str] | None = None + execution_context: str | None = field( + metadata=field_options(alias="executionContext"), default=None + ) + relationships: dict[str, Any] | list[dict[str, Any]] | None = None + # Network-specific objects (device-type dependent, issue #529) + zigbee: dict[str, Any] | None = None + zwave: dict[str, Any] | None = None + lan: dict[str, Any] | None = None + virtual: dict[str, Any] | None = None + edge_child: dict[str, Any] | None = field( + metadata=field_options(alias="edgeChild"), default=None + ) @classmethod def __pre_deserialize__(cls, d: dict[str, Any]) -> dict[str, Any]: """Pre deserialize hook.""" d["components"] = {component["id"]: component for component in d["components"]} + + # Handle empty relationships: API returns [] for empty, convert to None + if "relationships" in d and d["relationships"] == []: + d["relationships"] = None + + # Handle empty viper dict: has required fields, empty should be None + if "viper" in d and isinstance(d["viper"], dict) and not d["viper"]: + d["viper"] = None + return d diff --git a/test_connection.py b/test_connection.py new file mode 100644 index 0000000..4daccc6 --- /dev/null +++ b/test_connection.py @@ -0,0 +1,77 @@ +"""Quick test script to verify SmartThings API connection.""" + +import asyncio +from pathlib import Path + +from aiohttp import ClientSession + +from pysmartthings import SmartThings + + +async def test_connection() -> None: + """Test connection to SmartThings API and list devices.""" + # Load token from .env.local + env_file = Path(__file__).parent / ".env.local" + token = None + + if env_file.exists(): + for line in env_file.read_text().splitlines(): + if line.startswith("SMARTTHINGS_TOKEN="): + token = line.split("=", 1)[1] + break + + if not token: + print("āŒ No token found in .env.local") # noqa: T201 + return + + print(f"āœ… Token loaded: {token[:8]}...") # noqa: T201 + + async with ClientSession() as session: + api = SmartThings(session=session) + api.authenticate(token) + + try: + # Test API connection by getting locations + print("\nšŸ” Testing API connection...") # noqa: T201 + locations = await api.get_locations() + print(f"āœ… Connected! Found {len(locations)} location(s)") # noqa: T201 + + for location in locations: + print(f"\nšŸ“ Location: {location.name}") # noqa: T201 + print(f" ID: {location.location_id}") # noqa: T201 + + # Get devices + print("\nšŸ” Fetching devices...") # noqa: T201 + devices = await api.get_devices() + print(f"āœ… Found {len(devices)} device(s)") # noqa: T201 + + if devices: + print("\nšŸ“± Your devices:") # noqa: T201 + for device in devices: + print(f" ā€¢ {device.label or device.name}") # noqa: T201 + print(f" Type: {device.device_type_name or 'Unknown'}") # noqa: T201 + print(f" ID: {device.device_id}") # noqa: T201 + # Capabilities are in components (main component typically) + if device.components and "main" in device.components: + main_component = device.components["main"] + if main_component.capabilities: + caps = ", ".join( + str(c) for c in main_component.capabilities[:5] + ) + if len(main_component.capabilities) > 5: + num_more = len(main_component.capabilities) - 5 + caps += f", ... (+{num_more} more)" + print(f" Capabilities: {caps}") # noqa: T201 + print() # noqa: T201 + else: + msg = "No devices found. " + msg += "Make sure devices are connected to your SmartThings hub." + print(f" {msg}") # noqa: T201 + + except Exception as e: # noqa: BLE001 # pylint: disable=broad-exception-caught + print(f"āŒ Error: {e}") # noqa: T201 + print(f" Type: {type(e).__name__}") # noqa: T201 + + +if __name__ == "__main__": + asyncio.run(test_connection())