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
|
import asyncio
import logging
import sys
from pprint import pprint
from airtouch5py.airtouch5_client import Airtouch5Client, Airtouch5ConnectionStateChange
from airtouch5py.airtouch5_simple_client import Airtouch5SimpleClient
from airtouch5py.packets.ac_ability import AcAbilityData
from airtouch5py.packets.ac_error_information import AcErrorInformationData
from airtouch5py.packets.ac_status import AcStatusData
from airtouch5py.packets.console_version import ConsoleVersionData
from airtouch5py.packets.datapacket import DataPacket
from airtouch5py.packets.zone_control import ZoneControlData
from airtouch5py.packets.zone_name import ZoneNameData
from airtouch5py.packets.zone_status import ZoneStatusData
def print_packet(packet: DataPacket):
match packet.data:
case ZoneControlData():
print(f"Received {len(packet.data.zones)} zones data")
pprint(packet.data.zones)
case ZoneStatusData():
print(f"Received {len(packet.data.zones)} zones status")
pprint(packet.data.zones)
case AcStatusData():
print(f"Received {len(packet.data.ac_status)} ac status")
pprint(packet.data.ac_status)
case AcAbilityData():
print(f"Received {len(packet.data.ac_ability)} ac ability")
pprint(packet.data.ac_ability)
case AcErrorInformationData():
print(
f"Received ac error information {packet.data.ac_number} {packet.data.error_info}"
)
case ZoneNameData():
print(f"Received {len(packet.data.zone_names)} zones names")
pprint(packet.data.zone_names)
case ConsoleVersionData():
print(
f"Received console version. has update {packet.data.has_update}, version {packet.data.version}"
)
case _:
print(f"Received unknown packet {packet.data}")
async def main(ip: str):
logger = logging.getLogger("airtouch5pytest")
client = Airtouch5SimpleClient(ip)
print("Testing connection")
try:
await client.test_connection()
print("Succeeded")
except Exception as e:
print(f"Failed: {e}")
return
print("Connecting...")
try:
await client.connect_and_stay_connected()
except Exception as e:
print(f"Failed: {e}")
return
print(f"Connected, we have {len(client.zones)} zones and {len(client.ac)} acs")
print(f"Initial ac status {client.latest_ac_status}")
print(f"Initial zone status {client.latest_zone_status}")
client.connection_state_callbacks.append(
lambda x: print(f"Connection state changed to {x}")
)
client.data_packet_callbacks.append(print_packet)
client.zone_status_callbacks.append(lambda x: print(f"Zone status changed {x}"))
# await client.send_packet(client.data_packet_factory.zone_status_request())
# await client.send_packet(client.data_packet_factory.ac_status_request())
await asyncio.sleep(999)
if __name__ == "__main__":
asyncio.run(main(sys.argv[1]))
|
