factory_test_tools: use flimflam to get connection info directly.

gft_hwcomp.py

 #!/usr/bin/python
 #
 # Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 import glob
+import imp
 import os
 import pprint
 import re
 import sys
 import threading
 
 import flashrom_util
 import gft_common
 import gft_fwhash
 import vblock
 
 from gft_common import DebugMsg, VerboseMsg, WarningMsg, ErrorMsg, ErrorDie
 
 
-def Memorize(f):
-  return gft_common.ThreadSafe(gft_common.Memorize(f))
-
-
 class HardwareComponents(object):
   """ Hardware Components Scanner """
 
   # Function names in this class are used for reflection, so please don't change
   # the function names even if they are not comliant to coding style guide.
 
   version = 6
 
   # We divide all component IDs (cids) into 5 categories:
   #  - enumerable: able to get the results by running specific commands;
@@ skipping 50 matching lines @@
     ]
 
   # _not_test_cids and _to_be_tested_cids will be re-created for each match.
   _not_test_cids = []
   _to_be_tested_cids = []
 
   # TODO(hungte) unify the 'not available' style messages
   _not_present = ''
   _no_match = 'No match'
 
+  # Type id for connection management (compatible to flimflam)
+  _type_3g = 'cellular'
+  _type_ethernet = 'ethernet'
+  _type_wireless = 'wifi'
+
+  _flimflam_dir = '/usr/local/lib/flimflam/test'
+
   def __init__(self, verbose=False):
     self._initialized = False
     self._verbose = verbose
     self._pp = pprint.PrettyPrinter()
 
     # cache for firmware images
     self._flashrom = flashrom_util.flashrom_util(
         verbose_msg=VerboseMsg,
         exception_type=gft_common.GFTError,
         system_output=gft_common.SystemOutput)
     self._temp_files = []
 
     # variables for matching
     self._enumerable_system = None
     self._file_base = None
     self._system = None
     self._failures = None
 
   def __del__(self):
     for temp_file in self._temp_files:
       try:
         # DebugMsg('gft_hwcomp: delete temp file %s' % temp_file)
         os.remove(temp_file)
       except:
         pass
 
   # --------------------------------------------------------------------
   # System Probing Utilities
 
+  def Memorize(f):
+    """ Decorator for thread-safe memorization """
+    return gft_common.ThreadSafe(gft_common.Memorize(f))
+
   @Memorize
   def load_module(self, name):
     grep_cmd = ('lsmod 2>/dev/null | grep -q %s' % name)
     loaded = (os.system(grep_cmd) == 0)
     if not loaded:
       if os.system('modprobe %s >/dev/null 2>&1' % name) != 0:
         ErrorMsg("Cannot load module: %s" % name)
     return loaded
 
   @Memorize
@@ skipping 24 matching lines @@
     usb_devs = glob.glob('/sys/bus/usb/devices/*:*')
     for dev in usb_devs:
       path = os.path.join(os.path.realpath(dev), '..')
       device_list += ['%s:%s' %
                       (gft_common.ReadOneLine(os.path.join(path, 'idVendor')),
                        gft_common.ReadOneLine(os.path.join(path, 'idProduct')))]
     DebugMsg('Legacy device list: ' + ', '.join(device_list))
     return device_list
 
   @Memorize
