Wammu/PhoneSearch.py

   1 # -*- coding: UTF-8 -*-
   2 # vim: expandtab sw=4 ts=4 sts=4:
   3 '''
   4 Wammu - Phone manager
   5 Searching for phone
   6 '''
   7 __author__ = 'Michal Čihař'
   8 __email__ = 'michal@cihar.com'
   9 __license__ = '''
  10 Copyright © 2003 - 2008 Michal Čihař
  11
  12 This program is free software; you can redistribute it and/or modify it
  13 under the terms of the GNU General Public License version 2 as published by
  14 the Free Software Foundation.
  15
  16 This program is distributed in the hope that it will be useful, but WITHOUT
  17 ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  18 FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  19 more details.
  20
  21 You should have received a copy of the GNU General Public License along with
  22 this program; if not, write to the Free Software Foundation, Inc.,
  23 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  24 '''
  25
  26 import wx
  27 import threading
  28 import sys
  29 import Wammu
  30 if Wammu.gammu_error == None:
  31     import gammu
  32 import Wammu.Data
  33 import Wammu.Events
  34 import wx.lib.layoutf
  35 from Wammu.Locales import StrConv
  36 import Wammu.Utils
  37
  38 try:
  39     import bluetooth
  40     import Wammu.BluezDiscovery
  41     BLUETOOTH = 'bluez'
  42 except ImportError:
  43     BLUETOOTH = None
  44
  45 class AllSearchThread(threading.Thread):
  46     '''
  47     Root thread for phone searching. It spawns other threads for testing each
  48     device.
  49     '''
  50     def __init__(self,
  51             lock = 'no',
  52             level = 'nothing',
  53             msgcallback = None,
  54             callback = None,
  55             win = None,
  56             noticecallback = None,
  57             limit = None):
  58         threading.Thread.__init__(self)
  59         self.lock = lock
  60         self.list = []
  61         self.win = win
  62         self.listlock = threading.Lock()
  63         self.level = level
  64         self.threads = []
  65         self.callback = callback
  66         self.msgcallback = msgcallback
  67         self.noticecallback = noticecallback
  68         self.limit = limit
  69
  70     def create_search_thread(self, device, connections, name):
  71         '''
  72         Creates single thread for searching phone on device using listed
  73         connections. Name is just text which will be shown to user.
  74         '''
  75         newthread = SearchThread(
  76                 device,
  77                 connections,
  78                 self.list,
  79                 self.listlock,
  80                 self.lock,
  81                 self.level)
  82         newthread.setName(name)
  83         if self.msgcallback != None:
  84             self.msgcallback(
  85                     _('Checking %s') %
  86                     StrConv(name)
  87                     )
  88         self.threads.append(newthread)
  89         newthread.start()
  90
  91     def search_bt_device(self, address, name):
  92         '''
  93         Searches single bluetooth device.
  94         '''
  95         connections = Wammu.Data.Conn_Bluetooth_All
  96         vendorguess = _('Could not guess vendor')
  97         # Use better connection list for some known manufacturers
  98         for vendor in Wammu.Data.MAC_Prefixes.keys():
  99             if address[:8].upper() in Wammu.Data.MAC_Prefixes[vendor]:
 100                 connections = Wammu.Data.Conn_Bluetooth[vendor]
 101                 vendorguess = _('Guessed as %s') % vendor
 102
 103         self.create_search_thread(
 104                 address,
 105                 connections,
 106                 '%s (%s) - %s - %s' % (
 107                     address,
 108                     name,
 109                     vendorguess,
 110                     str(connections)))
 111
 112     def check_device(self, curdev):
 113         '''
 114         Checks whether it makes sense to perform searching on this device and
 115         possibly warns user about misconfigurations.
 116         '''
 117         res = Wammu.Utils.CheckDeviceNode(curdev)
 118
 119         if res[0] == 0:
 120             return True
 121         if res[0] == -1:
 122             return False
 123         if res[1] != '' and self.msgcallback != None:
 124             self.msgcallback(res[1])
 125         if res[2] != '' and self.noticecallback != None:
 126             self.noticecallback(res[2], res[3])
 127         return False
 128
 129     def search_device(self, curdev, dev):
 130         '''
 131         Performs search on one real device.
 132         '''
 133         if len(curdev) > 0 and curdev[0] == '/':
 134             if not self.check_device(curdev):
 135                 return
 136
 137         self.create_search_thread(
 138                 curdev,
 139                 dev[0],
 140                 '%s - %s' % (curdev, str(dev[0])))
 141
 142     def listed_device_search(self):
 143         '''
 144         Initiates searching of devices defined in Wammu.Data.AllDevices.
 145         '''
 146         for dev in Wammu.Data.AllDevices:
 147             if not (self.limit == 'all' or self.limit in dev[3]):
 148                 continue
 149             if dev[1].find('%d') >= 0:
 150                 for i in range(*dev[2]):
 151                     curdev = dev[1] % i
 152                     self.search_device(curdev, dev)
 153             else:
 154                 self.search_device(dev[1], dev)
 155
 156     def bluetooth_device_search_bluez(self):
 157         '''
 158         Initiates searching for Bluetooth devices using PyBluez stack.
 159         '''
 160         # read devices list
 161         if self.msgcallback != None:
 162             self.msgcallback(_('Scanning for bluetooth devices using %s') %
 163                     'PyBluez')
 164
 165         try:
 166             discovery = Wammu.BluezDiscovery.Discovery(self)
 167             discovery.find_devices()
 168             discovery.process_inquiry()
 169             if len(discovery.names_found) == 0 and self.msgcallback != None:
 170                 self.msgcallback(_('No bluetooth device found'))
 171             if self.msgcallback != None:
 172                 self.msgcallback(_('Bluetooth device scan completed'))
 173         except bluetooth.BluetoothError, txt:
 174             if self.msgcallback != None:
 175                 self.msgcallback(
 176                         _('Could not access Bluetooth subsystem (%s)') %
 177                         StrConv(txt))
 178
 179     def bluetooth_device_search(self):
 180         '''
 181         Initiates searching for Bluetooth devices.
 182         '''
 183         if not self.limit in ['all', 'bluetooth']:
 184             return
 185         if BLUETOOTH == 'bluez':
 186             self.bluetooth_device_search_bluez()
 187         else:
 188             if self.msgcallback != None:
 189                 self.msgcallback(_('PyBluez not found, it is not possible to scan for bluetooth devices.'))
 190             if self.noticecallback != None:
 191                 self.noticecallback(
 192                         _('No bluetooth searching'),
 193                         _('PyBluez not found, it is not possible to scan for bluetooth devices.'))
 194
 195     def run(self):
 196         try:
 197             self.listed_device_search()
 198             self.bluetooth_device_search()
 199
 200             i = 0
 201             while len(self.threads) > 0:
 202                 if self.threads[i].isAlive():
 203                     i += 1
 204                 else:
 205                     if self.msgcallback != None:
 206                         self.msgcallback(_('Finished %s') %
 207                                 StrConv(self.threads[i].getName()))
 208                     del self.threads[i]
 209                 if i >= len(self.threads):
 210                     i = 0
 211             if self.msgcallback != None:
 212                 self.msgcallback(_('All finished, found %d phones') %
 213                         len(self.list))
 214             if self.callback != None:
 215                 self.callback(self.list)
 216         except:
 217             evt = Wammu.Events.ExceptionEvent(data = sys.exc_info())
 218             wx.PostEvent(self.win, evt)
 219
 220 class SearchThread(threading.Thread):
 221     def __init__(self,
 222             device,
 223             connections,
 224             lst,
 225             listlock,
 226             lock = 'no',
 227             level = 'nothing',
 228             win = None):
 229         threading.Thread.__init__(self)
 230         self.device = device
 231         self.connections = connections
 232         self.lock = lock
 233         self.win = win
 234         self.level = level
 235         self.list = lst
 236         self.listlock = listlock
 237
 238     def try_connection(self, connection):
 239         '''
 240         Performs test on single connection.
 241         '''
 242         gsm = gammu.StateMachine()
 243         gsm.SetConfig(0,
 244                 {'StartInfo': 'no',
 245                  'UseGlobalDebugFile': 1,
 246                  'DebugFile': '',
 247                  'SyncTime': 'no',
 248                  'Connection': connection,
 249                  'LockDevice': self.lock,
 250                  'DebugLevel': self.level,
 251                  'Device': self.device,
 252                  'Localize': None,
 253                  'Model': ''})
 254         try:
 255             if self.level == 'textall':
 256                 print 'Trying at %s using %s' % (self.device, connection)
 257             gsm.Init()
 258             self.listlock.acquire()
 259             self.list.append((
 260                 self.device,
 261                 connection,
 262                 gsm.GetModel(),
 263                 gsm.GetManufacturer()
 264                 ))
 265             self.listlock.release()
 266             if self.level != 'nothing':
 267                 print '!!Found model %s at %s using %s' % (
 268                         gsm.GetModel(),
 269                         self.device,
 270                         connection)
 271             return
 272         except gammu.GSMError:
 273             if self.level == 'textall':
 274                 print 'Failed at %s using %s' % (self.device, connection)
 275
 276     def run(self):
 277         '''
 278         Tests all listed connections.
 279         '''
 280         try:
 281             for conn in self.connections:
 282                 self.try_connection(conn)
 283         except:
 284             evt = Wammu.Events.ExceptionEvent(data = sys.exc_info())
 285             wx.PostEvent(self.win, evt)
 286
 287 class PhoneInfoThread(threading.Thread):
 288     def __init__(self, win, device, connection):
 289         threading.Thread.__init__(self)
 290         self.device = device
 291         self.connection = connection
 292         self.result = None
 293         self.win = win
 294
 295     def run(self):
 296         res = Wammu.Utils.CheckDeviceNode(self.device)
 297         if res[0] != 0:
 298             evt = Wammu.Events.DataEvent(
 299                     data = None,
 300                     error = (res[2], res[3]))
 301             wx.PostEvent(self.win, evt)
 302             return
 303         try:
 304             sm = gammu.StateMachine()
 305             sm.SetConfig(0,
 306                     {'StartInfo': 'no',
 307                      'UseGlobalDebugFile': 1,
 308                      'DebugFile': '',
 309                      'SyncTime': 'no',
 310                      'Connection': self.connection,
 311                      'LockDevice': 'no',
 312                      'DebugLevel': 'nothing',
 313                      'Device': self.device,
 314                      'Localize': None,
 315                      'Model': '',
 316                      })
 317             sm.Init()
 318             self.result = {
 319                     'Model': sm.GetModel(),
 320                     'Manufacturer': sm.GetManufacturer(),
 321                     }
 322             evt = Wammu.Events.DataEvent(data = self.result)
 323             wx.PostEvent(self.win, evt)
 324         except gammu.GSMError, val:
 325             info = val[0]
 326             evt = Wammu.Events.DataEvent(
 327                 data = None,
 328                 error = (_('Failed to connect to phone'),
 329                     Wammu.Utils.FormatError('', info)
 330                     ))
 331             wx.PostEvent(self.win, evt)