Changeset 9557

Timestamp:

Aug 25, 2011, 10:24:36 AM (13 years ago)

Author:

rick

Message:

Re-did importing to make it roughly 100x times faster, by using proper database imports.

File:

• src/django_gheat/gheat/management/commands/import_droidstumbler.py (modified) (4 diffs)

src/django_gheat/gheat/management/commands/import_droidstumbler.py

@@ -13 +13 @@
 from django.db import connection, transaction
 from django.db.utils import IntegrityError
+from _mysql_exceptions import OperationalError
 from optparse import OptionParser, make_option
 from gheat.models import *
@@ -22 +23 @@
 import sys
 
-def user_feedback(output, count, dot_step, summary_report):
-  if (count % summary_report) == 0:
-    output.write(str(count))
-  elif (count % dot_step) == 0:
-    output.write(".")
-  output.flush()
-  return count + 1
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+def bulk_sql(sql_table, sql_values):
+  if len(sql_values) == 0:
+    raise ValueError, "No data to import"
+
+  cursor = connection.cursor()
+  try:
+    sql = "INSERT INTO %s VALUES %s" % (sql_table, ','.join(sql_values))
+    count = cursor.execute(sql)
+    transaction.commit_unless_managed()
+  except OperationalError, e:
+    logger.error("%s - %s ", sql_table, sql_values[0])
+    raise
+  except IntegrityError, e:
+    logger.error("Unable to import - %s" %  e)
+  return count
 
 
-def import_droidstumbler(location, meetrondje, gebruiker, kaart, email, datum,show_progres=False,output=sys.stdout,bulk_import=True):
+def import_droidstumbler(filename, meetrondje):
   """ Import all points, return tuple with summary"""
-  g, created = Gebruiker.objects.get_or_create(naam=gebruiker , email=email)
-  a, created = Apparatuur.objects.get_or_create(kaart=kaart)
-  mr, created = MeetRondje.objects.get_or_create(datum=datum , naam=meetrondje , gebruiker=g , apparatuur=a)
-  if not created:
-    logging.error("Meetrondje '%s' already imported" % mr)
-    sys.exit(1)
 
-  meting_count = 0
-  new_ap_count = 0
-  ap_cache = {}
+  # Open file for reading
+  if filename.endswith('.gz'):
+    fh = gzip.open(filename,'rb')
+  else:
+    fh = open(filename,'rb')
+  csvfile = csv.reader(fh, delimiter=',')
+
+  #Various statistics
+  counters = {'meting_added' : 0, 'meting_total' : 0, 'ap_added' : 0, 'ap_total' : 0}
+
+  # Temponary holders
   meting_pool = {}
-
-  if show_progres: output.write('#INFO: Pre-Processing file: ')
-  if location.endswith('.gz'):
-    fh = gzip.open(location,'rb')
-  else:
-    fh = open(location,'rb')
-  csvfile = csv.reader(fh, delimiter=',')
-  count = 0
+  ap_pool = {}
+  # Process file, preparing new access points and measurements
   for row in csvfile:
-    if show_progres: count = user_feedback(output, count, 100, 1000)
     try:
       epoch, msg_type, lat, lon, accuracy, ssid, bssid, level, frequency, capabilities = row
     except ValueError:
-      logging.error("Unable to parse line:%i '%s'" % (csvfile.line_num, row))
+      # Known error, please ignore
+      if row[1] == 'gps' and len(row) == 12: continue
+      logger.error("Unable to parse line:%i '%s'" % (csvfile.line_num, row))
       continue
     if msg_type == "data" and lat and lon:
-      if not ap_cache.has_key(bssid):
-        ap_cache[bssid], created = Accespoint.objects.get_or_create(mac=bssid, ssid=ssid, encryptie=capabilities)
-        if created: new_ap_count += 1
+      counters['meting_total'] += 1
+      if not ap_pool.has_key(bssid):
+        encryption = 'WPA' in capabilities or 'WEP' in capabilities
+        ap_pool[bssid] = (ssid, encryption)
 
       # We store the best value found
-      key = (ap_cache[bssid], lat, lon)
+      key = (bssid, lat, lon)
       signaal=(100 + int(level))
       if meting_pool.has_key(key):
@@ -71 +81 @@
       else:
         meting_pool[key] = signaal
