File: analysedump.py

package info (click to toggle)
galileo 0.5.1-4
  • links: PTS, VCS
  • area: main
  • in suites: stretch
  • size: 392 kB
  • ctags: 578
  • sloc: python: 3,462; xml: 23; makefile: 14
file content (239 lines) | stat: -rwxr-xr-x 6,204 bytes parent folder | download | duplicates (2)
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
#!/usr/bin/env python

import os
import sys
import time
import base64


def readlog(f):
    """ input is from f in the format of lines of long string starting
    with a tab ('\t') representing the hexadcimal representation of the data
    (megadump)"""
    d = []
    for line in f:
        if line[0] != '\t':
            if d:
                return d
            continue
        line = line.strip()
        for i in range(0, len(line), 2):
            d.append(int(line[i:i + 2], 16))
    return d


def readdump(f):
    """ imput is from ./galileo.py """
    d = []
    r = []
    current = d
    for line in f:
        if line.strip() == '':
            current = r
            continue
        current.extend(int(x, 16) for x in line.strip().split())
    return d, r


def a2s(array):
    """ array of int to string """
    return ''.join(chr(c) for c in array)


def a2x(array):
    """ array of int to hex representation """
    return ' '.join("%02X" % i for i in array)


def a2lsbi(array):
    """ array to int (LSB first) """
    integer = 0
    for i in range(len(array) - 1, -1, -1):
        integer *= 256
        integer += array[i]
#        print a2x(array), hex(integer)
    return integer


def a2msbi(array):
    """ array to int (MSB first) """
    integer = 0
    for i in range(len(array)):
        integer *= 256
        integer += array[i]
    return integer


def header(data):
    index = 40
    walkStrideLen = a2lsbi(data[index:index + 2])
    index += 2
    runStrideLen = a2lsbi(data[index:index + 2])
    index += 2
    print "Stride lengths: %dmm, %dmm" % (walkStrideLen, runStrideLen)
    print a2x(data[index:index + 4])
    index += 4
    # empirical value
    index += 12

    if index >= len(data): return
    # Greetings
    print "Greetings: '%s'" % a2s(data[index:index + 10])
    index += 10

    # Cheering
    print "Cheering"
    for i in range(3):
        print "'%s'" % a2s(data[index:index + 10])
        index += 10


def first_field(rec_len):
    def unknown(data):
        assert data[:3] == [END, END, 0xdd], a2x(data[:3])
        index = 3
        while index < len(data) - 1:
            tstamp = a2lsbi(data[index:index + 4])
            print time.strftime("%x %X", time.localtime(tstamp)), hex(tstamp)
            index += 4
            print "\t%s" % a2x(data[index:index + rec_len])
            index += rec_len
    return unknown


def minutely(rec_len):
    """ this analyses the minute-by-minute information
    """
    def minutes(data):
        assert data[:3] == [END, END, 0xdd], a2x(data[:3])
        index = 3
        tstamp = 0
        while index < len(data) - 1:
            if not (data[index] & 0x80):
                tstamp = a2msbi(data[index:index + 4])
                index += 4
            else:
                print time.strftime("%x %X", time.localtime(tstamp)), a2x(data[index:index + rec_len])
                tstamp += 60
                index += rec_len
    return minutes


def stairs(data):
    """ Looks like stairs informations are put here """
    assert data[:3] == [END, END, 0xdd], a2x(data)
    index = 3
    index = 3
    tstamp = 0
    while index < len(data) - 1:
        if not (data[index] & 0x80):
            tstamp = a2msbi(data[index:index + 4])
            index += 4
        else:
            if data[index] != 0x80:
                #print a2x([array[index]])
                index += 1
            print time.strftime("%x %X", time.localtime(tstamp)), a2x(data[index:index + 2])
            tstamp += 60
            index += 2


def daily(data):
    if len(data) == 2:
        assert data == [END, END], a2x(data)
        return
    assert data[:3] == [END, END, 0xdd], a2x(data[:3])
    index = 3
    while index < len(data) - 1:
        tstamp = a2lsbi(data[index:index + 4])
        index += 4
        print time.strftime("%x %X", time.localtime(tstamp)), a2x(data[index:index + 12])
        index += 12


def footer(data):
    assert len(data) == 9, data
    print 'Dump size: %d bytes' % a2lsbi(data[5:7])
    print a2x(data)

ESC = 0xdb
END = 0xc0
ESC_ = {0xdc: END, 0xdd: ESC}


def unSLIP(data):
    """ This remove SLIP escaping and yield the parts
    The magic are: The first part doesn't ends with 0xC0
    there are empty parts
    >>> list(unSLIP([1, 2, 0xc0, 5, 4, 0xc0, 0xc0, 8, 4]))
    [[1, 2], [192, 5, 4, 192], [192, 8, 4]]
    >>> list(unSLIP([12, 0xc0, 0, 0, 0xc0, 1, 2, 0xc0, 8, 9]))
    """
    first = True
    part = []
    escape = False
    for c in data:
#        print "%x" % c
        if not escape:
            if c == ESC and part and part[0] == END:
                escape = True
            else:
                part.append(c)
                if c == END:
                    if len(part) != 1:
                        if first or (part[0] != END):
                            yield part[:-1]
                            part = [part[-1]]
                            first = False
                        else:
                            yield part
                            part = []
        else:
            part.append(ESC_[c])
            escape = False
    yield part


def analyse(data):

    def onscreen(data):
        print a2x(data)

    def skip(data):
        pass

    display = [onscreen] * 20

    analyses_ZIP = [header, first_field(9), minutely(3), daily, footer]

    analyses_ONE = [header, first_field(11), minutely(4), stairs, daily, skip, onscreen, skip, footer]

    analyses = {
        0x26: analyses_ONE,
        0xF4: analyses_ZIP,
    }.get(data[0], display)

    for i, part in enumerate(unSLIP(data)):
        f = analyses[i]
        print "%s (%d): %d bytes" % (f.__name__, i, len(part))
        f(part)


def analysedump(dump_dir, index):
    for root, dirs, files in os.walk(dir):
        file = sorted(files)[idx]
    print "Analysing %s" % file
    with open(os.path.join(root, file)) as f:
        dump, response = readdump(f)
        analyse(dump)

if __name__ == "__main__":
    if len(sys.argv) == 1:
        dump, response = readdump(sys.stdin)
        analyse(dump)
    else:
        dir = sys.argv[1]
        idx = -1
        if len(sys.argv) > 2:
            idx = int(sys.argv[2])
        analysedump(dir, idx)