seb/tischlerCtrl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(tapo-agent): add schedule/countdown timer API support

Browse Source 
- Fork tapo crate to add missing schedule/timer APIs
- Add get_countdown_rules, get_schedule_rules, get_next_event methods
- New readings: countdown_active, countdown_remain, schedule_count,
  schedule_active_count, next_event_time
- Add local compilation to build script alongside cross-compilation
- Implement offline polling - device collection continues when server disconnects
- Add more device readings: on_time, signal_level, rssi, runtime_today/month, energy_month

Vendored tapo fork in tapo-fork/ with minimal changes to add schedule APIs.
This commit is contained in:
seb
2025-12-23 00:46:42 +01:00
parent f3cca149f9
commit 028763bdb2
274 changed files with 20784 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
agents/tapo/tapo-fork/tapo-py/examples/tapo_discover_devices.py Normal file
View File

@@ -0,0 +1,70 @@
"""Discover devices on the local network Example"""
import asyncio
import os
from tapo import ApiClient, DiscoveryResult
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
target = os.getenv("TARGET", "192.168.1.255")
timeout_s = int(os.getenv("TIMEOUT", 10))
print(f"Discovering Tapo devices on target: {target} for {timeout_s} seconds...")
api_client = ApiClient(tapo_username, tapo_password)
discovery = await api_client.discover_devices(target, timeout_s)
async for discovery_result in discovery:
try:
device = discovery_result.get()
match device:
case DiscoveryResult.GenericDevice(device_info, _handler):
print(
f"Found Unsupported Device '{device_info.nickname}' of model '{device_info.model}' at IP address '{device_info.ip}'."
)
case DiscoveryResult.Light(device_info, _handler):
print(
f"Found '{device_info.nickname}' of model '{device_info.model}' at IP address '{device_info.ip}'."
)
case DiscoveryResult.ColorLight(device_info, _handler):
print(
f"Found '{device_info.nickname}' of model '{device_info.model}' at IP address '{device_info.ip}'."
)
case DiscoveryResult.RgbLightStrip(device_info, _handler):
print(
f"Found '{device_info.nickname}' of model '{device_info.model}' at IP address '{device_info.ip}'."
)
case DiscoveryResult.RgbicLightStrip(device_info, _handler):
print(
f"Found '{device_info.nickname}' of model '{device_info.model}' at IP address '{device_info.ip}'."
)
case DiscoveryResult.Plug(device_info, _handler):
print(
f"Found '{device_info.nickname}' of model '{device_info.model}' at IP address '{device_info.ip}'."
)
case DiscoveryResult.PlugEnergyMonitoring(device_info, _handler):
print(
f"Found '{device_info.nickname}' of model '{device_info.model}' at IP address '{device_info.ip}'."
)
case DiscoveryResult.PowerStrip(device_info, _handler):
print(
f"Found Power Strip of model '{device_info.model}' at IP address '{device_info.ip}'."
)
case DiscoveryResult.PowerStripEnergyMonitoring(device_info, _handler):
print(
f"Found Power Strip with Energy Monitoring of model '{device_info.model}' at IP address '{device_info.ip}'."
)
case DiscoveryResult.Hub(device_info, _handler):
print(
f"Found '{device_info.nickname}' of model '{device_info.model}' at IP address '{device_info.ip}'."
)
except Exception as e:
print(f"Error discovering device: {e}")
if __name__ == "__main__":
asyncio.run(main())

31
agents/tapo/tapo-fork/tapo-py/examples/tapo_generic_device.py Normal file
View File

@@ -0,0 +1,31 @@
"""Generic Device Example"""
import asyncio
import os
from tapo import ApiClient
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
device = await client.generic_device(ip_address)
print("Turning device on...")
await device.on()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Turning device off...")
await device.off()
device_info = await device.get_device_info()
print(f"Device info: {device_info.to_dict()}")
if __name__ == "__main__":
asyncio.run(main())

30
agents/tapo/tapo-fork/tapo-py/examples/tapo_generic_device_toggle.py Normal file
View File