-  if show_progres: output.write("%s\n" % count)
 
 
-  if show_progres: output.write('#INFO: Importing the data into the database: ')
-  meting_count = 0
-  cursor = connection.cursor()
+  # Determine which entries we need to add
+  counters['ap_total'] = len(ap_pool)
+  bssid_list_present = Accespoint.objects.filter(mac__in=ap_pool.keys()).values_list('mac', flat=True)
+  bssid_list_insert = set(ap_pool.keys()) - set(bssid_list_present)
+
+  # Create a bulk import list and import
+  if bssid_list_insert:
+    sql_values = []
+    for bssid in bssid_list_insert:
+      ssid, encryption = ap_pool[bssid]
+      # Special trick in SSID ts avoid escaping in later stage
+      value_str = str((bssid,ssid.replace('%','%%'),encryption))
+      sql_values.append(value_str)
+  counters['ap_added'] = bulk_sql('gheat_accespoint (`mac`, `ssid`, `encryptie`)',sql_values)
+
+  # Build mapping for meting import
+  mac2id = {}
+  for mac,id in Accespoint.objects.filter(mac__in=ap_pool.keys()).values_list('mac','id'):
+    mac2id[mac] = int(id)
+
   sql_values = []
-  for (ap,lat,lon),signal in meting_pool.iteritems():
-    try:
-      if bulk_import:
-        value_str = str((int(mr.id),int(ap.id),float(lat),float(lon),int(signaal)))
-        sql_values.append(value_str)
-      else:
-        m = Meting.objects.create(meetrondje=mr, accespoint=ap, latitude=lat, longitude=lon, signaal=signaal)
-    except IntegrityError, e:
-      logging.error("Unable to import - %s" %  e)
-      continue
-
-    # Give some feedback to the user
-    if show_progres: meting_count = user_feedback(output, meting_count, 100, 1000)
-
-  # Make sure to include closing newline
-  if show_progres: output.write("%s\n" % meting_count)
+  for (bssid,lat,lon),signal in meting_pool.iteritems():
+    value_str = str((int(meetrondje.id),mac2id[bssid],float(lat),float(lon),int(signaal)))
+    sql_values.append(value_str)
 
   # Bulk Import data if possible
-  if bulk_import:
-    if len(sql_values) == 0:
-      logging.warn("No data to import")
-    else:
-      try:
-        sql = "INSERT INTO gheat_meting (`meetrondje_id`, `accespoint_id`, `lat`, `lng`, `signaal`) VALUES %s" % ','.join(sql_values)
-        cursor.execute(sql)
-        transaction.commit_unless_managed()
-      except IntegrityError, e:
-        logging.error("Unable to import - %s" %  e)
-        pass
-
-  return (len(ap_cache), new_ap_count, meting_count, len(meting_pool) - meting_count)
+  if sql_values:
+    counters['meting_added'] = bulk_sql('gheat_meting (`meetrondje_id`, `accespoint_id`, `lat`, `lng`, `signaal`)',sql_values)
+  return counters
 
 
@@ -146 +149 @@
          datum = datetime.datetime.strptime(datum,'%Y-%m-%d-%H%M%S')
       except ValueError:
-        raise CommandError("Invalid date '%s'\n" % options['datum'])
+        raise CommandError("Invalid date '%s'" % options['datum'])
 
-      self.stdout.write('#INFO: Meetrondje: %s @ %s\n' % (meetrondje, datum))
-      self.stdout.write("#INFO: Going to import '%s' for gebruiker '%s <%s>'\n" % (os.path.basename(csv_file), options['gebruiker'], options['email']))
-      (ap_count, new_ap_count, meting_count, meting_error_count) = import_droidstumbler(csv_file,meetrondje,options['kaart'],options['gebruiker'],options['email'], datum, True)
-      self.stdout.write("#INFO: Import summary accespoints: added:%s processed:%s\n" % (new_ap_count, ap_count))
-      self.stdout.write("#INFO: Import summary metingen: added:%s error:%s\n" % (meting_count, meting_error_count))
+      logger.info("Going to import '%s'" % (os.path.basename(csv_file)))
+      # Create meetrondje object
+      g, created = Gebruiker.objects.get_or_create(naam=options['gebruiker'] , email=options['email'])
+      a, created = Apparatuur.objects.get_or_create(kaart=options['kaart'])
+      mr, created = MeetRondje.objects.get_or_create(datum=datum , naam=meetrondje , gebruiker=g , apparatuur=a)
+      logger.info('Meetrondje: %s @ %s' % (meetrondje, datum))
+      if not created:
+        logger.error("Meetrondje '%s' already imported" % mr)
+        sys.exit(1)
+      counters = import_droidstumbler(csv_file,mr)
+      logger.info("Summary accespoints: added:%(ap_added)-6s processed:%(ap_total)-6s" % counters)
+      logger.info("Summary metingen   : added:%(meting_added)-6s processed:%(meting_total)-6s" % counters)