1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
|
"""Comprehensive example demonstrating the ActronAirAPI library.
This example shows:
1. OAuth2 device code flow authentication
2. System information retrieval
3. Status monitoring with typed models
4. Object-oriented control methods (optional)
Usage:
python example.py
Environment Variables:
ACTRON_ACCESS_TOKEN - Saved access token to skip authentication
ACTRON_REFRESH_TOKEN - Saved refresh token to skip authentication
ACTRON_DEMO_CONTROLS - Set to 'true' to enable control demonstrations
The example will guide you through authentication and then display your AC system
information. Control demonstrations are disabled by default to avoid accidentally
changing your AC settings.
"""
import asyncio
import logging
import os
from typing import Optional, Tuple
from actron_neo_api import ActronAirAPI, ActronAirAPIError, ActronAirAuthError, ActronAirStatus
# Set up logging
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
async def oauth2_authentication_example() -> Tuple[Optional[str], Optional[str]]:
"""Example of OAuth2 device code flow authentication."""
print("\n=== OAUTH2 AUTHENTICATION FLOW ===\n")
async with ActronAirAPI() as api:
try:
# Step 1: Request device code
logger.info("Requesting device code...")
device_code_response = await api.request_device_code()
device_code = device_code_response["device_code"]
user_code = device_code_response["user_code"]
verification_uri = device_code_response["verification_uri"]
verification_uri_complete = device_code_response["verification_uri_complete"]
expires_in = device_code_response["expires_in"]
interval = device_code_response["interval"]
# Step 2: Display instructions to user
print("\n" + "=" * 60)
print("OAUTH2 DEVICE CODE FLOW")
print("=" * 60)
print(f"1. Open this URL in your browser: {verification_uri}")
print(f"2. Enter this code: {user_code}")
print(f"3. Or use the complete URL: {verification_uri_complete}")
print(f"4. Complete authorization within {expires_in // 60} minutes")
print("=" * 60)
print("Waiting for authorization...")
# Step 3: Poll for token (with automatic polling)
try:
token_data = await api.poll_for_token(
device_code, interval=interval, timeout=expires_in
)
if token_data:
logger.info("Authorization successful!")
else:
logger.error("Authorization timed out")
return None, None
except Exception as e:
logger.error("Error during authorization: %s", e)
return None, None
# Return tokens for use in the main example
access_token = api.access_token
refresh_token = api.refresh_token_value
print("\n" + "=" * 60)
print("AUTHENTICATION SUCCESSFUL!")
print("=" * 60)
if access_token:
print(f"Access Token: {access_token[:20]}...")
if refresh_token:
print(f"Refresh Token: {refresh_token[:20]}...")
print("Save these tokens for future use!")
print("=" * 60)
return access_token, refresh_token
except Exception as e:
logger.error("OAuth2 authentication failed: %s", e, exc_info=True)
raise
async def api_usage_example(refresh_token: str) -> None:
"""Example of using the ActronAirAPI with saved OAuth2 tokens."""
print("\n=== API USAGE EXAMPLE ===\n")
try:
# Initialize with refresh token - token will be refreshed on first API call
logger.info("Initializing API with refresh token...")
api = ActronAirAPI(refresh_token=refresh_token)
# Get user information
logger.info("Getting user information...")
user_info = await api.get_user_info()
logger.info("Authenticated as: %s", user_info.get("name", "Unknown"))
# Get AC systems
logger.info("Fetching AC systems...")
systems = await api.get_ac_systems()
logger.info("Found %d AC systems", len(systems))
if not systems:
logger.warning("No AC systems found in your account")
return
# Update status to get the latest data (may fail if events API is not accessible)
logger.info("Updating system status...")
await api.update_status()
# Work with the first system
system = systems[0]
serial = system.get("serial")
family = system.get("family")
name = system.get("name", "Unknown System")
logger.info("Working with system: %s (Serial: %s)", name, serial)
# Get the typed status object
if not serial:
logger.warning("System has no serial number, skipping")
return
status = api.state_manager.get_status(serial)
if not status:
logger.warning(
"Status data not available for system %s (events API may not be accessible)", serial
)
print("\n" + "=" * 60)
print("SYSTEM INFORMATION")
print("=" * 60)
print(f"System Name: {name}")
print(f"Serial: {serial}")
print(f"Family: {family}")
print("Status: Events API not accessible - limited information available")
print("=" * 60)
else:
# Display current system information
print("\n" + "=" * 60)
print("SYSTEM INFORMATION")
print("=" * 60)
if status.ac_system:
print(f"System Name: {status.ac_system.system_name}")
print(f"Model: {status.ac_system.master_wc_model}")
print(f"Firmware: {status.ac_system.master_wc_firmware_version}")
print(f"Serial: {status.ac_system.master_serial}")
# Display current settings
if status.user_aircon_settings:
settings = status.user_aircon_settings
print("\nCURRENT SETTINGS:")
print(f"Power: {'ON' if settings.is_on else 'OFF'}")
print(f"Mode: {settings.mode}")
print(f"Fan Mode: {'Enabled' if settings.continuous_fan_enabled else 'Disabled'}")
print(f"Cool Setpoint: {settings.temperature_setpoint_cool_c}°C")
print(f"Heat Setpoint: {settings.temperature_setpoint_heat_c}°C")
# Get current temperature from zones
current_temp = None
if status.remote_zone_info:
for zone in status.remote_zone_info:
if zone.live_temp_c is not None and zone.live_temp_c > 0:
current_temp = zone.live_temp_c
break
elif zone.peripheral_temperature is not None:
current_temp = zone.peripheral_temperature
break
if current_temp is not None:
print(f"Current Temperature: {current_temp}°C")
else:
print("Current Temperature: Not available")
print(f"Quiet Mode: {'Enabled' if settings.quiet_mode_enabled else 'Disabled'}")
print(f"Turbo Mode: {'Enabled' if settings.turbo_mode_enabled else 'Disabled'}")
print(f"Away Mode: {'Enabled' if settings.away_mode else 'Disabled'}")
# Display zone information
if status.remote_zone_info:
print("\nZONE INFORMATION:")
for i, zone in enumerate(status.remote_zone_info):
print(
f"Zone {i + 1} ({zone.title}): "
f"{'Enabled' if zone.is_active else 'Disabled'}"
)
print(f" Set Temperature Cool: {zone.temperature_setpoint_cool_c}°C")
print(f" Set Temperature Heat: {zone.temperature_setpoint_heat_c}°C")
# Display actual temperature from zone or peripheral
if zone.live_temp_c is not None and zone.live_temp_c > 0:
print(f" Current Temperature: {zone.live_temp_c}°C")
elif zone.peripheral_temperature is not None:
print(f" Current Temperature: {zone.peripheral_temperature}°C (sensor)")
else:
print(" Current Temperature: Not available")
# Display humidity
if zone.humidity is not None and zone.humidity > 0:
print(f" Current Humidity: {zone.humidity}%")
else:
print(" Current Humidity: Not available")
# Display battery level if available
if zone.battery_level is not None:
print(f" Sensor Battery: {zone.battery_level}%")
else:
print(" Sensor Battery: Not available")
print(f" Zone Position: {zone.zone_position}%")
print("=" * 60)
# Demonstrate control capabilities
print("\n=== DEMONSTRATING CONTROL CAPABILITIES ===\n")
# Check if user wants to run control examples
# if os.environ.get("ACTRON_DEMO_CONTROLS", "").lower() == "true":
await demonstrate_controls(api, status, serial)
# else:
# print("Control demonstrations are disabled by default.")
# print("Set ACTRON_DEMO_CONTROLS=true environment variable to enable.")
# print("\nThe following controls would be available:")
# print("- AC System control (power, mode, temperature)")
# print("- Fan control (speed, continuous mode)")
# print("- Zone control (enable/disable, temperature)")
# print("- Special modes (quiet, turbo, away)")
print("\n=== EXAMPLE COMPLETE ===")
print("Successfully demonstrated:")
print("1. OAuth2 token authentication")
print("2. System information retrieval")
print("3. Status monitoring with typed models")
print("4. Object-oriented API access")
# Close the API session properly
await api.close()
except ActronAirAuthError as e:
logger.error("Authentication error: %s", e)
print("Authentication failed. Your tokens may have expired.")
print("Please run the OAuth2 flow again to get new tokens.")
except ActronAirAPIError as e:
logger.error("API error: %s", e)
raise
except Exception as e:
logger.error("Unexpected error: %s", e, exc_info=True)
raise
async def demonstrate_controls(api: ActronAirAPI, status: ActronAirStatus, serial: str) -> None:
"""Demonstrate actual control operations."""
print("Running control demonstrations...")
print("WARNING: This will change your AC system settings!")
try:
# AC System Control
if status.ac_system:
print("\n--- AC System Control ---")
# Set system to COOL mode
logger.info("Setting system to COOL mode...")
await status.ac_system.set_system_mode(mode="COOL")
print("✓ System mode set to COOL")
# Wait a moment for the change to take effect
await asyncio.sleep(2)
# Settings Control
if status.user_aircon_settings:
print("\n--- Settings Control ---")
settings = status.user_aircon_settings
# Set temperature to 23°C
logger.info("Setting temperature to 23°C...")
await settings.set_temperature(23.0)
print("✓ Temperature set to 23°C")
# Set fan mode to AUTO
logger.info("Setting fan mode to AUTO...")
await settings.set_fan_mode("AUTO")
print("✓ Fan mode set to AUTO")
# Enable quiet mode
logger.info("Enabling quiet mode...")
await settings.set_quiet_mode(enabled=True)
print("✓ Quiet mode enabled")
await asyncio.sleep(2)
# Zone Control
if status.remote_zone_info and len(status.remote_zone_info) > 0:
print("\n--- Zone Control ---")
# Enable the first zone
first_zone = status.remote_zone_info[0]
logger.info("Enabling zone %d (%s)...", first_zone.zone_id, first_zone.title)
await first_zone.enable(is_enabled=True)
print(f"✓ Zone {first_zone.zone_id} enabled")
# Set zone temperature to 22°C
logger.info("Setting zone temperature to 22°C...")
await first_zone.set_temperature(22.0)
print("✓ Zone temperature set to 22°C")
await asyncio.sleep(2)
# Update status to show changes
print("\n--- Updating Status ---")
logger.info("Refreshing status to show changes...")
await api.update_status()
updated_status = api.state_manager.get_status(serial)
if updated_status and updated_status.user_aircon_settings:
settings = updated_status.user_aircon_settings
print("✓ Updated settings:")
print(f" Power: {'ON' if settings.is_on else 'OFF'}")
print(f" Mode: {settings.mode}")
print(f" Temperature: {settings.temperature_setpoint_cool_c}°C")
print(f" Fan Mode: {settings.fan_mode}")
print(f" Quiet Mode: {'Enabled' if settings.quiet_mode_enabled else 'Disabled'}")
print("✓ Control demonstrations completed successfully!")
except Exception as e:
logger.error("Error during control demonstration: %s", e, exc_info=True)
raise
async def main() -> None:
"""Main example function that demonstrates both authentication and API usage."""
print("=== ACTRON AIR API EXAMPLE ===")
print("This example demonstrates OAuth2 authentication and API usage.")
# Check if we have saved tokens in environment variables
saved_access_token = os.environ.get("ACTRON_ACCESS_TOKEN")
saved_refresh_token = os.environ.get("ACTRON_REFRESH_TOKEN")
if saved_access_token and saved_refresh_token:
print("\nUsing saved tokens from environment variables...")
access_token: Optional[str] = saved_access_token
refresh_token: Optional[str] = saved_refresh_token
else:
print("\nNo saved tokens found. Starting OAuth2 authentication flow...")
access_token, refresh_token = await oauth2_authentication_example()
if not access_token or not refresh_token:
print("Authentication failed. Cannot continue with API example.")
return
print("\nTo skip authentication in future runs, set these environment variables:")
print(f"export ACTRON_ACCESS_TOKEN='{access_token}'")
print(f"export ACTRON_REFRESH_TOKEN='{refresh_token}'")
# Run the API usage example
# At this point we know refresh_token is not None due to the check above
assert refresh_token is not None
await api_usage_example(refresh_token)
if __name__ == "__main__":
asyncio.run(main())
|