@@ -0,0 +1,30 @@
"""Toggle Generic Device Example"""
import asyncio
import os
from tapo import ApiClient
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
device = await client.generic_device(ip_address)
device_info = await device.get_device_info()
if device_info.device_on == True:
print("Device is on. Turning it off...")
await device.off()
elif device_info.device_on == False:
print("Device is off. Turning it on...")
await device.on()
else:
print("This device does not support on/off functionality.")
if __name__ == "__main__":
asyncio.run(main())

120
agents/tapo/tapo-fork/tapo-py/examples/tapo_h100.py Normal file
View File

@@ -0,0 +1,120 @@
"""H100 Example"""
import asyncio
import os
from tapo import ApiClient
from tapo.requests import AlarmRingtone, AlarmVolume, AlarmDuration
from tapo.responses import KE100Result, S200BResult, T100Result, T110Result, T300Result, T31XResult
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
hub = await client.h100(ip_address)
device_info = await hub.get_device_info()
print(f"Device info: {device_info.to_dict()}")
child_device_list = await hub.get_child_device_list()
for child in child_device_list:
if child is None:
print("Found unsupported device.")
elif isinstance(child, KE100Result):
print(
"Found KE100 child device with nickname: {}, id: {}, current temperature: {:.2f} {} and target temperature: {:.2f} {}.".format(
child.nickname,
child.device_id,
child.current_temperature,
child.temperature_unit,
child.target_temperature,
child.temperature_unit,
)
)
elif isinstance(child, S200BResult):
s200b = await hub.s200b(device_id=child.device_id)
trigger_logs = await s200b.get_trigger_logs(5, 0)
print(
"Found S200B child device with nickname: {}, id: {}, last 5 trigger logs: {}.".format(
child.nickname,
child.device_id,
[log.to_dict() for log in trigger_logs.logs],
)
)
elif isinstance(child, T100Result):
t100 = await hub.t100(device_id=child.device_id)
trigger_logs = await t100.get_trigger_logs(5, 0)
print(
"Found T100 child device with nickname: {}, id: {}, detected: {}, last 5 trigger logs: {}.".format(
child.nickname,
child.device_id,
child.detected,
[log.to_dict() for log in trigger_logs.logs],
)
)
elif isinstance(child, T110Result):
t110 = await hub.t110(device_id=child.device_id)
trigger_logs = await t110.get_trigger_logs(5, 0)
print(
"Found T110 child device with nickname: {}, id: {}, open: {}, last 5 trigger logs: {}.".format(
child.nickname,
child.device_id,
child.open,
[log.to_dict() for log in trigger_logs.logs],
)
)
elif isinstance(child, T300Result):
t300 = await hub.t300(device_id=child.device_id)
trigger_logs = await t300.get_trigger_logs(5, 0)
print(
"Found T300 child device with nickname: {}, id: {}, in_alarm: {}, water_leak_status: {}, last 5 trigger logs: {}.".format(
child.nickname,
child.device_id,
child.in_alarm,
child.water_leak_status,
[log.to_dict() for log in trigger_logs.logs],
)
)
elif isinstance(child, T31XResult):
t31x = await hub.t315(device_id=child.device_id)
temperature_humidity_records = await t31x.get_temperature_humidity_records()
print(
"Found T31X child device with nickname: {}, id: {}, temperature: {:.2f} {}, humidity: {}%, earliest temperature and humidity record available: {}.".format(
child.nickname,
child.device_id,
child.current_temperature,
child.temperature_unit,
child.current_humidity,
(
temperature_humidity_records.records[0].to_dict()
if temperature_humidity_records.records
else None
),
)
)
print(f"Triggering the alarm ringtone 'Alarm 1' at a 'Low' volume for '3 Seconds'...")
await hub.play_alarm(AlarmRingtone.Alarm1, AlarmVolume.Low, AlarmDuration.Seconds, seconds=3)
device_info = await hub.get_device_info()
print(f"Is device ringing?: {device_info.in_alarm}")
print("Stopping the alarm after 1 Second...")
await asyncio.sleep(1)
await hub.stop_alarm()
device_info = await hub.get_device_info()
print(f"Is device ringing?: {device_info.in_alarm}")
if __name__ == "__main__":
asyncio.run(main())

