src/qweechat/weechat/protocol.py

   1 #!/usr/bin/python
   2 # -*- coding: utf-8 -*-
   3 #
   4 # Copyright (C) 2011 Sebastien Helleu <flashcode@flashtux.org>
   5 #
   6 # This file is part of QWeeChat, a Qt remote GUI for WeeChat.
   7 #
   8 # QWeeChat is free software; you can redistribute it and/or modify
   9 # it under the terms of the GNU General Public License as published by
  10 # the Free Software Foundation; either version 3 of the License, or
  11 # (at your option) any later version.
  12 #
  13 # QWeeChat is distributed in the hope that it will be useful,
  14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16 # GNU General Public License for more details.
  17 #
  18 # You should have received a copy of the GNU General Public License
  19 # along with QWeeChat.  If not, see <http://www.gnu.org/licenses/>.
  20 #
  21
  22 #
  23 # Decode binary messages received from WeeChat/relay.
  24 #
  25 # For info about protocol and format of messages, please read document "WeeChat Relay Protocol",
  26 # available at:  http://www.weechat.org/doc/
  27 #
  28 # History:
  29 #
  30 # 2011-11-23, Sebastien Helleu <flashcode@flashtux.org>:
  31 #     start dev
  32 #
  33
  34 import struct, zlib
  35
  36 class WeechatObject:
  37     def __init__(self, objtype, value):
  38         self.objtype = objtype;
  39         self.value = value
  40
  41     def _str_value(self, v):
  42         if type(v) is str and not v is None:
  43             return '\'%s\'' % v
  44         return str(v)
  45
  46     def _str_value_hdata(self):
  47         lines = ['',
  48                  '  keys: %s' % str(self.value['keys']),
  49                  '  path: %s' % str(self.value['path'])]
  50         for i, item in enumerate(self.value['items']):
  51             lines.append('  item %d:' % (i + 1))
  52             lines.append('\n'.join(['    %s: %s' % (key, self._str_value(value)) for key, value in sorted(item.items())]))
  53         return '\n'.join(lines)
  54
  55     def _str_value_infolist(self):
  56         lines = ['', '  name: %s' % self.value['name']]
  57         for i, item in enumerate(self.value['items']):
  58             lines.append('  item %d:' % (i + 1))
  59             lines.append('\n'.join(['    %s: %s' % (key, self._str_value(value)) for key, value in sorted(item.items())]))
  60         return '\n'.join(lines)
  61
  62     def _str_value_other(self):
  63         return self._str_value(self.value)
  64
  65     def __str__(self):
  66         self._obj_cb = {'hda': self._str_value_hdata,
  67                         'inl': self._str_value_infolist,
  68                         }
  69         return '%s: %s' % (self.objtype, self._obj_cb.get(self.objtype, self._str_value_other)())
  70
  71
  72 class WeechatObjects(list):
  73     def __str__(self):
  74         return '\n'.join([str(obj) for obj in self])
  75
  76
  77 class WeechatMessage:
  78     def __init__(self, size, size_uncompressed, compression, uncompressed, msgid, objects):
  79         self.size = size
  80         self.size_uncompressed = size_uncompressed
  81         self.compression = compression
  82         self.uncompressed = uncompressed
  83         self.msgid = msgid
  84         self.objects = objects
  85
  86     def __str__(self):
  87         if self.compression != 0:
  88             return 'size: %d/%d (%d%%), id=\'%s\', objects:\n%s' % (
  89                 self.size, self.size_uncompressed,
  90                 100 - ((self.size * 100) // self.size_uncompressed),
  91                 self.msgid, self.objects)
  92         else:
  93             return 'size: %d, id=\'%s\', objects:\n%s' % (self.size, self.msgid, self.objects)
  94
  95
  96 class Protocol:
  97     """Decode binary message received from WeeChat/relay."""
  98
  99     def __init__(self):
 100         self._obj_cb = {'chr': self._obj_char,
 101                         'int': self._obj_int,
 102                         'lon': self._obj_long,
 103                         'str': self._obj_str,
 104                         'buf': self._obj_buffer,
 105                         'ptr': self._obj_ptr,
 106                         'tim': self._obj_time,
 107                         'htb': self._obj_hashtable,
 108                         'hda': self._obj_hdata,
 109                         'inf': self._obj_info,
 110                         'inl': self._obj_infolist,
 111                         }
 112
 113     def _obj_type(self):
 114         """Read type in data (3 chars)."""
 115         if len(self.data) < 3:
 116             self.data = ''
 117             return ''
 118         objtype = str(self.data[0:3])
 119         self.data = self.data[3:]
 120         return objtype
 121
 122     def _obj_len_data(self, length_size):
 123         """Read length (1 or 4 bytes), then value with this length."""
 124         if len(self.data) < length_size:
 125             self.data = ''
 126             return None
 127         if length_size == 1:
 128             length = struct.unpack('B', self.data[0:1])[0]
 129             self.data = self.data[1:]
 130         else:
 131             length = self._obj_int()
 132         if length < 0:
 133             return None
 134         if length > 0:
 135             value = self.data[0:length]
 136             self.data = self.data[length:]
 137         else:
 138             value = ''
 139         return value
 140
 141     def _obj_char(self):
 142         """Read a char in data."""
 143         if len(self.data) < 1:
 144             return 0
 145         value = struct.unpack('b', self.data[0:1])[0]
 146         self.data = self.data[1:]
 147         return value
 148
 149     def _obj_int(self):
 150         """Read an integer in data (4 bytes)."""
 151         if len(self.data) < 4:
 152             self.data = ''
 153             return 0
 154         value = struct.unpack('>i', self.data[0:4])[0]
 155         self.data = self.data[4:]
 156         return value
 157
 158     def _obj_long(self):
 159         """Read a long integer in data (length on 1 byte + value as string)."""
 160         value = self._obj_len_data(1)
 161         if value is None:
 162             return None
 163         return int(str(value))
 164
 165     def _obj_str(self):
 166         """Read a string in data (length on 4 bytes + content)."""
 167         value = self._obj_len_data(4)
 168         if value is None:
 169             return None
 170         return str(value)
 171
 172     def _obj_buffer(self):
 173         """Read a buffer in data (length on 4 bytes + data)."""
 174         return self._obj_len_data(4)
 175
 176     def _obj_ptr(self):
 177         """Read a pointer in data (length on 1 byte + value as string)."""
 178         value = self._obj_len_data(1)
 179         if value is None:
 180             return None
 181         return '0x%s' % str(value)
 182
 183     def _obj_time(self):
 184         """Read a time in data (length on 1 byte + value as string)."""
 185         value = self._obj_len_data(1)
 186         if value is None:
 187             return None
 188         return str(value)
 189
 190     def _obj_hashtable(self):
 191         """Read a hashtable in data (type for keys + type for values + count + items)."""
 192         type_keys = self._obj_type()
 193         type_values = self._obj_type()
 194         count = self._obj_int()
 195         hashtable = {}
 196         for i in range(0, count):
 197             key = self._obj_cb[type_keys]()
 198             value = self._obj_cb[type_values]()
 199             hashtable[key] = value
 200         return hashtable
 201
 202     def _obj_hdata(self):
 203         """Read a hdata in data."""
 204         path = self._obj_str()
 205         keys = self._obj_str()
 206         count = self._obj_int()
 207         list_path = path.split('/')
 208         list_keys = keys.split(',')
 209         keys_types = []
 210         dict_keys = {}
 211         for key in list_keys:
 212             items = key.split(':')
 213             keys_types.append(items)
 214             dict_keys[items[0]] = items[1]
 215         items = []
 216         for i in range(0, count):
 217             item = {}
 218             pointers = []
 219             for p in range(0, len(list_path)):
 220                 pointers.append(self._obj_ptr())
 221             for key, objtype in keys_types:
 222                 item[key] = self._obj_cb[objtype]()
 223             item['__path'] = pointers
 224             items.append(item)
 225         return {'path': list_path,
 226                 'keys': dict_keys,
 227                 'count': count,
 228                 'items': items,
 229                 }
 230
 231     def _obj_info(self):
 232         """Read an info in data."""
 233         name = self._obj_str()
 234         value = self._obj_str()
 235         return (name, value)
 236
 237     def _obj_infolist(self):
 238         """Read an infolist in data."""
 239         name = self._obj_str()
 240         count_items = self._obj_int()
 241         items = []
 242         for i in range(0, count_items):
 243             count_vars = self._obj_int()
 244             variables = {}
 245             for v in range(0, count_vars):
 246                 var_name = self._obj_str()
 247                 var_type = self._obj_type()
 248                 var_value = self._obj_cb[var_type]()
 249                 variables[var_name] = var_value
 250             items.append(variables)
 251         return {'name': name, 'items': items}
 252
 253     def decode(self, data):
 254         """Decode binary data and return list of objects."""
 255         self.data = data
 256         size = len(self.data)
 257         size_uncompressed = size
 258         uncompressed = None
 259         # uncompress data (if it is compressed)
 260         compression = struct.unpack('b', self.data[4:5])[0]
 261         if compression:
 262             uncompressed = zlib.decompress(self.data[5:])
 263             size_uncompressed = len(uncompressed) + 5
 264             uncompressed = '%s%s%s' % (struct.pack('>i', size_uncompressed), struct.pack('b', 0), uncompressed)
 265             self.data = uncompressed
 266         # skip length and compression flag
 267         self.data = self.data[5:]
 268         # read id
 269         msgid = self._obj_str()
 270         if msgid is None:
 271             msgid = ''
 272         # read objects
 273         objects = WeechatObjects()
 274         while len(self.data) > 0:
 275             objtype = self._obj_type()
 276             value = self._obj_cb[objtype]()
 277             objects.append(WeechatObject(objtype, value))
 278         return WeechatMessage(size, size_uncompressed, compression, uncompressed, msgid, objects)
 279
 280
 281 def hex_and_ascii(data, bytes_per_line=10):
 282     """Convert a QByteArray to hex + ascii output."""
 283     num_lines = ((len(data) - 1) // bytes_per_line) + 1
 284     if num_lines == 0:
 285         return ''
 286     lines = []
 287     for i in range(0, num_lines):
 288         str_hex = []
 289         str_ascii = []
 290         for char in data[i*bytes_per_line:(i*bytes_per_line)+bytes_per_line]:
 291             byte = struct.unpack('B', char)[0]
 292             str_hex.append('%02X' % int(byte))
 293             if byte >= 32 and byte <= 127:
 294                 str_ascii.append(char)
 295             else:
 296                 str_ascii.append('.')
 297         fmt = '%%-%ds %%s' % ((bytes_per_line * 3) - 1)
 298         lines.append(fmt % (' '.join(str_hex), ''.join(str_ascii)))
 299     return '\n'.join(lines)