Examples
Monitoring Example
Simple demonstration of connecting to a known device and printing out telemetry updates.
1"""Example usage of SolixBLE.
2
3.. moduleauthor:: Harvey Lelliott (flip-dots) <harveylelliott@duck.com>
4
5"""
6
7import asyncio
8import logging
9
10from SolixBLE import C300, discover_devices
11
12logging.basicConfig(level=logging.DEBUG)
13
14
15async def main():
16
17 # Find device
18 devices = await discover_devices()
19
20 selected_device = None
21 for device in devices:
22 if device.name is not None and "C300" in device.name:
23 selected_device = device
24 break
25
26 if selected_device is None:
27 print("Device not found!")
28 return
29
30 # Initialize the device
31 device = C300(selected_device)
32 # device = C1000(selected_device)
33
34 # Connect
35 connected = await device.connect()
36
37 if not connected:
38 raise Exception
39
40 # Do nothing, the library will print status updates in debug mode
41 await asyncio.sleep(900)
42
43
44if __name__ == "__main__":
45 asyncio.run(main())
Control Example
Simple demonstration of connecting to a known device and sending commands to control its outputs.
1"""Example usage of SolixBLE for controlling a C1000.
2
3.. moduleauthor:: Harvey Lelliott (flip-dots) <harveylelliott@duck.com>
4
5"""
6
7import asyncio
8import logging
9
10from SolixBLE import C1000, SolixBLEDevice, discover_devices
11from SolixBLE.states import LightStatus
12
13logging.basicConfig(level=logging.DEBUG)
14
15
16async def test_ac_output(device: SolixBLEDevice):
17
18 await asyncio.sleep(10)
19 await device.turn_ac_on()
20
21 await asyncio.sleep(10)
22 await device.turn_ac_off()
23
24
25async def test_light_mode(device: SolixBLEDevice):
26
27 await asyncio.sleep(5)
28 await device.set_light_mode(LightStatus.LOW)
29
30 await asyncio.sleep(5)
31 await device.set_light_mode(LightStatus.MEDIUM)
32
33 await asyncio.sleep(5)
34 await device.set_light_mode(LightStatus.HIGH)
35
36 await asyncio.sleep(5)
37 await device.set_light_mode(LightStatus.SOS)
38
39 await asyncio.sleep(5)
40 await device.set_light_mode(LightStatus.OFF)
41
42
43async def main():
44
45 # Find device
46 devices = await discover_devices()
47
48 selected_device = None
49 for device in devices:
50 if device.name is not None and "C1000" in device.name:
51 selected_device = device
52 break
53
54 if selected_device is None:
55 print("Device not found!")
56 return
57
58 # Initialize the device
59 # device = C300(selected_device)
60 device = C1000(selected_device)
61
62 # Connect
63 connected = await device.connect()
64
65 if not connected:
66 raise Exception
67
68 await test_light_mode(device)
69
70 await asyncio.sleep(300)
71
72
73if __name__ == "__main__":
74 asyncio.run(main())
Complex Example
This is a more advanced demonstration program which prompts the user for the device to connect to, its model, and then prints out the telemetry data on demand and when there is an update. This can be used to add support for new devices, see New devices.
1"""More advanced usage example of SolixBLE.
2
3.. moduleauthor:: Harvey Lelliott (flip-dots) <harveylelliott@duck.com>
4
5"""
6
7import asyncio
8import logging
9import sys
10
11# Allows for reading and writing to the console at the same time
12# pip3 install aioconsole
13from aioconsole import ainput
14from bleak import BLEDevice
15
16from SolixBLE import (
17 C300,
18 C300DC,
19 C800,
20 C1000,
21 C1000G2,
22 F2000,
23 F3800,
24 Generic,
25 PrimeCharger160w,
26 PrimeCharger250w,
27 PrimePowerBank20k,
28 Solarbank2,
29 Solarbank3,
30 SolixBLEDevice,
31 discover_devices,
32)
33
34MODELS = {
35 "C300": C300,
36 "C300DC": C300DC,
37 "C800": C800,
38 "C1000": C1000,
39 "C1000 G2": C1000G2,
40 "F2000 (767 PowerHouse)": F2000,
41 "F3800": F3800,
42 "Solarbank 2": Solarbank2,
43 "Solarbank 3": Solarbank3,
44 "PrimeCharger160w": PrimeCharger160w,
45 "PrimeCharger250w": PrimeCharger250w,
46 "Prime Power Bank 20k/220w": PrimePowerBank20k,
47 "Unknown": Generic,
48}
49
50
51async def prompt_debug_mode():
52 """
53 Prompt the user to enable/disable debug logging.
54 """
55 while True:
56
57 print("Show debug info? [Y/N]")
58 response = input().lower()
59
60 if response == "y":
61 logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
62 break
63
64 elif response == "n":
65 break
66
67
68async def prompt_select_device(devices: list[BLEDevice]) -> SolixBLEDevice:
69 """
70 Prompt the user to select the device and model they want to connect to.
71
72 :param devices: List of discovered power stations.
73 :returns: Selected power station with initialized class.
74 """
75
76 # Get Bluetooth device
77 print("Which device would you like to connect to?")
78 for i in range(1, len(devices) + 1):
79 device = devices[i - 1]
80 print(f"""{i}. "{device.name}" ({device.address}) """)
81 ble_device = devices[int(await ainput("> ")) - 1]
82
83 # Get model/class of device
84 print("What model is this device?")
85 model_keys = list(MODELS.keys())
86 model_values = list(MODELS.values())
87 for i in range(1, len(MODELS) + 1):
88 print(f"""{i}. "{model_keys[i - 1]}" """)
89 device_class = model_values[int(await ainput("> ")) - 1]
90
91 # Return instantiated device
92 return device_class(ble_device)
93
94
95async def main():
96 """
97 Main program loop.
98
99 This prompts the user for logging mode.
100 Looks for power stations.
101 Asks user to select a power station.
102 Asks user to select the model of the power station.
103 Connects to the power station.
104 Prints status updates of the power station.
105 """
106
107 # Ask user for desired logging mode
108 await prompt_debug_mode()
109
110 # Find power stations
111 devices: list[BLEDevice] = []
112 while not devices:
113
114 print("Looking for power station...")
115 devices = await discover_devices()
116 if not devices:
117 print("No devices found! Trying again...")
118
119 # Ask user to select power station and model
120 device = await prompt_select_device(devices)
121
122 # Register callback
123 def my_callback():
124 """
125 Callback executed by library whenever there is a telemetry update.
126 """
127 print("==== REMOTE STATE CHANGE DETECTED ====")
128 print(device)
129 print("==== END OF STATE CHANGE REPORT ====")
130
131 device.add_callback(my_callback)
132
133 # Connect to device
134 await device.connect()
135
136 # Prompt user for action
137 while True:
138 print("What action would you like to perform?")
139 print("1. Print state")
140 print("2. Exit")
141
142 try:
143 action = int(await ainput("> ")) - 1
144 match action:
145
146 # Print state of device
147 case 0:
148 print(device)
149
150 # Exit program
151 case 1:
152 await device.disconnect()
153 print("Goodbye :)")
154 return
155
156 except ValueError:
157 pass
158
159
160if __name__ == "__main__":
161 asyncio.run(main())