forked from dgibson/python-smadata2
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathsma2-explore
executable file
·360 lines (303 loc) · 12.4 KB
/
sma2-explore
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
#! /usr/bin/python3
#
# sma2-explore - Interactive tool for analyzing SMAData2 protocol
# Copyright (C) 2014 David Gibson <[email protected]>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#AF checked and file is python 3, so not needed
#from __future__ import print_function
#from __future__ import division
import sys
import os
import signal
#import gnureadline #readline is deprecated
#import pyreadline
import threading
import time
from smadata2.inverter.smabluetooth import Connection
from smadata2.inverter.smabluetooth import OTYPE_HELLO, OTYPE_ERROR, \
OTYPE_VARVAL, OTYPE_GETVAR, OTYPE_PPP, OTYPE_PPP2, OVAR_SIGNAL
from smadata2.inverter.smabluetooth import bytes2int, int2bytes16
class Quit(Exception):
def __init__(self):
super(Quit, self).__init__("Quit at user request")
def hexdump(data, prefix):
s = ''
for i, b in enumerate(data):
if (i % 16) == 0:
s += '%s%04x: ' % (prefix, i)
s += '%02X' % b
if (i % 16) == 15:
s += '\n'
elif (i % 16) == 7:
s += '-'
else:
s += ' '
if s and (s[-1] == '\n'):
s = s[:-1]
return s
def dump_outer(prefix, from_, to_, type_, payload):
print("%s%s -> %s TYPE %02X" % (prefix, from_, to_, type_))
prefix = prefix + " "
if type_ == OTYPE_HELLO:
print("%sHELLO!" % prefix)
elif type_ == OTYPE_ERROR:
print("%sERROR in previous packet:" % prefix)
print(hexdump(payload[4:], prefix + " "))
def dump_ppp_raw(prefix, payload):
info = ""
if payload[0] == 0x7e:
info += " frame begins"
if payload[-1] == 0x7e:
info += " frame ends"
print("%sPartial PPP data%s" % (prefix, info))
def dump_ppp(prefix, protocol, payload):
print("%sPPP frame; protocol 0x%04x [%d bytes]"
% (prefix, protocol, len(payload)))
print(hexdump(payload, prefix + " "))
def a65602str(addr):
return "%02X.%02X.%02X.%02X.%02X.%02X" % tuple(addr)
#called from
def dump_6560(prefix, from2, to2, a2, b1, b2, c1, c2, tag,
type_, subtype, arg1, arg2, extra,
response, error, pktcount, first):
print("%sSMA INNER PROTOCOL PACKET" % prefix)
print("%s %s => %s" % (prefix, a65602str(from2), a65602str(to2)))
print("%s control %02X %02X %02X %02X %02X"
% (prefix, a2, b1, b2, c1, c2))
if error:
print("%s ERROR!! code 0x%04x" % (prefix, error))
s = []
if first:
s.append('first')
if pktcount:
s.append('%d packets left' % pktcount)
else:
s.append('last')
s = ', '.join(s)
print("%s tag %04x (%s)" % (prefix, tag, s))
if response:
cr = "response"
else:
cr = "command"
print("%s %s 0x%04x subtype 0x%04x" % (prefix, cr, type_, subtype))
print(hexdump(extra, prefix + " "))
class SMAData2CLI(Connection):
"""CLI class overrides the Connection class defined in smabluetooth
Implements most of the same functions, but calls a dump_ function first
the dump function prints a formatted packet format to the stdout
"""
def __init__(self, addr):
super(SMAData2CLI, self).__init__(addr)
print("Connected %s -> %s"
% (self.local_addr, self.remote_addr))
self.rxpid = None
def __del__(self):
if self.rxpid:
os.kill(self.rxpid, signal.SIGTERM)
# Function call order
# wait() > rx() > rx-raw() > rx_outer() > rx_ppp_raw() > rx_ppp() >rx_6560 > rxfilter_6560
def rxloop(self):
while True:
#print ("Thread: running")
self.rx()
# todo can we use https://docs.python.org/3.6/library/multiprocessing.html or asyncio so this works in windows?
# This attempt to work with threading did not work on windows - does not return control to the main thread
# def start_rxthread(self):
# print('start_rxthread')
# print(__name__)
# print(self.rxpid)
# if __name__ == '__main__':
# p = threading.Thread(target = self.rxloop(), args=(), daemon=True, name="rx_listen")
# p.start()
# print('start_main_thread')
def start_rxthread(self):
self.rxpid = os.fork() # creates another process which will resume at exactly the same place as this one
if self.rxpid == 0: # is zero for child process
while True:
try:
self.rxloop()
except Exception as e:
print(e)
def rx_raw(self, pkt):
print("\n" + hexdump(pkt, "Rx< "))
super(SMAData2CLI, self).rx_raw(pkt)
def rx_outer(self, from_, to_, type_, payload):
dump_outer("Rx< ", from_, to_, type_, payload)
if type_ == OTYPE_VARVAL:
varid = bytes2int(payload[:2])
print("Rx< VARVAL 0x%02x" % varid)
if varid == OVAR_SIGNAL:
print("Rx< Signal level %0.1f%%"
% (payload[4] / 255 * 100))
elif (type_ in [OTYPE_PPP, OTYPE_PPP2]):
dump_ppp_raw("Rx< ", payload)
super(SMAData2CLI, self).rx_outer(from_, to_,
type_, payload)
def rx_ppp(self, from_, protocol, payload):
dump_ppp("Rx< ", protocol, payload)
super(SMAData2CLI, self).rx_ppp(from_, protocol, payload)
# in rx_ppp this is set: error = bytes2int(payload[18:20])
def rx_6560(self, from2, to2, a2, b1, b2, c1, c2, tag,
type_, subtype, arg1, arg2, extra,
response, error, pktcount, first):
dump_6560("Rx< ", from2, to2, a2, b1, b2, c1, c2, tag,
type_, subtype, arg1, arg2, extra,
response, error, pktcount, first)
super(SMAData2CLI, self).rx_6560(from2, to2, a2, b1, b2, c1, c2,
tag, type_, subtype,
arg1, arg2,
extra, response, error,
pktcount, first)
def tx_raw(self, pkt):
super(SMAData2CLI, self).tx_raw(pkt)
print("\n" + hexdump(pkt, "Tx> "))
def tx_outer(self, from_, to_, type_, payload):
"""Builds a SMA Level 1 packet from supplied Level 2, calls tx_raw to transmit
:param from_: str source Bluetooth address in string representation
:param to_: str destination Bluetooth address in string representation
:param type_: int the command to send, e.g. OTYPE_PPP = 0x01 L2 Packet start
:param payload: bytearray Data or payload in Level 1 packet
:return:
"""
super(SMAData2CLI, self).tx_outer(from_, to_,
type_, payload)
dump_outer("Tx> ", from_, to_, type_, payload)
if type_ == OTYPE_GETVAR:
varid = bytes2int(payload[:2])
print("Tx> GETVAR 0x%02x" % varid)
def tx_ppp(self, to_, protocol, payload):
super(SMAData2CLI, self).tx_ppp(to_, protocol, payload)
dump_ppp("Tx> ", protocol, payload)
def tx_6560(self, from2, to2, a2, b1, b2, c1, c2, tag,
type_, subtype, arg1, arg2, extra=bytearray(),
response=False, error=0, pktcount=0, first=True):
super(SMAData2CLI, self).tx_6560(from2, to2, a2, b1, b2, c1, c2,
tag, type_, subtype,
arg1, arg2, extra, response,
error, pktcount, first)
dump_6560("Tx> ", from2, to2, a2, b1, b2, c1, c2, tag,
type_, subtype, arg1, arg2, extra,
response, error, pktcount, first)
def cli(self):
while True:
sys.stdout.write("SMA2 %s >> " % self.remote_addr)
try:
line = input().split()
except EOFError:
return
if not line:
continue
cmd = 'cmd_' + line[0].lower()
try:
getattr(self, cmd)(*line[1:])
except Quit:
return
except Exception as e:
print("ERROR! %s" % e)
def cmd_quit(self):
raise Quit()
def cmd_raw(self, *args):
pkt = bytearray([int(x, 16) for x in args])
self.tx_raw(pkt)
def parse_addr(self, addr):
if addr.lower() == "zero":
return "00:00:00:00:00:00"
elif addr.lower() == "local":
return self.local_addr
elif addr.lower() == "remote":
return self.remote_addr
elif addr.lower() == "bcast":
return "ff:ff:ff:ff:ff:ff"
else:
return addr
def cmd_send(self, from_, to_, type_, *args):
from_ = self.parse_addr(from_)
to_ = self.parse_addr(to_)
type_ = int(type_, 16)
payload = bytearray([int(x, 16) for x in args])
self.tx_outer(from_, to_, type_, payload)
def cmd_hello(self):
"""Level 1 hello command responds to the SMA with the same data packet sent,
the smabluetooth.hello function checks hello packet NetID from inverter and returns the same one
This is hardcoded, as the inbound packet has not been read.
Byte 5 below was x01, needed x04 - inspect the packet from your inverter.
"""
self.tx_outer("00:00:00:00:00:00", self.remote_addr, OTYPE_HELLO,
# bytearray(b'\x00\x04\x70\x00\x01\x00\x00\x00\x00\x01\x00\x00\x00'))
bytearray(b'\x00\x04\x70\x00\x04\x00\x00\x00\x00\x01\x00\x00\x00')) #from TL5000
def cmd_getvar(self, varid):
"""Level 1 getvar requests the value or a variable from the SMA inverter
values include:
"""
varid = int(varid, 16)
self.tx_outer("00:00:00:00:00:00", self.remote_addr, OTYPE_GETVAR,
int2bytes16(varid))
def cmd_ppp(self, protocol, *args):
protocol = int(protocol, 0)
payload = bytearray([int(x, 16) for x in args])
self.tx_ppp("ff:ff:ff:ff:ff:ff", protocol, payload)
def cmd_send2(self, *args):
"""Sends a Level 2 packet request to the inverter
COMMAND:
A2: A0
B1,B2: 00 00
C1,C2: 00 00
Type: 0200
Subtype: 5400
Arg1: 0x00260100
Arg2: 0x002601ff
:param args:
:return:
"""
bb = [int(x, 16) for x in args]
a2 = bb[0]
b1, b2 = bb[1], bb[2]
c1, c2 = bb[3], bb[4]
type_, subtype = bb[5], bb[6]
arg1, arg2 = bb[7], bb[8]
extra = bytearray(bb[9:])
self.tx_6560(self.local_addr2, self.BROADCAST2,
a2, b1, b2, c1, c2, self.gettag(),
type_, subtype, arg1, arg2, extra)
#AF added b'
def cmd_logon(self, password=b'0000', timeout=900):
timeout = int(timeout)
self.tx_logon(password, timeout)
def cmd_gdy(self):
self.tx_gdy()
def cmd_yield(self):
self.tx_yield()
def cmd_spotacvoltage(self):
self.tx_spotacvoltage()
def cmd_historic(self, fromtime=None, totime=None):
if fromtime is None and totime is None:
fromtime = 1356958800 # 1 Jan 2013
totime = int(time.time()) # now
else:
fromtime = int(fromtime)
totime = int(totime)
self.tx_historic(fromtime, totime)
if __name__ == '__main__':
if len(sys.argv) != 2:
print("Usage: %s <BD addr>" % sys.argv[0], file=sys.stderr)
sys.exit(1)
cli = SMAData2CLI(sys.argv[1])
# attempt to thread under Windows
# p = threading.Thread(target=cli.rxloop(), args=(), daemon=True, name="rx_listen")
# p = threading.Thread(target=cli.cli(), args=(), daemon=True, name="rx_listen")
# p.start()
cli.start_rxthread()
cli.cli()