CT-3 under Linux

BatGeek

Hello,

We wouldlike to use serial communication protocol to interface with CT-3 with Linux. Doyou have API or Libraries for linux? Or could you detail the serial protocol used?

zhoudada

Same here. Is there a Linux version of the Shizuku System Box?
请问有Linux的版本的连接PC的软件吗？

admin

Hello customer,
We only develop software that supports windows.

TheNetStriker

I just rewrote a Python script for the CT-2 I found here for the CT-3:

1. #!/usr/bin/env python
   
2. from serial import Serial as Serial
   
3. 
   
4. from struct import unpack as unpack
   
5. from struct import pack_into as pack_into
   
6. from argparse import ArgumentParser as ArgumentParser
   
7. from argparse import FileType as FileType
   
8. 
   
9. from time import sleep as sleep
   
10. 
    
11. from sys import exit as exit
    
12. 
    
13. # actions along with their description
    
14. __actions = ["list", "read"]
    
15. __action_descriptions = {
    
16.     "list": "Show this list and exit",
    
17.     "read": "Reads energy value from the meter",
    
18. }
    
19. 
    
20. # values for read operation
    
21. __gets = ["all", "voltage", "current", "power", "voltageDP", "voltageDM", "energy"]
    
22. 
    
23. 
    
24. class AVHzY_CT2:
    
25. 
    
26.     def __init__(self, device, action, repeat, time, reads, output):
    
27.         self.__action = action
    
28.         self.__repeat = repeat
    
29.         self.__time = time
    
30.         self.__output = output
    
31.         self.__ser = Serial(device)
    
32. 
    
33.         self.__first_exec = True
    
34.         self.__timestamp = 0
    
35.         self.__energy = 0
    
36.         self.__prev_timestamp = 0
    
37. 
    
38.         if reads == "all":
    
39.             self.__reads = [
    
40.                 "voltage",
    
41.                 "current",
    
42.                 "power",
    
43.                 "voltageDP",
    
44.                 "voltageDM",
    
45.                 "energy",
    
46.             ]
    
47.         else:
    
48.             self.__reads = reads
    
49. 
    
50.         # action handlers in a dictionary
    
51.         self.__action_handlers = {
    
52.             "list": self.__action_list,
    
53.             "read": self.__action_read,
    
54.         }
    
55. 
    
56.     def __del__(self):
    
57.         self.__ser.close()
    
58.         self.__output.close()
    
59. 
    
60.     def __action_list(self):
    
61.         print("Actions list:")
    
62.         for action in __action_descriptions:
    
63.             print("  ", action, "\t", __action_descriptions[action])
    
64. 
    
65.     def xor_checksum(self, byte_array, start, end):
    
66.         checksum = 0
    
67.         for byte in byte_array[start : end + 1]:
    
68.             checksum ^= byte
    
69.         return checksum
    
70. 
    
71.     def __action_read(self):
    
72. 
    
73.         if self.__first_exec:
    
74.             self.__output.write("time")
    
75.             for r in self.__reads:
    
76.                 self.__output.write(",{0}".format(r))
    
77.             self.__output.write("\n")
    
78.             self.__first_exec = False
    
79. 
    
80.         packet = self.__ser.read(35)
    
81.         packetData = unpack("ffffQ", packet[9:33])
    
82. 
    
83.         voltage = packetData[0]
    
84.         current = abs(packetData[1])
    
85.         power = packetData[1] * packetData[0]
    
86.         voltageDP = packetData[2]
    
87.         voltageDM = packetData[3]
    
88.         self.__timestamp = packetData[4]
    
89. 
    
90.         self.__output.write("{0}".format(self.__timestamp))
    
91.         if "voltage" in self.__reads:
    
92.             self.__output.write(",{0}".format(voltage))
    
93.         if "current" in self.__reads:
    
94.             self.__output.write(",{0}".format(current))
    
95.         if "power" in self.__reads:
    
96.             self.__output.write(",{0}".format(power))
    
97.         if "voltageDP" in self.__reads:
    
98.             self.__output.write(",{0}".format(voltageDP))
    
99.         if "voltageDM" in self.__reads:
    