40
agents/tapo/tapo-fork/tapo-py/examples/tapo_ke100.py Normal file
View File

@@ -0,0 +1,40 @@
"""KE100 TRV Example"""
import asyncio
import os
from tapo import ApiClient
from tapo.requests import TemperatureUnitKE100
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
# Name of the KE100 device.
# Can be obtained from the Tapo App or by executing `get_child_device_component_list()` on the hub device.
device_name = os.getenv("DEVICE_NAME")
target_temperature = int(os.getenv("TARGET_TEMPERATURE"))
client = ApiClient(tapo_username, tapo_password)
hub = await client.h100(ip_address)
# Get a handler for the child device
device = await hub.ke100(nickname=device_name)
# Get the device info of the child device
device_info = await device.get_device_info()
print(f"Device info: {device_info.to_dict()}")
# Set target temperature.
# KE100 currently only supports Celsius as temperature unit.
print(f"Setting target temperature to {target_temperature} degrees Celsius...")
await device.set_target_temperature(target_temperature, TemperatureUnitKE100.Celsius)
# Get the device info of the child device
device_info = await device.get_device_info()
print(f"Device info: {device_info.to_dict()}")
if __name__ == "__main__":
asyncio.run(main())

40
agents/tapo/tapo-fork/tapo-py/examples/tapo_l510.py Normal file
View File

@@ -0,0 +1,40 @@
"""L510, L520 and L610 Example"""
import asyncio
import os
from tapo import ApiClient
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
device = await client.l510(ip_address)
print("Turning device on...")
await device.on()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting the brightness to 30%...")
await device.set_brightness(30)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Turning device off...")
await device.off()
device_info = await device.get_device_info()
print(f"Device info: {device_info.to_dict()}")
device_usage = await device.get_device_usage()
print(f"Device usage: {device_usage.to_dict()}")
if __name__ == "__main__":
asyncio.run(main())

62
agents/tapo/tapo-fork/tapo-py/examples/tapo_l530.py Normal file
View File

@@ -0,0 +1,62 @@
"""L530, L535 and L630 Example"""
import asyncio
import os
from tapo import ApiClient
from tapo.requests import Color
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
device = await client.l530(ip_address)
print("Turning device on...")
await device.on()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting the brightness to 30%...")
await device.set_brightness(30)
print("Setting the color to `Chocolate`...")
await device.set_color(Color.Chocolate)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting the color to `Deep Sky Blue` using the `hue` and `saturation`...")
await device.set_hue_saturation(195, 100)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting the color to `Incandescent` using the `color temperature`...")
await device.set_color_temperature(2700)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Using the `set` API to set multiple properties in a single request...")
await device.set().brightness(50).color(Color.HotPink).send(device)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Turning device off...")
await device.off()
device_info = await device.get_device_info()
print(f"Device info: {device_info.to_dict()}")
device_usage = await device.get_device_usage()
print(f"Device usage: {device_usage.to_dict()}")
if __name__ == "__main__":
asyncio.run(main())

62
agents/tapo/tapo-fork/tapo-py/examples/tapo_l900.py Normal file
View File

@@ -0,0 +1,62 @@
"""L900 Example"""
import asyncio
import os
from tapo import ApiClient
from tapo.requests import Color
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
device = await client.l900(ip_address)
print("Turning device on...")
await device.on()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting the brightness to 30%...")
await device.set_brightness(30)
print("Setting the color to `Chocolate`...")
await device.set_color(Color.Chocolate)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting the color to `Deep Sky Blue` using the `hue` and `saturation`...")
await device.set_hue_saturation(195, 100)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting the color to `Incandescent` using the `color temperature`...")
await device.set_color_temperature(2700)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Using the `set` API to set multiple properties in a single request...")
await device.set().brightness(50).color(Color.HotPink).send(device)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Turning device off...")
await device.off()
device_info = await device.get_device_info()
print(f"Device info: {device_info.to_dict()}")
device_usage = await device.get_device_usage()
print(f"Device usage: {device_usage.to_dict()}")
if __name__ == "__main__":
asyncio.run(main())

103
agents/tapo/tapo-fork/tapo-py/examples/tapo_l930.py Normal file
View File