+  def import_flimflam_module(self):
+    """ Tries to load flimflam module from current system """
+    if not os.path.exists(self._flimflam_dir):
+      DebugMsg('no flimflam installed in %s' % self._flimflam_dir)
+      return None
+    try:
+      return imp.load_module('flimflam',
+                             *imp.find_module('flimflam', [self._flimflam_dir]))
+    except ImportError:
+      ErrorMsg('Failed to import flimflam.')
+    except:
+      ErrorMsg('Failed to load flimflam.')
+    return None
+
+  @Memorize
+  def load_flimflam(self):
+    """Gets information provided by flimflam (connection manager)
+
+    Returns
+    (None, None) if failed to load module, otherwise
+    (connection_path, connection_info) where
+     connection_path is a dict in {type: device_path},
+     connection_info is a dict of {type: {attribute: value}}.
+    """
+    flimflam = self.import_flimflam_module()
+    if not flimflam:
+      return (None, None)
+    path = {}
+    info = {}
+    info_attribute_names = {
+        self._type_3g: ['Carrier', 'FirmwareRevision', 'HardwareRevision',
+                        'ModelID', 'Manufacturer'],
+    }
+    devices = flimflam.FlimFlam().GetObjectList('Device')
+    unpack = flimflam.convert_dbus_value
+    # TODO(hungte) support multiple devices existence in future
+    for device in devices:
+      # populate the 'path' collection
+      prop = device.GetProperties()
+      prop_type = unpack(prop['Type'])
+      path[prop_type] = '/sys/class/net/%s/device' % unpack(prop['Interface'])
+      if prop_type not in info_attribute_names:
+        continue
+      # populate the 'info' collection
+      info[prop_type] = dict((
+          (key, unpack(prop['Cellular.%s' % key]))
+          for key in info_attribute_names[prop_type]
+          if ('Cellular.%s' % key) in prop))
+    return (path, info)
+
+  @Memorize
   def _get_all_connection_info(self):
     """ Probes available connectivity and device information """
     connection_info = {
-        'wireless': '/sys/class/net/wlan0/device',
-        'ethernet': '/sys/class/net/eth0/device',
-        '3g': '/sys/class/tty/ttyUSB0/device',
+        self._type_wireless: '/sys/class/net/wlan0/device',
+        self._type_ethernet: '/sys/class/net/eth0/device',
+        # 3g(cellular) may also be /sys/class/net/usb0
+        self._type_3g: '/sys/class/tty/ttyUSB0/device',
     }
-    flimflam_scripts_base = '/usr/local/lib/flimflam/test'
-    if os.path.exists(flimflam_scripts_base):
-      # TODO(hungte) use flimflam to update the device list.
-      # Currently flimflam seem to be not very stable, so let's ignore it.
-      list_modems = os.path.join(flimflam_scripts_base, 'mm-list-modems')
-      part_id = gft_common.SystemOutput(list_modems, ignore_status=True).strip()
-      if part_id:
-        DebugMsg('mm-list-modems reported: %s' % part_id)
+    (path, _) = self.load_flimflam()
+    if path is not None:
+      # trust flimflam instead.
+      for k in connection_info:
+        connection_info[k] = (path[k] if k in path else '')
     return connection_info
 
   def _get_sysfs_device_info(self, path, primary, optional=[]):
     """Gets the device information of a sysfs node.
 
     Args
       path: the sysfs device path.
       primary: mandatory list of elements to read.
       optional: optional list of elements to read.
 
@@ skipping 44 matching lines @@
     return info_string
 
   def get_sysfs_device_id(self, path):
     """Gets a sysfs device identifier. (Currently supporting USB/PCI)
     Args
       path: a path to sysfs device (ex, /sys/class/net/wlan0/device)
 
     Returns
       An identifier string, or self._not_present if not available.
     """
+    if not path:
+      return self._not_present
     path = os.path.realpath(path)
     if not os.path.isdir(path):
       return self._not_present
 
     info = (self._get_pci_device_info(path) or
             self._get_usb_device_info(path))
     return info or self._not_present
 
   # --------------------------------------------------------------------
   # Firmware Processing Utilities
@@ skipping 91 matching lines @@
     Returns a hash of Read Only (BIOS) firmware parts,
     to confirm we have proper keys / boot code / recovery image installed.
     """
 
     image_file = self.load_main_firmware()
     if not image_file:
       ErrorDie('get_hash_ro_firmware: cannot read firmware')
     return gft_fwhash.GetBIOSReadOnlyHash(file_source=image_file)
 
   def get_part_id_3g(self):
-    device_path = self._get_all_connection_info()['3g']
+    device_path = self._get_all_connection_info()[self._type_3g]
     return self.get_sysfs_device_id(device_path) or self._not_present
 
   def get_part_id_audio_codec(self):
     cmd = 'grep -R Codec: /proc/asound/* | head -n 1 | sed s/.\*Codec://'
     part_id = gft_common.SystemOutput(
         cmd, progress_messsage='Searching Audio Codecs: ',
         show_progress=self._verbose).strip()
     return part_id
 
   def get_part_id_bluetooth(self):
@@ skipping 59 matching lines @@
         show_progress=self._verbose,
         ignore_status=True).splitlines()
     for line in res:
       match = re.search(r'Found (.*) at', line)
       if match:
         parts.append(match.group(1))
     part_id = ', '.join(parts)
     return part_id
 
   def get_part_id_ethernet(self):