100.             self.__output.write(",{0}".format(voltageDM))
     
101.         if "energy" in self.__reads:
     
102.             if self.__prev_timestamp == 0:
     
103.                 self.__output.write(",0")
     
104.             else:
     
105.                 self.__energy += (
     
106.                     (self.__timestamp - self.__prev_timestamp) * power
     
107.                 ) / 3600000000
     
108.                 self.__output.write(",{0}".format(self.__energy))
     
109.         self.__output.write("\n")
     
110.         self.__output.flush()
     
111. 
     
112.         self.__prev_timestamp = self.__timestamp
     
113. 
     
114.     def perform_action(self):
     
115.         if self.__action == "list":
     
116.             self.__action_list()
     
117.             return
     
118. 
     
119.         # Clear input buffer
     
120.         self.__ser.reset_input_buffer()
     
121. 
     
122.         # Reset device
     
123.         resetCommand = bytearray.fromhex("A5 04 00 00 00 01 07 00 00 06 5A")
     
124.         self.__ser.write(resetCommand)
     
125.         self.__ser.read(11)  # Clear response to command
     
126. 
     
127.         # Reset record timer
     
128.         resetRecordTimerCommand = bytearray.fromhex("A5 04 00 00 00 01 0C 0A 00 07 5A")
     
129.         self.__ser.write(resetRecordTimerCommand)
     
130.         self.__ser.read(11)  # Clear response to command
     
131. 
     
132.         # Start reading
     
133.         startReadCommand = bytearray.fromhex(
     
134.             "A5 08 00 00 00 01 09 0B 00 00 00 00 00 00 5A"
     
135.         )
     
136.         pack_into("I", startReadCommand, 9, self.__time)
     
137.         startReadCommand[13] = self.xor_checksum(startReadCommand, 5, 12)
     
138.         self.__ser.write(startReadCommand)
     
139.         self.__ser.read(11)  # Clear response to command
     
140. 
     
141.         count = 0
     
142.         while count != self.__repeat:
     
143. 
     
144.             try:
     
145.                 self.__action_handlers[self.__action]()
     
146.             except KeyboardInterrupt:
     
147.                 break
     
148. 
     
149.             if self.__repeat != -1:
     
150.                 count += 1
     
151. 
     
152.     def change_output(self, output):
     
153.         self.__output.close()
     
154.         self.__output = output
     
155. 
     
156. 
     
157. def main():
     
158. 
     
159.     # -------------------------------------- OPTION PARSING
     
160.     parser = ArgumentParser(
     
161.         description="Program to interact with the AVHzY CT-2 power meter"
     
162.     )
     
163.     parser.add_argument(
     
164.         "action",
     
165.         metavar="action",
     
166.         choices=__actions,
     
167.         help="The action to perform [choices: %(choices)s]",
     
168.     )
     
169.     parser.add_argument(
     
170.         "-d",
     
171.         "--device",
     
172.         default="/dev/ttyACM0",
     
173.         help="Path to the device [default: %(default)s]",
     
174.     )
     
175.     parser.add_argument(
     
176.         "-r",
     
177.         "--repeat",
     
178.         type=int,
     
179.         default=-1,
     
180.         help="How many times to repeat the operation. Must be in [-1, inf[ (-1: infinite) [default: %(default)s]",
     
181.     )
     
182.     parser.add_argument(
     
183.         "-t",
     
184.         "--time",
     
185.         type=int,
     
186.         default=100,
     
187.         help="The time (in miliseconds) to wait between each action iteration.",
     
188.     )
     
189.     parser.add_argument(
     
190.         "-o",
     
191.         "--output",
     
192.         default="-",
     
193.         type=FileType("w"),
     
194.         help="Where to write output of the action [default: stdout]",
     
195.     )
     
196.     parser.add_argument(
     
197.         "-g",
     
198.         "--get",
     
199.         default="all",
     
200.         choices=__gets,
     
201.         nargs="+",
     
202.         help="For read operation: what to get from power meter [choices: %(choices)s default: %(default)s",
     
203.     )
     
204.     args = parser.parse_args()
     
205. 
     
206.     if args.repeat < -1:
     
