dalvik_trace.py

import os
import datetime 
import pdb
from struct import unpack
from collections import OrderedDict
from cStringIO import StringIO

path = "/data/project/py/github/android/demo/testok.stacks" 
data = open(path, "r").read()

class DalvikTrace:
    def __init__(self, data):
        self.data = data 
        threads_offset = data.find("*threads") + len("*threads\n")
        threads_end_offset =  data.find("*methods") - 1
        threads_data = data[threads_offset:threads_end_offset]

        methods_offset = data.find("*methods") + len("*methods\n")
        methods_end_offset = data.find("*end") - 1
        methods_data = data[methods_offset:methods_end_offset]
        self.threads = self.build_threads(threads_data)
        self.methods = self.build_methods(methods_data)
        
        stacks_offset = data.find("*end") + len("*end\n")
        self.stacks = data[stacks_offset:] 

    def build_pathname(self, className, pathname):
        index = className.rfind("/")
        if index > 0 and index < len(className) - 1 and pathname.endswith(".java"):
            pathname = className[:index+1] + pathname
            return pathname

    def build_threads(self, data):
        threads = {}
        for line in  data.split("\n"):
            thread = line.split("\t")
            threads[int(thread[0])] = thread[1]
        return threads
        
    def build_methods(self, data):
        methods = {}
        for line in data.split("\n"): 
            tokens = line.split("\t")
            methodid = int(tokens[0], 16) 
            methods[methodid] = {} 
            mdict =methods[methodid]
            mdict['className'] = tokens[1]
            if len(tokens) == 6:
                mdict['methodName'] = tokens[2]
                mdict['signature'] = tokens[3]
                mdict['pathname'] = self.build_pathname(tokens[1], tokens[4])
                mdict['lineNumber'] = tokens[5]
            elif len(tokens) > 2:
                if tokens[3].startswith("("):
                    mdict['methodName'] = tokens[2]
                    mdict['signature'] = tokens[3]
                else:
                    mdict['pathname'] = tokens[2]
                    mdict['lineNumber'] = token(3)
        return methods            

    def build_stacks(self, data):
        stacks = OrderedDict()

        #ignore read header
        buffer = StringIO()
        buffer.write(data)  
        buffer.seek(0)
        magic_number = buffer.read(4)
        version = unpack("H", buffer.read(2))[0] 
        header_length = unpack("H", buffer.read(2))[0] 
        timestamp = unpack("Q", buffer.read(8))[0]
        time = datetime.date.fromtimestamp(timestamp/1000000)
        #ignore 16 chars
        buffer.read(16)
        while True: 
            try:
                threadid = unpack("H", buffer.read(2))[0]
                method = unpack("I", buffer.read(4))[0]
                method_action = method & 0x03 
                methodid = method & ~0x03 
                time = unpack("I", buffer.read(4))[0]
                if method_action == 0:
                    action = "ENTER"
                elif method_action == 1:
                    action = "LEAVE"
                elif method_action == 2:
                    action = "UNROLL"
                elif method_action == 4:
                    action = "NATIVE ENTER"
                elif method_action == 5:
                    action = "NATIVE LEAVE"
                elif method_action == 6:
                    action = "NATIVE UNROLL" 
                method = self.methods[methodid]
                if threadid not in stacks:
                    stacks[threadid] = []
                stacks[threadid].append((action, method['methodName'], method['className'], method['lineNumber']))
                #ignore 4 chars
                buffer.read(4)
            except Exception as e:
                #done 
                return stacks 
        return stacks            

    def build_backtrace(self):
        links = []
        newline = chr(0x0a)
        threads = self.build_stacks(self.stacks)
        for s in threads:
            ENTER_list = []
            LEVAE_list = [] 
            for m in threads[s]:
                action = m[0]
                ENTER_list_length = len(ENTER_list)
                if action == "ENTER": 
                    this = " ".join((m[1], m[2], m[3])) 
                    ENTER_list.append(this) 
                    if ENTER_list_length + 1 > 0: 
                       links.append(newline.join(ENTER_list)) 
                elif action == "LEAVE":
                    if ENTER_list_length > 0: 
                        ENTER_list.pop() 
        return links

dm = DalvikTrace(data)
dm.build_backtrace()