@@ -0,0 +1,103 @@
"""L920 and L930 Example"""
import asyncio
import os
from tapo import ApiClient
from tapo.requests import Color, LightingEffect, LightingEffectPreset, LightingEffectType
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
device = await client.l930(ip_address)
print("Turning device on...")
await device.on()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting the brightness to 30%...")
await device.set_brightness(30)
print("Setting the color to `Chocolate`...")
await device.set_color(Color.Chocolate)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting the color to `Deep Sky Blue` using the `hue` and `saturation`...")
await device.set_hue_saturation(195, 100)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting the color to `Incandescent` using the `color temperature`...")
await device.set_color_temperature(2700)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Using the `set` API to set multiple properties in a single request...")
await device.set().brightness(50).color(Color.HotPink).send(device)
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Setting a preset Lighting effect...")
await device.set_lighting_effect(LightingEffectPreset.BubblingCauldron)
print("Waiting 10 seconds...")
await asyncio.sleep(10)
print("Setting a custom static Lighting effect...")
custom_effect = (
LightingEffect(
"My Custom Static Effect", LightingEffectType.Static, True, True, 100, [(359, 85, 100)]
)
.with_expansion_strategy(1)
.with_segments([0, 1, 2])
.with_sequence([(359, 85, 100), (0, 0, 100), (236, 72, 100)])
)
await device.set_lighting_effect(custom_effect)
print("Waiting 10 seconds...")
await asyncio.sleep(10)
print("Setting a custom sequence Lighting effect...")
custom_effect = (
LightingEffect(
"My Custom Sequence Effect",
LightingEffectType.Sequence,
True,
True,
100,
[(359, 85, 100)],
)
.with_expansion_strategy(1)
.with_segments([0, 1, 2])
.with_sequence([(359, 85, 100), (0, 0, 100), (236, 72, 100)])
.with_direction(1)
.with_duration(50)
)
await device.set_lighting_effect(custom_effect)
print("Waiting 10 seconds...")
await asyncio.sleep(10)
print("Turning device off...")
await device.off()
device_info = await device.get_device_info()
print(f"Device info: {device_info.to_dict()}")
device_usage = await device.get_device_usage()
print(f"Device usage: {device_usage.to_dict()}")
if __name__ == "__main__":
asyncio.run(main())

34
agents/tapo/tapo-fork/tapo-py/examples/tapo_p100.py Normal file
View File

@@ -0,0 +1,34 @@
"""P100 and P105 Example"""
import asyncio
import os
from tapo import ApiClient
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
device = await client.p100(ip_address)
print("Turning device on...")
await device.on()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Turning device off...")
await device.off()
device_info = await device.get_device_info()
print(f"Device info: {device_info.to_dict()}")
device_usage = await device.get_device_usage()
print(f"Device usage: {device_usage.to_dict()}")
if __name__ == "__main__":
asyncio.run(main())

118
agents/tapo/tapo-fork/tapo-py/examples/tapo_p110.py Normal file
View File