207.         print("ERROR: repeat must be in >= -1")
     
208.         parser.print_usage()
     
209.         exit(1)
     
210. 
     
211.     if args.time < 1:
     
212.         print("ERROR: time must be > 1")
     
213.         parser.print_usage()
     
214.         exit(1)
     
215. 
     
216.     AVHzY_CT2(
     
217.         args.device, args.action, args.repeat, args.time, args.get, args.output
     
218.     ).perform_action()
     
219.     exit(0)
     
220. 
     
221. 
     
222. if __name__ == "__main__":
     
223.     main()
     

复制代码

joadswe

Looking promising! @TheNetStriker do u have any more info on the commands sent or any more decoded data?
Works ok for me but a didn't get any timestamp and updated the code to use the power value from the CT-3.

Is there any one that has more information regarding the data sent from the CT-3, eg temperature, energy (mWh, mAh), timestamp. I wasn't able to find any more useful data.

1. from serial import Serial
   
2. from struct import unpack as unpack
   
3. from struct import pack_into as pack_into
   
4. from argparse import ArgumentParser, FileType
   
5. from time import time, sleep, gmtime, localtime, mktime
   
6. from sys import exit
   
7. 
   
8. # actions along with their description, no list function
   
9. __actions = ["read"]
   
10. __action_descriptions = {
    
11.     "read": "Reads energy value from the meter"
    
12. }
    
13. 
    
14. # values for read operation
    
15. __gets = ["all", "voltage", "current", "power", "voltageDP", "voltageDM", "energy"]
    
16. 
    
17. class AVHzY_CT3:
    
18.     def __init__(self, device, action, repeat, time, reads, output):
    
19.         self.__action = action
    
20.         self.__repeat = repeat
    
21.         self.__time = time
    
22.         self.__output = output
    
23.         self.__ser = Serial(device)
    
24. 
    
25.         self.__first_exec = True
    
26.         self.__timestamp = 0
    
27.         self.__energy = 0
    
28.         self.__prev_timestamp = 0
    
29. 
    
30.         if reads == "all":
    
31.             self.__reads = ["voltage", "current", "power", "voltageDP", "voltageDM", "energy"]
    
32.         else:
    
33.             self.__reads = reads
    
34. 
    
35.         # action handlers in a dictionary
    
36.         self.__action_handlers = {
    
37.             "list": self.__action_list,
    
38.             "read": self.__action_read,
    
39.         }
    
40.    
    
41.     def __del__(self):
    
42.         self.__ser.close()
    
43.         self.__output.close()
    
44.    
    
45.     def __action_list(self):
    
46.         print("Actions list:")
    
47.         for action in __action_descriptions:
    
48.             print("  ", action, "\t", __action_descriptions[action])
    
49. 
    
50.     def xor_checksum(self, byte_array, start, end):
    
51.         checksum = 0
    
52.         for byte in byte_array[start : end + 1]:
    
53.             checksum ^= byte
    
54.         return checksum
    
55. 
    
56.     def __action_read(self):
    
57.         if self.__first_exec:
    
58.             self.__output.write("time")
    
59.             for r in self.__reads:
    
60.                 self.__output.write(",{0}".format(r))
    
61.             self.__output.write("\n")
    
62.             self.__first_exec = False
    
63.         
    
64.         self.__ser.flushInput()
    
65.         while True:
    
66.             packet = self.__ser.read(30)
    
67.             if len(packet) == 30:
    
68.                 break
    
69. 
    
70.         packetData = unpack("<fffff", packet[9:29])  # Adjust byte order if necessary
    
71.         
    
72.         voltage = packetData[0]
    
73.         current = packetData[1]
    
74.         power = packetData[2]
    
75.         voltageDP = packetData[3]
    
76.         voltageDM = packetData[4]      
    
77.         self.__timestamp = int(time()*1000)
    
78.         
    
79.         self.__output.write("Timestamp: {0}".format(self.__timestamp))
    
80.         if "voltage" in self.__reads:
    
81.             self.__output.write(", Voltage: {0:.6f}".format(voltage))
    
82.         if "current" in self.__reads:
    
83.             self.__output.write(", Current: {0:.6f}".format(current))
    