-    device_path = self._get_all_connection_info()['ethernet']
+    device_path = self._get_all_connection_info()[self._type_ethernet]
     return self.get_sysfs_device_id(device_path) or self._not_present
 
   def get_part_id_flash_chip(self):
     (chip_id, _) = self._load_firmware('main')
     return chip_id
 
   def get_part_id_hwqual(self):
     part_id = gft_common.SystemOutput('crossystem hwid').strip()
     return (part_id if part_id else self._not_present)
 
@@ skipping 35 matching lines @@
     usb_host_info.sort(reverse=True)
     return ' '.join(usb_host_info)
 
   def get_part_id_vga(self):
     return self.get_sysfs_device_id('/sys/class/graphics/fb0/device')
 
   def get_part_id_webcam(self):
     return self.get_sysfs_device_id('/sys/class/video4linux/video0/device')
 
   def get_part_id_wireless(self):
-    device_path = self._get_all_connection_info()['wireless']
+    device_path = self._get_all_connection_info()[self._type_wireless]
     return self.get_sysfs_device_id(device_path) or self._not_present
 
   def get_vendor_id_touchpad_synaptics(self):
     part_id = ''
     detect_program = '/opt/Synaptics/bin/syndetect'
     model_string_str = 'Model String'
     firmware_id_str = 'Firmware ID'
     if os.path.exists(detect_program):
       data = gft_common.SystemOutput(
           detect_program,
@@ skipping 26 matching lines @@
     part_id = ''
     for method in method_list:
       part_id = getattr(self, 'get_vendor_id_touchpad_%s' % method)()
       DebugMsg('get_vendor_id_touchpad: %s: %s' %
                (method, part_id or '<failed>'))
       if part_id:
         break
     return part_id
 
   def get_version_3g_firmware(self):
-    version = 'Unknown'
-    # TODO(hungte) use mm.py directly to prevent the need of shell scripting
-    modem_status = gft_common.SystemOutput('modem status',
-                                           ignore_status=True).splitlines()
-    if not modem_status:
-      return self._not_present
-
-    # status format:
-    # Manufacturer: $VENDOR
-    # Modem: $MODEM
-    # Version: $VERSION_MAY_BE_MULTILINE
-    def ParseInfo(name):
-      data = [line for line in modem_status if line.find('%s: ' % name) >= 0]
-      assert len(data) < 2
-      return data[0].split(': ', 1)[1].strip()
-
-    version = ParseInfo('Version')
-    vendor = ParseInfo('Manufacturer')
-    modem = ParseInfo('Modem')
-
-    if vendor == 'Samsung' and modem == 'GT-Y3300X':
-      # The real version is in "Version:" +2 lines
-      version = ''
-      for i, line in enumerate(modem_status):
-        if 'Version: ' in line:
-          version = modem_status[i + 2].strip()
-          break
-    return version or 'Unknown'
+    (_, attributes) = self.load_flimflam()

[Jason Glasgow, 2011/04/15 22:39:44]

Looks okay to me.  I would have used mm.py and gone directly to the modem
instead of going through flimflam, but I guess this works.

+    if attributes and (self._type_3g in attributes):
+      # A list of possible version info combination, since the supported fields
+      # may differ for partners.
+      version_formats = [
+          ['Carrier', 'FirmwareRevision'],
+          ['FirmwareRevision'],
+          ['HardwareRevision']]
+      info = attributes[self._type_3g]
+      for version_format in version_formats:
+        if not set(version_format).issubset(set(info)):
+          continue
+        # compact all fields into single line with limited space
+        version_string = ' '.join([info[key] for key in version_format])
+        return re.sub('\s+', ' ', version_string)
+    return self._not_present
 
   def get_version_rw_firmware(self):
     """
     Returns the version of Read-Write (writable) firmware from VBLOCK sections.
 
     If A/B has different version, that means this system needs a reboot +
     firmwar update so return value is a "error report" in the form "A=x, B=y".
     """
 
     versions = [None, None]
@@ skipping 237 matching lines @@
   print 'Starting to match system...'
   for arg in compdb_list:
     (matched, failures) = components.match_current_system(arg)
     print 'Probed (%s):' % arg
     print components.pformat(matched)
     print 'Failures (%s):' % arg
     print components.pformat(failures)
 
 if __name__ == '__main__':
   _main(sys.argv[0], sys.argv[1:])