decoders/ook_oregon/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2018 Steve R <steversig@virginmedia.com>
   5 ##
   6 ## This program is free software; you can redistribute it and/or modify
   7 ## it under the terms of the GNU General Public License as published by
   8 ## the Free Software Foundation; either version 2 of the License, or
   9 ## (at your option) any later version.
  10 ##
  11 ## This program is distributed in the hope that it will be useful,
  12 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 ## GNU General Public License for more details.
  15 ##
  16 ## You should have received a copy of the GNU General Public License
  17 ## along with this program; if not, see <http://www.gnu.org/licenses/>.
  18 ##
  19
  20 import sigrokdecode as srd
  21 import math
  22 from .lists import *
  23
  24 class Decoder(srd.Decoder):
  25     api_version = 3
  26     id = 'ook_oregon'
  27     name = 'Oregon'
  28     longname = 'Oregon Scientific'
  29     desc = 'Oregon Scientific weather sensor protocol.'
  30     license = 'gplv2+'
  31     inputs = ['ook']
  32     outputs = []
  33     tags = ['Sensor']
  34     annotations = (
  35         ('bit', 'Bit'),
  36         ('field', 'Field'),
  37         ('l2', 'Level 2'),
  38         ('pre', 'Preamble'),
  39         ('syn', 'Sync'),
  40         ('id', 'SensorID'),
  41         ('ch', 'Channel'),
  42         ('roll', 'Rolling code'),
  43         ('f1', 'Flags1'),
  44     )
  45     annotation_rows = (
  46         ('bits', 'Bits', (0,)),
  47         ('fields', 'Fields', (1, 3, 4)),
  48         ('l2', 'Level 2', (2,)),
  49     )
  50     binary = (
  51         ('data-hex', 'Hex data'),
  52     )
  53     options = (
  54         {'id': 'unknown', 'desc': 'Unknown type is', 'default': 'Unknown',
  55          'values': ('Unknown', 'Temp', 'Temp_Hum', 'Temp_Hum1', 'Temp_Hum_Baro',
  56                     'Temp_Hum_Baro1', 'UV', 'UV1', 'Wind', 'Rain', 'Rain1')},
  57     )
  58
  59     def __init__(self):
  60         self.reset()
  61
  62     def reset(self):
  63         self.decoded = [] # Local cache of decoded OOK.
  64         self.skip = None
  65
  66     def start(self):
  67         self.out_ann = self.register(srd.OUTPUT_ANN)
  68         self.out_binary = self.register(srd.OUTPUT_BINARY)
  69         self.unknown = self.options['unknown']
  70
  71     def putx(self, data):
  72         self.put(self.ss, self.es, self.out_ann, data)
  73
  74     def dump_oregon_hex(self, start, finish):
  75         nib = self.decoded_nibbles
  76         hexstring = ''
  77         for x in nib:
  78             hexstring += str(x[3]) if x[3] != '' else ' '
  79         s = 'Oregon ' + self.ver + ' \"' + hexstring.upper() + '\"\n'
  80         self.put(start, finish, self.out_binary,
  81                 [0, bytes([ord(c) for c in s])])
  82
  83     def oregon_put_pre_and_sync(self, len_pream, len_sync, ver):
  84         ook = self.decoded
  85         self.decode_pos = len_pream
  86         self.ss, self.es = ook[0][0], ook[self.decode_pos][0]
  87         self.putx([1, ['Oregon ' + ver + ' Preamble', ver + ' Preamble',
  88                         ver + ' Pre', ver]])
  89         self.decode_pos += len_sync
  90         self.ss, self.es = ook[len_pream][0], ook[self.decode_pos][0]
  91         self.putx([1, ['Sync', 'Syn', 'S']])
  92
  93         # Strip off preamble and sync bits.
  94         self.decoded = self.decoded[self.decode_pos:]
  95         self.ookstring = self.ookstring[self.decode_pos:]
  96         self.ver = ver
  97
  98     def oregon(self):
  99         self.ookstring = ''
 100         self.decode_pos = 0
 101         ook = self.decoded
 102         for i in range(len(ook)):
 103             self.ookstring += ook[i][2]
 104         if '10011001' in self.ookstring[:40]:
 105             (preamble, data) = self.ookstring.split('10011001', 1)
 106             if len(data) > 0 and len(preamble) > 16:
 107                 self.oregon_put_pre_and_sync(len(preamble), 8, 'v2.1')
 108                 self.oregon_v2()
 109         elif 'E1100' in self.ookstring[:17]:
 110             (preamble, data) = self.ookstring.split('E1100', 1)
 111             if len(data) > 0 and len(preamble) <= 12:
 112                 self.oregon_put_pre_and_sync(len(preamble), 5, 'v1')
 113                 self.oregon_v1()
 114         elif '0101' in self.ookstring[:28]:
 115             (preamble, data) = self.ookstring.split('0101', 1)
 116             if len(data) > 0 and len(preamble) > 12:
 117                 self.oregon_put_pre_and_sync(len(preamble), 4, 'v3')
 118                 self.oregon_v3()
 119         elif len(self.ookstring) > 16: # Ignore short packets.
 120             error_message = 'Not Oregon or wrong preamble'
 121             self.ss, self.es = ook[0][0], ook[len(ook) - 1][1]
 122             self.putx([1,[error_message]])
 123
 124     def oregon_v1(self):
 125         ook = self.decoded
 126         self.decode_pos = 0
 127         self.decoded_nibbles = []
 128         if len(self.decoded) >= 32: # Check there are at least 8 nibbles.
 129             self.oregon_put_nib('RollingCode', ook[self.decode_pos][0],
 130                                 ook[self.decode_pos + 3][1], 4)
 131             self.oregon_put_nib('Ch', ook[self.decode_pos][0],
 132                                 ook[self.decode_pos + 3][1], 4)
 133             self.oregon_put_nib('Temp', ook[self.decode_pos][0],
 134                                 ook[self.decode_pos + 15][1], 16)
 135             self.oregon_put_nib('Checksum', ook[self.decode_pos][0],
 136                                 ook[self.decode_pos + 7][1], 8)
 137
 138             self.dump_oregon_hex(ook[0][0], ook[len(ook) - 1][1])
 139
 140             # L2 decode.
 141             self.oregon_temp(2)
 142             self.oregon_channel(1)
 143             self.oregon_battery(2)
 144             self.oregon_checksum_v1()
 145
 146     def oregon_v2(self): # Convert to v3 format - discard odd bits.
 147         self.decode_pos = 0
 148         self.ookstring = self.ookstring[1::2]
 149         for i in range(len(self.decoded)):
 150             if i % 2 == 1:
 151                 self.decoded[i][0] = self.decoded[i - 1][0] # Re-align start pos.
 152         self.decoded = self.decoded[1::2] # Discard left hand bits.
 153         self.oregon_v3() # Decode with v3 decoder.
 154
 155     def oregon_nibbles(self, ookstring):
 156         num_nibbles = int(len(ookstring) / 4)
 157         nibbles = []
 158         for i in range(num_nibbles):
 159             nibble = ookstring[4 * i : 4 * i + 4]
 160             nibble = nibble[::-1] # Reversed from right.
 161             nibbles.append(nibble)
 162         return nibbles
 163
 164     def oregon_put_nib(self, label, start, finish, numbits):
 165         param = self.ookstring[self.decode_pos:self.decode_pos + numbits]
 166         param = self.oregon_nibbles(param)
 167         if 'E' in ''.join(param): # Blank out fields with errors.
 168             result = ''
 169         else:
 170             result = hex(int(''.join(param), 2))[2:]
 171             if len(result) < numbits / 4: # Reinstate leading zeros.
 172                 result = '0' * (int(numbits / 4) - len(result)) + result
 173         if label != '':
 174             label += ': '
 175         self.put(start, finish, self.out_ann, [1, [label + result, result]])
 176         if label == '': # No label - use nibble position.
 177             label = int(self.decode_pos / 4)
 178         for i in range(len(param)):
 179             ss = self.decoded[self.decode_pos + (4 * i)][0]
 180             es = self.decoded[self.decode_pos + (4 * i) + 3][1]
 181             # Blank out nibbles with errors.
 182             result = '' if ('E' in param[i]) else hex(int(param[i], 2))[2:]
 183             # Save nibbles for L2 decoder.
 184             self.decoded_nibbles.append([ss, es, label, result])
 185         self.decode_pos += numbits
 186
 187     def oregon_v3(self):
 188         self.decode_pos = 0
 189         self.decoded_nibbles = []
 190         ook = self.decoded
 191
 192         if len(self.decoded) >= 32: # Check there are at least 8 nibbles.
 193             self.oregon_put_nib('SensorID', ook[self.decode_pos][0],
 194                                 ook[self.decode_pos + 16][0], 16)
 195             self.oregon_put_nib('Ch', ook[self.decode_pos][0],
 196                                 ook[self.decode_pos + 3][1], 4)
 197             self.oregon_put_nib('RollingCode', ook[self.decode_pos][0],
 198                                 ook[self.decode_pos + 7][1], 8)
 199             self.oregon_put_nib('Flags1', ook[self.decode_pos][0],
 200                                 ook[self.decode_pos + 3][1], 4)
 201
 202             rem_nibbles = len(self.ookstring[self.decode_pos:]) // 4
 203             for i in range(rem_nibbles): # Display and save rest of nibbles.
 204                 self.oregon_put_nib('', ook[self.decode_pos][0],
 205                                     ook[self.decode_pos + 3][1], 4)
 206             self.dump_oregon_hex(ook[0][0], ook[len(ook) - 1][1])
 207             self.oregon_level2() # Level 2 decode.
 208         else:
 209             error_message = 'Too short to decode'
 210             self.put(ook[0][0], ook[-1][1], self.out_ann, [1, [error_message]])
 211
 212     def oregon_put_l2_param(self, offset, digits, dec_point, pre_label, label):
 213         nib = self.decoded_nibbles
 214         result = 0
 215         out_string = ''.join(str(x[3]) for x in nib[offset:offset + digits])
 216         if len(out_string) == digits:
 217             for i in range(dec_point, 0, -1):
 218                 result += int(nib[offset + dec_point - i][3], 16) / pow(10, i)
 219             for i in range(dec_point, digits):
 220                 result += int(nib[offset + i][3], 16) * pow(10, i - dec_point)
 221             result = '%g' % (result)
 222         else:
 223             result = ''
 224         es = nib[offset + digits - 1][1]
 225         if label == '\u2103':
 226             es = nib[offset + digits][1] # Align temp to include +/- nibble.
 227         self.put(nib[offset][0], es, self.out_ann,
 228                 [2, [pre_label + result + label, result]])
 229
 230     def oregon_temp(self, offset):
 231         nib = self.decoded_nibbles
 232         if nib[offset + 3][3] != '':
 233             temp_sign = str(int(nib[offset + 3][3], 16))
 234             temp_sign = '-' if temp_sign != '0' else '+'
 235         else:
 236             temp_sign = '?'
 237         self.oregon_put_l2_param(offset, 3, 1, temp_sign, '\u2103')
 238
 239     def oregon_baro(self, offset):
 240         nib = self.decoded_nibbles
 241         baro = ''
 242         if not (nib[offset + 2][3] == '' or nib[offset + 1][3] == ''
 243            or nib[offset][3] == ''):
 244             baro = str(int(nib[offset + 1][3] + nib[offset][3], 16) + 856)
 245         self.put(nib[offset][0], nib[offset + 3][1],
 246                  self.out_ann, [2, [baro + ' mb', baro]])
 247
 248     def oregon_wind_dir(self, offset):
 249         nib = self.decoded_nibbles
 250         if nib[offset][3] != '':
 251             w_dir = int(int(nib[offset][3], 16) * 22.5)
 252             w_compass = dir_table[math.floor((w_dir + 11.25) / 22.5)]
 253             self.put(nib[offset][0], nib[offset][1], self.out_ann,
 254                 [2, [w_compass + ' (' + str(w_dir) + '\u00b0)', w_compass]])
 255
 256     def oregon_channel(self, offset):
 257         nib = self.decoded_nibbles
 258         channel = ''
 259         if nib[offset][3] != '':
 260             ch = int(nib[offset][3], 16)
 261             if self.ver != 'v3': # May not be true for all v2.1 sensors.
 262                 if ch != 0:
 263                     bit_pos = 0
 264                     while ((ch & 1) == 0):
 265                         bit_pos += 1
 266                         ch = ch >> 1
 267                     if self.ver == 'v2.1':
 268                         bit_pos += 1
 269                     channel = str(bit_pos)
 270             elif self.ver == 'v3': # Not sure if this applies to all v3's.
 271                 channel = str(ch)
 272         if channel != '':
 273             self.put(nib[offset][0], nib[offset][1],
 274                      self.out_ann, [2, ['Ch ' + channel, channel]])
 275
 276     def oregon_battery(self, offset):
 277         nib = self.decoded_nibbles
 278         batt = 'OK'
 279         if nib[offset][3] != '':
 280             if (int(nib[offset][3], 16) >> 2) & 0x1 == 1:
 281                 batt = 'Low'
 282             self.put(nib[offset][0], nib[offset][1],
 283                      self.out_ann, [2, ['Batt ' + batt, batt]])
 284
 285     def oregon_level2(self): # v2 and v3 level 2 decoder.
 286         nib = self.decoded_nibbles
 287         self.sensor_id = (nib[0][3] + nib[1][3] + nib[2][3] + nib[3][3]).upper()
 288         nl, sensor_type = sensor.get(self.sensor_id, [['Unknown'], 'Unknown'])
 289         names = ','.join(nl)
 290         # Allow user to try decoding an unknown sensor.
 291         if sensor_type == 'Unknown' and self.unknown != 'Unknown':
 292             sensor_type = self.unknown
 293         self.put(nib[0][0], nib[3][1], self.out_ann,
 294             [2, [names + ' - ' + sensor_type, names, nl[0]]])
 295         self.oregon_channel(4)
 296         self.oregon_battery(7)
 297         if sensor_type == 'Rain':
 298             self.oregon_put_l2_param(8, 4, 2, '', ' in/hr')     # Rain rate
 299             self.oregon_put_l2_param(12, 6, 3, 'Total ', ' in') # Rain total
 300             self.oregon_checksum(18)
 301         if sensor_type == 'Rain1':
 302             self.oregon_put_l2_param(8, 3, 1, '', ' mm/hr')     # Rain rate
 303             self.oregon_put_l2_param(11, 5, 1, 'Total ', ' mm') # Rain total
 304             self.oregon_checksum(18)
 305         if sensor_type == 'Temp':
 306             self.oregon_temp(8)
 307             self.oregon_checksum(12)
 308         if sensor_type == 'Temp_Hum_Baro':
 309             self.oregon_temp(8)
 310             self.oregon_put_l2_param(12, 2, 0, 'Hum ', '%') # Hum
 311             self.oregon_baro(15)                            # Baro
 312             self.oregon_checksum(19)
 313         if sensor_type == 'Temp_Hum_Baro1':
 314             self.oregon_temp(8)
 315             self.oregon_put_l2_param(12, 2, 0, 'Hum ', '%') # Hum
 316             self.oregon_baro(14)                            # Baro
 317         if sensor_type == 'Temp_Hum':
 318             self.oregon_temp(8)
 319             self.oregon_put_l2_param(12, 2, 0, 'Hum ', '%') # Hum
 320             self.oregon_checksum(15)
 321         if sensor_type == 'Temp_Hum1':
 322             self.oregon_temp(8)
 323             self.oregon_put_l2_param(12, 2, 0, 'Hum ', '%') # Hum
 324             self.oregon_checksum(14)
 325         if sensor_type == 'UV':
 326             self.oregon_put_l2_param(8, 2, 0, '', '') # UV
 327         if sensor_type == 'UV1':
 328             self.oregon_put_l2_param(11, 2, 0,'' ,'') # UV
 329         if sensor_type == 'Wind':
 330             self.oregon_wind_dir(8)
 331             self.oregon_put_l2_param(11, 3, 1, 'Gust ', ' m/s')  # Wind gust
 332             self.oregon_put_l2_param(14, 3, 1, 'Speed ', ' m/s') # Wind speed
 333             self.oregon_checksum(17)
 334
 335     def oregon_put_checksum(self, nibbles, checksum):
 336         nib = self.decoded_nibbles
 337         result = 'BAD'
 338         if (nibbles + 1) < len(nib):
 339             if (nib[nibbles + 1][3] != '' and nib[nibbles][3] != ''
 340                  and checksum != -1):
 341                 if self.ver != 'v1':
 342                     if checksum == (int(nib[nibbles + 1][3], 16) * 16 +
 343                                     int(nib[nibbles][3], 16)):
 344                         result = 'OK'
 345                 else:
 346                     if checksum == (int(nib[nibbles][3], 16) * 16 +
 347                                     int(nib[nibbles + 1][3], 16)):
 348                         result = 'OK'
 349             rx_check = (nib[nibbles + 1][3] + nib[nibbles][3]).upper()
 350             details = '%s Calc %s Rx %s ' % (result, hex(checksum)[2:].upper(),
 351                                              rx_check)
 352             self.put(nib[nibbles][0], nib[nibbles + 1][1],
 353                      self.out_ann, [2, ['Checksum ' + details, result]])
 354
 355     def oregon_checksum(self, nibbles):
 356         checksum = 0
 357         for i in range(nibbles):        # Add reversed nibbles.
 358             nibble = self.ookstring[i * 4 : i * 4 + 4]
 359             nibble = nibble[::-1]       # Reversed from right.
 360             if 'E' in nibble:           # Abort checksum if there are errors.
 361                 checksum = -1
 362                 break
 363             checksum += int(nibble, 2)
 364             if checksum > 255:
 365                 checksum -= 255         # Make it roll over at 255.
 366         chk_ver, comment = sensor_checksum.get(self.sensor_id,
 367                                                ['Unknown', 'Unknown'])
 368         if chk_ver != 'Unknown':
 369             self.ver = chk_ver
 370         if self.ver == 'v2.1':
 371             checksum -= 10              # Subtract 10 from v2 checksums.
 372         self.oregon_put_checksum(nibbles, checksum)
 373
 374     def oregon_checksum_v1(self):
 375         nib = self.decoded_nibbles
 376         checksum = 0
 377         for i in range(3):              # Add the first three bytes.
 378             if nib[2 * i][3] == '' or nib[2 * i + 1][3] == '': # Abort if blank.
 379                 checksum = -1
 380                 break
 381             checksum += ((int(nib[2 * i][3], 16) & 0xF) << 4 |
 382                          (int(nib[2 * i + 1][3], 16) & 0xF))
 383             if checksum > 255:
 384                 checksum -= 255         # Make it roll over at 255.
 385         self.oregon_put_checksum(6, checksum)
 386
 387     def decode(self, ss, es, data):
 388         self.decoded = data
 389         self.oregon()