84.         if "power" in self.__reads:
    
85.             self.__output.write(", Power: {0:.6f}".format(power))
    
86.         if "voltageDP" in self.__reads:
    
87.             self.__output.write(", D+: {0:.6f}".format(voltageDP))
    
88.         if "voltageDM" in self.__reads:
    
89.             self.__output.write(", D-: {0:.6f}".format(voltageDM))
    
90.         
    
91.         if "energy" in self.__reads:
    
92.             if self.__prev_timestamp == 0:
    
93.                 self.__output.write(",0")
    
94.             else:
    
95.                 self.__energy += ((self.__timestamp - self.__prev_timestamp) * power) / 3600000000
    
96.                 self.__output.write(", Energy: {0:.6f}".format(self.__energy))
    
97.         
    
98.         self.__output.write("\n")
    
99.         self.__output.flush()
    
100. 
     
101.         self.__prev_timestamp = self.__timestamp
     
102. 
     
103.     def perform_action(self):
     
104. 
     
105.         self.__ser.reset_input_buffer()
     
106. 
     
107.         resetCommand = bytearray.fromhex("A5 04 00 00 00 01 07 00 00 06 5A")
     
108.         self.__ser.write(resetCommand)
     
109.         self.__ser.read(11)
     
110. 
     
111.         resetRecordTimerCommand = bytearray.fromhex("A5 04 00 00 00 01 0C 0A 00 07 5A")
     
112.         self.__ser.write(resetRecordTimerCommand)
     
113.         self.__ser.read(11)
     
114. 
     
115.         startReadCommand = bytearray.fromhex(
     
116.             "A5 08 00 00 00 01 09 0B 00 00 00 00 00 00 5A"
     
117.         )
     
118.         pack_into("I", startReadCommand, 9, self.__time)
     
119.         startReadCommand[13] = self.xor_checksum(startReadCommand, 5, 12)
     
120.         self.__ser.write(startReadCommand)
     
121.         self.__ser.read(11)
     
122. 
     
123.         count = 0
     
124.         while count != self.__repeat:
     
125.             try:
     
126.                 self.__action_handlers[self.__action]()
     
127.             except KeyboardInterrupt:
     
128.                 break
     
129. 
     
130.             if self.__repeat != -1:
     
131.                 count += 1
     
132.             sleep(self.__time / 1000.0)
     
133. 
     
134.     def change_output(self, output):
     
135.         self.__output.close()
     
136.         self.__output = output
     
137. 
     
138. 
     
139. def main():
     
140.     parser = ArgumentParser(description="Program to interact with the AVHzY CT-2 power meter")
     
141.     parser.add_argument("action", metavar="action", choices=__actions, help="The action to perform [choices: %(choices)s]")
     
142.     parser.add_argument("-d", "--device", default="/dev/ttyACM0", help="Path to the device [default: %(default)s]")
     
143.     parser.add_argument("-r", "--repeat", type=int, default=-1, help="How many times to repeat the operation. Must be in [-1, inf[ (-1: infinite) [default: %(default)s]")
     
144.     parser.add_argument("-t", "--time", type=int, default=100, help="The time (in milliseconds) to wait between each action iteration.")
     
145.     parser.add_argument("-o", "--output", default="-", type=FileType("w"), help="Where to write output of the action [default: stdout]")
     
146.     parser.add_argument("-g", "--get", default="all", choices=__gets, nargs="+", help="For read operation: what to get from power meter [choices: %(choices)s default: %(default)s")
     
147.     args = parser.parse_args()
     
148. 
     
149.     if args.repeat < -1:
     
150.         print("ERROR: repeat must be in >= -1")
     
151.         parser.print_usage()
     
152.         exit(1)
     
153. 
     
154.     if args.time < 1:
     
155.         print("ERROR: time must be > 1")
     
156.         parser.print_usage()
     
157.         exit(1)
     
158. 
     
159.     AVHzY_CT3(args.device, args.action, args.repeat, args.time, args.get, args.output).perform_action()
     
160.     exit(0)
     
161. 
     
162. if __name__ == '__main__':
     
163.     main()

复制代码