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
|
#!/usr/bin/python3
import struct
import sys
import pycomfoconnect
from pycomfoconnect import zehnder_pb2
"""
Pipe the messages to this programm to decode them. If they are stored in HEX, you should pass them to xxd -r -p first.
You can combine both input and output streams.
Usage:
$ cat captures/raw_*.txt | xxd -r -p | ./manually_decode.py
"""
pdids = {}
rmis = {}
while True:
# Read packet from file
msg_len_buf = sys.stdin.buffer.read(4)
if not msg_len_buf:
break
msg_len = struct.unpack('>L', msg_len_buf)[0]
msg_buf = sys.stdin.buffer.read(msg_len)
# Decode packet
message = pycomfoconnect.Message.decode(msg_len_buf + msg_buf)
if message.cmd.type == zehnder_pb2.GatewayOperation.CnRmiRequestType:
if not message.cmd.reference in rmis:
rmis[message.cmd.reference] = {}
rmis[message.cmd.reference]['tx'] = message.msg.message.hex()
if message.cmd.type == zehnder_pb2.GatewayOperation.CnRmiResponseType:
if not message.cmd.reference in rmis:
rmis[message.cmd.reference] = {}
rmis[message.cmd.reference]['rx'] = message.msg.message.hex()
if message.cmd.type == zehnder_pb2.GatewayOperation.CnRpdoRequestType:
if not message.msg.pdid in pdids:
pdids[message.msg.pdid] = {}
try:
pdids[message.msg.pdid]['tx'].append(message.msg.type)
except KeyError:
pdids[message.msg.pdid]['tx'] = [message.msg.type]
if message.cmd.type == zehnder_pb2.GatewayOperation.CnRpdoConfirmType:
pass
if message.cmd.type == zehnder_pb2.GatewayOperation.CnRpdoNotificationType:
if not message.msg.pdid in pdids:
pdids[message.msg.pdid] = {}
try:
pdids[message.msg.pdid]['rx'].append(message.msg.data.hex())
except KeyError:
pdids[message.msg.pdid]['rx'] = [message.msg.data.hex()]
print("CnRpdoRequestType")
for pdid in pdids:
print(pdid, pdids[pdid])
#
# # print("CnRmiRequestType")
# # for rmi in rmis:
# # print(rmi, rmis[rmi])
#
|
