xref: /aosp_15_r20/external/oboe/apps/OboeTester/scripts/dsp_timing.py (revision 05767d913155b055644481607e6fa1e35e2fe72c)

#!/usr/bin/python2.7
"""
Copyright (C) 2019 The Android Open Source Project

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""

# Run OboeTester with progressive timing changes
# to measure the DSP timing profile.
# Print a CSV table of offsets and glitch counts.
#
# Run Automated Test using an Intent
# https://github.com/google/oboe/blob/main/apps/OboeTester/docs/AutomatedTesting.md

import array
import collections
import os
import os.path
import sys
import subprocess
import time

from datetime import datetime

kPropertyOutputOffset = "aaudio.out_mmap_offset_usec"
kMinPeakAmplitude = 0.04
kOffsetMin = 0000
kOffsetMax = 4000
kOffsetIncrement = 100
kOutputFileBase = "/sdcard/dsp_timing_"
gOutputFile = kOutputFileBase + "now.txt"

def launchLatencyTest():
    command = ["adb", "shell", "am", \
               "start", "-n", "com.mobileer.oboetester/.MainActivity", \
               "--es", "test", "latency", \
               "--es", "file", gOutputFile, \
               "--ei", "buffer_bursts", "1"]
    return subprocess.check_output(command)

def launchGlitchTest():
    command = ["adb", "shell", "am", \
               "start", "-n", "com.mobileer.oboetester/.MainActivity", \
               "--es", "test", "glitch", \
               "--es", "file", gOutputFile, \
               "--es", "in_perf", "lowlat", \
               "--es", "out_perf", "lowlat", \
               "--es", "in_sharing", "exclusive", \
               "--es", "out_sharing", "exclusive", \
               "--ei", "buffer_bursts", "1"]
    return subprocess.check_output(command)

def setAndroidProperty(property, value):
    return subprocess.check_output(["adb", "shell", "setprop", property, value])

def getAndroidProperty(property):
    return subprocess.check_output(["adb", "shell", "getprop", property])

def setOutputOffset(offset):
    setAndroidProperty(kPropertyOutputOffset, str(offset))

def getOutputOffset():
    offsetText = getAndroidProperty(kPropertyOutputOffset).strip()
    if len(offsetText) == 0:
        return 0;
    return int(offsetText)

def loadNameValuePairsFromFile(filename):
    myvars = {}
    with open(filename) as myfile:
        for line in myfile:
            name, var = line.partition("=")[::2]
            myvars[name.strip()] = var
    return myvars

def loadNameValuePairsFromString(text):
    myvars = {}
    listOutput = text.splitlines()
    for line in listOutput:
            name, var = line.partition("=")[::2]
            myvars[name.strip()] = var
    return myvars

def waitForTestResult():
    testOutput = ""
    for i in range(10):
        if subprocess.call(["adb", "shell", "ls", gOutputFile, "2>/dev/null"]) == 0:
            testOutput = subprocess.check_output(["adb", "shell", "cat", gOutputFile])
            break
        else:
            print str(i) + ": waiting until test finishes..."
            time.sleep(2)
    # print testOutput
    subprocess.call(["adb", "shell", "rm", gOutputFile])
    return loadNameValuePairsFromString(testOutput)

# volume too low?
# return true if bad
def checkPeakAmplitude(pairs):
    if not pairs.has_key('peak.amplitude'):
        print "ERROR no peak.amplitude"
        return True
    peakAmplitude = float(pairs['peak.amplitude'])
    if peakAmplitude < kMinPeakAmplitude:
        print "ERROR peakAmplitude = " + str(peakAmplitude) \
            + " < " + str(kMinPeakAmplitude) \
            + ", turn up volume"
        return True
    return False

def startOneTest(offset):
    print "=========================="
    setOutputOffset(offset)
    print "try offset = " + getAndroidProperty(kPropertyOutputOffset)
    subprocess.call(["adb", "shell", "rm", gOutputFile, "2>/dev/null"])

def runOneGlitchTest(offset):
    startOneTest(offset)
    print launchGlitchTest()
    pairs = waitForTestResult()
    if checkPeakAmplitude(pairs):
        return -1
    if not pairs.has_key('glitch.count'):
        print "ERROR no glitch.count"
        return -1
    return int(pairs['glitch.count'])

def runOneLatencyTest(offset):
    startOneTest(offset)
    print launchLatencyTest()
    pairs = waitForTestResult()
    if not pairs.has_key('latency.msec'):
        print "ERROR no latency.msec"
        return -1
    return float(pairs['latency.msec'])

def runGlitchSeries():
    offsets = array.array('i')
    glitches = array.array('i')
    for offset in range(kOffsetMin, kOffsetMax, kOffsetIncrement):
        offsets.append(offset)
        result = runOneGlitchTest(offset)
        glitches.append(result)
        if result < 0:
            break
        print "offset = " + str(offset) + ", glitches = " + str(result)
    print "offsetUs, glitches"
    for i in range(len(offsets)):
        print "  " + str(offsets[i]) + ", " + str(glitches[i])

# return true if bad
def checkLatencyValid():
    startOneTest(0)
    print launchLatencyTest()
    pairs = waitForTestResult()
    print "burst = " + pairs['out.burst.frames']
    capacity = int(pairs['out.buffer.capacity.frames'])
    if capacity < 0:
        print "ERROR capacity = " + str(capacity)
        return True
    sampleRate = int(pairs['out.rate'])
    capacityMillis = capacity * 1000.0 / sampleRate
    print "capacityMillis = " + str(capacityMillis)
    if pairs['in.mmap'].strip() != "yes":
        print "ERROR Not using input MMAP"
        return True
    if pairs['out.mmap'].strip() != "yes":
        print "ERROR Not using output MMAP"
        return True
    # Check whether we can change latency a moving the DSP pointer
    # past the CPU pointer and wrapping the buffer.
    latencyMin = runOneLatencyTest(kOffsetMin)
    latencyMax = runOneLatencyTest(kOffsetMax + 1000)
    print "latency = " + str(latencyMin) + " => " + str(latencyMax)
    if latencyMax < (latencyMin + (capacityMillis / 2)):
        print "ERROR Latency not affected by changing offset!"
        return True
    return False

def isRunningAsRoot():
    userName = subprocess.check_output(["adb", "shell", "whoami"]).strip()
    if userName != "root":
        print "WARNING: changing to 'adb root'"
        subprocess.call(["adb", "root"])
        userName = subprocess.check_output(["adb", "shell", "whoami"]).strip()
        if userName != "root":
            print "ERROR: cannot set 'adb root'"
            return False
    return True

def isMMapSupported():
    mmapPolicy = int(getAndroidProperty("aaudio.mmap_policy").strip())
    if mmapPolicy < 2:
        print "ERROR: AAudio MMAP not enabled, aaudio.mmap_policy = " + str(mmapPolicy)
        return False
    if checkLatencyValid():
        return False;
    return True

def isTimingSeriesSupported():
    if not isRunningAsRoot():
        return False
    if not isMMapSupported():
        return False
    return True

def main():
    global gOutputFile
    print "gOutputFile = " + gOutputFile
    now = datetime.now() # current date and time
    gOutputFile = kOutputFileBase \
            + now.strftime("%Y%m%d_%H%M%S") \
            + ".txt"
    print "gOutputFile = " + gOutputFile

    initialOffset = getOutputOffset()
    print "initial offset = " + str(initialOffset)

    print "Android version = " + \
            getAndroidProperty("ro.build.id").strip()
    print "    release " + \
            getAndroidProperty("ro.build.version.release").strip()
    if (isTimingSeriesSupported()):
        runGlitchSeries()
        setOutputOffset(initialOffset)

if __name__ == '__main__':
    main()