source: src/django_gheat/gheat/dataimport/netstumbler.py@ 11406

#!/usr/bin/env python
#
# Reading NS1 files - http://www.stumbler.net/ns1files.html
#
# Rick van der Zwet <info@rickvanderzwet.nl>
#
import datetime
from struct import unpack

from collections import defaultdict
import logging

logger = logging.getLogger(__name__)

def parse_ns1(fh):
  def get_int32(size=1):
    v = unpack('<' + 'i'*size,fh.read(4*size))
    return v[0] if size == 1 else v
  
  def get_uint32(size=1):
    v = unpack('<' + 'I'*size,fh.read(4*size))
    return v[0] if size == 1 else v
  
  def get_uint64(size=1):
    v = unpack('<' + 'Q'*size,fh.read(8*size))
    return v[0] if size == 1 else v
  
  def get_uint8(size=1):
    v = unpack('<' + 'B'*size,fh.read(1*size))
    return v[0] if size == 1 else v
  
  def get_filetime():
    _FILETIME_null_date = datetime.datetime(1601, 1, 1, 0, 0, 0)
    ns = unpack('<Q',fh.read(8))[0] * 10
    sec = ns / 10**8
    d = datetime.timedelta(seconds=sec)
    return _FILETIME_null_date + d
  
  def get_char(size):
    return fh.read(size)
  
  def get_double(size=1):
    v = unpack('<' + 'd'*size,fh.read(8*size))
    return v[0] if size == 1 else v
  
  def get_mac():
    return ':'.join(["%02X" % x for x in unpack('BBBBBB',fh.read(6))])

  data = {}
  
  data["dwSignature"] = get_char(4)
  data["dwFileVerunpack"] = get_uint32()
  ApCount = get_uint32()
  data["ApCount"] = ApCount
  
  data["aps"] = []
  for a in range(0,ApCount):
    ap = {}
    SSIDLength = get_uint8()
    ap["SSIDLength"] = SSIDLength
    ap["SSID"] = get_char(SSIDLength)
    ap["BSSID"] = get_mac()
    ap["MaxSignal"] = get_int32()
    ap["MinNoise"] = get_int32()
    ap["MaxSNR"] = get_int32()
    ap["Flags"] =  get_uint32()
    ap["BeaconInterval"] = get_uint32()
    ap["FirstSeen"] = get_filetime()
    ap["LastSeen"] = get_filetime()
    ap["BestLat"] = get_double()
    ap["BestLong"] = get_double()
    DataCount = get_uint32()
    ap["DataCount"] = DataCount
    ap["measurements"] = []
    for c in range(0,DataCount):
      ms = {}
      ms["Time"] = get_filetime()
      ms["Signal"] = get_int32()
      ms["Noice"] = get_int32()
      LocationSource = get_int32()
      ms["LocationSource"] = LocationSource
      if LocationSource == 1:
        ms["Latitude"] = get_double()
        ms["Longitude"] = get_double()
        ms["Altitude"] = get_double()
        ms["NumStats"] = get_uint32()
        ms["Speed"] = get_double()
        ms["Track"] = get_double()
        ms["MagVariation"] = get_double()
        ms["Hdop"] = get_double()
      ap["measurements"].append(ms)
    NameLength = get_uint8()
    ap["NameLength"] = NameLength
    ap["Name"] = get_char(NameLength)
    ap["Channels"] = get_uint64()
    ap["LastChannel"] = get_uint32()
    ap["IPAddress"] = get_uint32()
    ap["MinSignal"] = get_int32()
    ap["MaxSignal"] = get_int32()
    ap["DataRate"] = get_uint32()
    ap["IPSubnet"] = get_uint32()
    ap["IPMask"] = get_uint32()
    ap["ApFlags"] = get_uint32()
    IELength = get_uint32()
    ap["IELength"] = IELength
    ap["InformationElements"] = get_uint8(IELength)
    data["aps"].append(ap)
  return data

def process_ns1(fh, counters):
  data = parse_ns1(fh)


  # Temponary holders
  meting_pool = defaultdict(list)
  ap_pool = {}

  for ap in data['aps']:
    # XXX: How is encryption coded?
    encryption = False
    ap_pool[ap['BSSID']]= (ap['SSID'], encryption)
    for point in ap["measurements"]:
     counters['meting_total'] += 1
     if point['LocationSource'] == 0:
       logger.debug("No GPS Coordinates found for BSSID %s @ %s", 
         ap['BSSID'], point['Time'])
       counters['meting_ignored'] += 1
       continue
     # We store all values found, avg or max will be done later on
     key = (ap['BSSID'], point["Latitude"], point["Longitude"])

     # Known measurement error
     if point['Signal'] == -32767: continue

     # XXX: Signal need properly be a relation of signal_dbm and noice_dbm
     signaal= 100 + point['Signal']
     if signaal > 100 or signaal < 0:
       logger.warning("Signal %s is not valid entry for BSSID %s @ %s",
         point['Signal'], ap['BSSID'], point['Time'])
       continue
     meting_pool[key].append(signaal)

  return (counters, ap_pool, None, meting_pool)
if __name__ == '__main__':
  import sys
  import pprint
  pp = pprint.PrettyPrinter(indent=2)
  pp.pprint(parse_ns1(open(sys.argv[1],'r')))