@@ -0,0 +1,118 @@
"""P110, P110M and P115 Example"""
import asyncio
import os
from datetime import datetime, timedelta, timezone
from tapo import ApiClient
from tapo.requests import EnergyDataInterval, PowerDataInterval
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
device = await client.p110(ip_address)
print("Turning device on...")
await device.on()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Turning device off...")
await device.off()
device_info = await device.get_device_info()
print(f"Device info: {device_info.to_dict()}")
current_power = await device.get_current_power()
print(f"Current power: {current_power.to_dict()}")
device_usage = await device.get_device_usage()
print(f"Device usage: {device_usage.to_dict()}")
energy_usage = await device.get_energy_usage()
print(f"Energy usage: {energy_usage.to_dict()}")
today = datetime.now(timezone.utc)
# Energy data - Hourly interval
# `start_date` and `end_date` are an inclusive interval that must not be greater than 8 days.
energy_data_hourly = await device.get_energy_data(EnergyDataInterval.Hourly, today)
print(
"Energy data (hourly): "
f"Start date time '{energy_data_hourly.start_date_time}', "
f"Entries {len(energy_data_hourly.entries)}, "
f"First entry: {energy_data_hourly.entries[0].to_dict() if energy_data_hourly.entries else None}"
)
# Energy data - Daily interval
# `start_date` must be the first day of a quarter.
energy_data_daily = await device.get_energy_data(
EnergyDataInterval.Daily,
datetime(today.year, get_quarter_start_month(today), 1),
)
print(
"Energy data (daily): "
f"Start date time '{energy_data_daily.start_date_time}', "
f"Entries {len(energy_data_daily.entries)}, "
f"First entry: {energy_data_daily.entries[0].to_dict() if energy_data_daily.entries else None}"
)
# Energy data - Monthly interval
# `start_date` must be the first day of a year.
energy_data_monthly = await device.get_energy_data(
EnergyDataInterval.Monthly,
datetime(today.year, 1, 1),
)
print(
"Energy data (monthly): "
f"Start date time '{energy_data_monthly.start_date_time}', "
f"Entries {len(energy_data_monthly.entries)}, "
f"First entry: {energy_data_monthly.entries[0].to_dict() if energy_data_monthly.entries else None}"
)
# Power data - Every 5 minutes interval
# `start_date_time` and `end_date_time` describe an exclusive interval.
# If the result would yield more than 144 entries (i.e. 12 hours),
# the `end_date_time` will be adjusted to an earlier date and time.
power_data_every_5_minutes = await device.get_power_data(
PowerDataInterval.Every5Minutes,
today - timedelta(hours=12),
today,
)
print(
"Power data (every 5 minutes): "
f"Start date time '{power_data_every_5_minutes.start_date_time}', "
f"End date time '{power_data_every_5_minutes.end_date_time}', "
f"Entries {len(power_data_every_5_minutes.entries)}, "
f"First entry: {power_data_every_5_minutes.entries[0].to_dict() if power_data_every_5_minutes.entries else None}"
)
# Power data - Hourly interval
# `start_date_time` and `end_date_time` describe an exclusive interval.
# If the result would yield more than 144 entries (i.e. 6 days),
# the `end_date_time` will be adjusted to an earlier date and time.
power_data_hourly = await device.get_power_data(
PowerDataInterval.Hourly,
today - timedelta(days=3),
today,
)
print(
"Power data (hourly): "
f"Start date time '{power_data_hourly.start_date_time}', "
f"End date time '{power_data_hourly.end_date_time}', "
f"Entries {len(power_data_hourly.entries)}, "
f"First entry: {power_data_hourly.entries[0].to_dict() if power_data_hourly.entries else None}"
)
def get_quarter_start_month(today: datetime) -> int:
return ((today.month - 1) // 3) * 3 + 1
if __name__ == "__main__":
asyncio.run(main())

45
agents/tapo/tapo-fork/tapo-py/examples/tapo_p300.py Normal file
View File

@@ -0,0 +1,45 @@
"""P300 and P306 Example"""
import asyncio
import os
from tapo import ApiClient
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
power_strip = await client.p300(ip_address)
device_info = await power_strip.get_device_info()
print(f"Device info: {device_info.to_dict()}")
print("Getting child devices...")
child_device_list = await power_strip.get_child_device_list()
print(f"Found {len(child_device_list)} plugs")
for index, child in enumerate(child_device_list):
print(f"=== ({index + 1}) {child.nickname} ===")
print(f"Device ID: {child.device_id}")
print(f"State: {child.device_on}")
plug = await power_strip.plug(device_id=child.device_id)
print("Turning device on...")
await plug.on()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Turning device off...")
await plug.off()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
if __name__ == "__main__":
asyncio.run(main())

132
agents/tapo/tapo-fork/tapo-py/examples/tapo_p304.py Normal file
View File

@@ -0,0 +1,132 @@
"""P304M and P316M Example"""
import asyncio
from datetime import datetime, timedelta, timezone
import os
from tapo import ApiClient
from tapo.requests import EnergyDataInterval, PowerDataInterval
async def main():
tapo_username = os.getenv("TAPO_USERNAME")
tapo_password = os.getenv("TAPO_PASSWORD")
ip_address = os.getenv("IP_ADDRESS")
client = ApiClient(tapo_username, tapo_password)
power_strip = await client.p304(ip_address)
device_info = await power_strip.get_device_info()
print(f"Device info: {device_info.to_dict()}")
print("Getting child devices...")
child_device_list = await power_strip.get_child_device_list()
print(f"Found {len(child_device_list)} plugs")
for index, child in enumerate(child_device_list):
print(f"=== ({index + 1}) {child.nickname} ===")
print(f"Device ID: {child.device_id}")
print(f"State: {child.device_on}")
plug = await power_strip.plug(device_id=child.device_id)
print("Turning device on...")
await plug.on()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
print("Turning device off...")
await plug.off()
print("Waiting 2 seconds...")
await asyncio.sleep(2)
current_power = await plug.get_current_power()
print(f"Current power: {current_power.to_dict()}")
device_usage = await plug.get_device_usage()
print(f"Device usage: {device_usage.to_dict()}")
energy_usage = await plug.get_energy_usage()
print(f"Energy usage: {energy_usage.to_dict()}")
today = datetime.now(timezone.utc)
# Energy data - Hourly interval
# `start_date` and `end_date` are an inclusive interval that must not be greater than 8 days.
energy_data_hourly = await plug.get_energy_data(EnergyDataInterval.Hourly, today)
print(
"Energy data (hourly): "
f"Start date time '{energy_data_hourly.start_date_time}', "
f"Entries {len(energy_data_hourly.entries)}, "
f"First entry: {energy_data_hourly.entries[0].to_dict() if energy_data_hourly.entries else None}"
)
# Energy data - Daily interval
# `start_date` must be the first day of a quarter.
energy_data_daily = await plug.get_energy_data(
EnergyDataInterval.Daily,
datetime(today.year, get_quarter_start_month(today), 1),
)
print(
"Energy data (daily): "
f"Start date time '{energy_data_daily.start_date_time}', "
f"Entries {len(energy_data_daily.entries)}, "
f"First entry: {energy_data_daily.entries[0].to_dict() if energy_data_daily.entries else None}"
)
# Energy data - Monthly interval
# `start_date` must be the first day of a year.
energy_data_monthly = await plug.get_energy_data(
EnergyDataInterval.Monthly,
datetime(today.year, 1, 1),
)
print(
"Energy data (monthly): "
f"Start date time '{energy_data_monthly.start_date_time}', "
f"Entries {len(energy_data_monthly.entries)}, "
f"First entry: {energy_data_monthly.entries[0].to_dict() if energy_data_monthly.entries else None}"
)
# Power data - Every 5 minutes interval
# `start_date_time` and `end_date_time` describe an exclusive interval.
# If the result would yield more than 144 entries (i.e. 12 hours),
# the `end_date_time` will be adjusted to an earlier date and time.
power_data_every_5_minutes = await plug.get_power_data(
PowerDataInterval.Every5Minutes,
today - timedelta(hours=12),
today,
)
print(
"Power data (every 5 minutes): "
f"Start date time '{power_data_every_5_minutes.start_date_time}', "
f"End date time '{power_data_every_5_minutes.end_date_time}', "
f"Entries {len(power_data_every_5_minutes.entries)}, "
f"First entry: {power_data_every_5_minutes.entries[0].to_dict() if power_data_every_5_minutes.entries else None}"
)
# Power data - Hourly interval
# `start_date_time` and `end_date_time` describe an exclusive interval.
# If the result would yield more than 144 entries (i.e. 6 days),
# the `end_date_time` will be adjusted to an earlier date and time.
power_data_hourly = await plug.get_power_data(
PowerDataInterval.Hourly,
today - timedelta(days=3),
today,
)
print(
"Power data (hourly): "
f"Start date time '{power_data_hourly.start_date_time}', "
f"End date time '{power_data_hourly.end_date_time}', "
f"Entries {len(power_data_hourly.entries)}, "
f"First entry: {power_data_hourly.entries[0].to_dict() if power_data_hourly.entries else None}"
)
def get_quarter_start_month(today: datetime) -> int:
return ((today.month - 1) // 3) * 3 + 1
if __name__ == "__main__":
asyncio.run(main())