xref: /third_party/astc-encoder/Test/astc_trace_analysis.py

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# -----------------------------------------------------------------------------
# Copyright 2021 Arm Limited
#
# 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.
# -----------------------------------------------------------------------------
"""
The ``astc_trace_analysis`` utility provides a tool to analyze trace files.

WARNING: Trace files are an engineering tool, and not part of the standard
product, so traces and their associated tools are volatile and may change
significantly without notice.
"""

import argparse
from collections import defaultdict as ddict
import json
import numpy as np
import sys

QUANT_TABLE = {
	 0:   2,
	 1:   3,
	 2:   4,
	 3:   5,
	 4:   6,
	 5:   8,
	 6:  10,
	 7:  12,
	 8:  16,
	 9:  20,
	10:  24,
	11:  32
}

CHANNEL_TABLE = {
	 0: "R",
	 1: "G",
	 2: "B",
	 3: "A"
}

class Trace:

    def __init__(self, block_x, block_y, block_z):
        self.block_x = block_x
        self.block_y = block_y
        self.block_z = block_z
        self.blocks = []

    def add_block(self, block):
        self.blocks.append(block)

    def __getitem__(self, i):
       return self.blocks[i]

    def __delitem__(self, i):
       del self.blocks[i]

    def __len__(self):
       return len(self.blocks)

class Block:

    def __init__(self, pos_x, pos_y, pos_z, error_target):
        self.pos_x = pos_x
        self.pos_y = pos_y
        self.pos_z = pos_z

        self.raw_min = None
        self.raw_max = None

        self.ldr_min = None
        self.ldr_max = None

        self.error_target = error_target
        self.passes = []
        self.qualityHit = None

    def add_minimums(self, r, g, b, a):
        self.raw_min = (r, g, b, a)

        def ldr(x):
            cmax = 65535.0
            return int((r / cmax) * 255.0)

        self.ldr_min = (ldr(r), ldr(g), ldr(b), ldr(a))

    def add_maximums(self, r, g, b, a):
        self.raw_max = (r, g, b, a)

        def ldr(x):
            cmax = 65535.0
            return int((r / cmax) * 255.0)

        self.ldr_max = (ldr(r), ldr(g), ldr(b), ldr(a))

    def add_pass(self, pas):
        self.passes.append(pas)

    def __getitem__(self, i):
       return self.passes[i]

    def __delitem__(self, i):
       del self.passes[i]

    def __len__(self):
       return len(self.passes)


class Pass:

    def __init__(self, partitions, partition, planes, target_hit, mode, component):
        self.partitions = partitions
        self.partition_index = 0 if partition is None else partition
        self.planes = planes
        self.plane2_component = component
        self.target_hit = target_hit
        self.search_mode = mode
        self.candidates = []

    def add_candidate(self, candidate):
        self.candidates.append(candidate)

    def __getitem__(self, i):
       return self.candidates[i]

    def __delitem__(self, i):
       del self.candidates[i]

    def __len__(self):
       return len(self.candidates)


class Candidate:

    def __init__(self, weight_x, weight_y, weight_z, weight_quant):
        self.weight_x = weight_x
        self.weight_y = weight_y
        self.weight_z = weight_z
        self.weight_quant = weight_quant
        self.refinement_errors = []

    def add_refinement(self, errorval):
        self.refinement_errors.append(errorval)


def get_attrib(data, name, multiple=False, hard_fail=True):
    results = []
    for attrib in data:
        if len(attrib) == 2 and attrib[0] == name:
            results.append(attrib[1])

    if not results:
        if hard_fail:
            print(json.dumps(data, indent=2))
            assert False, "Attribute %s not found" % name
        if multiple:
            return list()
        return None

    if not multiple:
        if len(results) > 1:
            print(json.dumps(data, indent=2))
            assert False, "Attribute %s found %u times" % (name, len(results))
        return results[0]

    return results


def rev_enumerate(seq):
    return zip(reversed(range(len(seq))), reversed(seq))

def foreach_block(data):

    for block in data:
        yield block

def foreach_pass(data):

    for block in data:
        for pas in block:
            yield (block, pas)

def foreach_candidate(data):

    for block in data:
        for pas in block:
            # Special case - None candidates for 0 partition
            if not len(pas):
                yield (block, pas, None)

            for candidate in pas:
                yield (block, pas, candidate)

def get_node(data, name, multiple=False, hard_fail=True):
    results = []
    for attrib in data:
        if len(attrib) == 3 and attrib[0] == "node" and attrib[1] == name:
            results.append(attrib[2])

    if not results:
        if hard_fail:
            print(json.dumps(data, indent=2))
            assert False, "Node %s not found" % name
        return None

    if not multiple:
        if len(results) > 1:
            print(json.dumps(data, indent=2))
            assert False, "Node %s found %u times" % (name, len(results))
        return results[0]

    return results


def find_best_pass_and_candidate(block):
    explicit_pass = None

    best_error = 1e30
    best_pass = None
    best_candidate = None

    for pas in block:
        # Special case for constant color blocks - no trial candidates
        if pas.target_hit and pas.partitions == 0:
            return (pas, None)

        for candidate in pas:
            errorval = candidate.refinement_errors[-1]
            if errorval <= best_error:
                best_error = errorval
                best_pass = pas
                best_candidate = candidate

    # Every other return type must have both best pass and best candidate
    assert (best_pass and best_candidate)
    return (best_pass, best_candidate)


def generate_database(data):
    # Skip header
    assert(data[0] == "node")
    assert(data[1] == "root")
    data = data[2]

    bx = get_attrib(data, "block_x")
    by = get_attrib(data, "block_y")
    bz = get_attrib(data, "block_z")
    dbStruct = Trace(bx, by, bz)

    for block in get_node(data, "block", True):
        px = get_attrib(block, "pos_x")
        py = get_attrib(block, "pos_y")
        pz = get_attrib(block, "pos_z")

        minr = get_attrib(block, "min_r")
        ming = get_attrib(block, "min_g")
        minb = get_attrib(block, "min_b")
        mina = get_attrib(block, "min_a")

        maxr = get_attrib(block, "max_r")
        maxg = get_attrib(block, "max_g")
        maxb = get_attrib(block, "max_b")
        maxa = get_attrib(block, "max_a")

        et = get_attrib(block, "tune_error_threshold")

        blockStruct = Block(px, py, pz, et)
        blockStruct.add_minimums(minr, ming, minb, mina)
        blockStruct.add_maximums(maxr, maxg, maxb, maxa)
        dbStruct.add_block(blockStruct)

        for pas in get_node(block, "pass", True):
            # Don't copy across passes we skipped due to heuristics
            skipped = get_attrib(pas, "skip", False, False)
            if skipped:
                continue

            prts = get_attrib(pas, "partition_count")
            prti = get_attrib(pas, "partition_index", False, False)
            plns = get_attrib(pas, "plane_count")
            chan = get_attrib(pas, "plane_component", False, plns > 2)
            mode = get_attrib(pas, "search_mode", False, False)
            ehit = get_attrib(pas, "exit", False, False) == "quality hit"

            passStruct = Pass(prts, prti, plns, ehit, mode, chan)
            blockStruct.add_pass(passStruct)

            # Constant color blocks don't have any candidates
            if prts == 0:
                continue

            for candidate in get_node(pas, "candidate", True):
                # Don't copy across candidates we couldn't encode
                failed = get_attrib(candidate, "failed", False, False)
                if failed:
                    continue

                wx = get_attrib(candidate, "weight_x")
                wy = get_attrib(candidate, "weight_y")
                wz = get_attrib(candidate, "weight_z")
                wq = QUANT_TABLE[get_attrib(candidate, "weight_quant")]
                epre = get_attrib(candidate, "error_prerealign", True, False)
                epst = get_attrib(candidate, "error_postrealign", True, False)

                candStruct = Candidate(wx, wy, wz, wq)
                passStruct.add_candidate(candStruct)
                for value in epre:
                    candStruct.add_refinement(value)
                for value in epst:
                    candStruct.add_refinement(value)

    return dbStruct


def filter_database(data):

    for block in data:
        best_pass, best_candidate = find_best_pass_and_candidate(block)

        for i, pas in rev_enumerate(block):
            if pas != best_pass:
                del block[i]
                continue

            if best_candidate is None:
                continue

            for j, candidate in rev_enumerate(pas):
                if candidate != best_candidate:
                    del pas[j]


def generate_pass_statistics(data):
    pass


def generate_feature_statistics(data):
    # -------------------------------------------------------------------------
    # Config
    print("Compressor Config")
    print("=================")

    if data.block_z > 1:
        dat = (data.block_x, data.block_y, data.block_z)
        print("  - Block size: %ux%ux%u" % dat)
    else:
        dat = (data.block_x, data.block_y)
        print("  - Block size: %ux%u" % dat)

    print("")

    # -------------------------------------------------------------------------
    # Block metrics
    result = ddict(int)

    RANGE_QUANT = 16

    for block in foreach_block(data):
        ranges = []
        for i in range(0, 4):
            ranges.append(block.ldr_max[i] - block.ldr_min[i])

        max_range = max(ranges)
        max_range = int(max_range / RANGE_QUANT) * RANGE_QUANT

        result[max_range] += 1

    print("Channel Range")
    print("=============")
    keys = sorted(result.keys())
    for key in keys:
        dat = (key, key + RANGE_QUANT - 1, result[key])
        print("  - %3u-%3u dynamic range = %6u blocks" % dat)

    print("")

    # -------------------------------------------------------------------------
    # Partition usage
    result_totals = ddict(int)
    results = ddict(lambda: ddict(int))

    for _, pas in foreach_pass(data):
        result_totals[pas.partitions] += 1
        results[pas.partitions][pas.partition_index] += 1

    print("Partition Count")
    print("===============")
    keys = sorted(result_totals.keys())
    for key in keys:
        dat = (key, result_totals[key], len(results[key]))
        print("  - %u partition(s) = %6u blocks / %4u indicies" % dat)

    print("")

    # -------------------------------------------------------------------------
    # Plane usage
    result_count = ddict(lambda: ddict(int))
    result_channel = ddict(lambda: ddict(int))

    for _, pas in foreach_pass(data):
        result_count[pas.partitions][pas.planes] += 1
        if (pas.planes > 1):
            result_channel[pas.partitions][pas.plane2_component] += 1

    print("Plane Usage")
    print("===========")
    keys = sorted(result_count.keys())
    for key in keys:
        keys2 = sorted(result_count[key])
        for key2 in keys2:
            val2 = result_count[key][key2]
            dat = (key, key2, val2)
            print("  - %u partition(s) %u plane(s) = %6u blocks" % dat)
            if key2 == 2:
                keys3 = sorted(result_channel[key])
                for key3 in keys3:
                    dat = (CHANNEL_TABLE[key3], result_channel[key][key3])
                    print("    - %s plane                 = %6u blocks" % dat)

    print("")

    # -------------------------------------------------------------------------
    # Decimation usage
    decim_count = ddict(lambda: ddict(int))
    quant_count = ddict(lambda: ddict(lambda: ddict(int)))


    MERGE_ROTATIONS = True

    for _, pas, can in foreach_candidate(data):
        # Skip constant color blocks
        if can is None:
            continue

        wx = can.weight_x
        wy = can.weight_y

        if MERGE_ROTATIONS and wx < wy:
            wx, wy = wy, wx

        decim_count[wx][wy] += 1
        quant_count[wx][wy][can.weight_quant] += 1

    print("Decimation Usage")
    print("================")

    if MERGE_ROTATIONS:
        print("  - Note: data merging grid rotations")

    x_keys = sorted(decim_count.keys())
    for x_key in x_keys:
        y_keys = sorted(decim_count[x_key])

        for y_key in y_keys:
            count = decim_count[x_key][y_key]
            dat = (x_key, y_key, count)
            print("  - %ux%u weights      = %6u blocks" % dat)

            q_keys = sorted(quant_count[x_key][y_key])
            for q_key in q_keys:
                count = quant_count[x_key][y_key][q_key]
                dat = (q_key, count)
                print("    - %2u quant range = %6u blocks" % dat)

    print("")

    # -------------------------------------------------------------------------
    # Refinement usage

    total_count = 0
    better_count = 0
    could_have_count = 0
    success_count = 0

    refinement_step = []

    for block, pas, candidate in foreach_candidate(data):
        # Ignore zero partition blocks - they don't use refinement
        if not candidate:
            continue

        target_error = block.error_target
        start_error = candidate.refinement_errors[0]
        end_error = candidate.refinement_errors[-1]

        rpf = float(start_error - end_error) / float(len(candidate.refinement_errors))
        rpf = abs(rpf)
        refinement_step.append(rpf / start_error)

        total_count += 1
        if end_error <= start_error:
            better_count += 1

        if end_error <= target_error:
            success_count += 1
        else:
            for refinement in candidate.refinement_errors:
                if refinement <= target_error:
                    could_have_count += 1
                    break


    print("Refinement Usage")
    print("================")
    print("  - %u refinements(s)" % total_count)
    print("  - %u refinements(s) improved" % better_count)
    print("  - %u refinements(s) worsened" % (total_count - better_count))
    print("  - %u refinements(s) could hit target, but didn't" % could_have_count)
    print("  - %u refinements(s) hit target" % success_count)
    print("  - %f mean step improvement" % np.mean(refinement_step))


def parse_command_line():
    """
    Parse the command line.

    Returns:
        Namespace: The parsed command line container.
    """
    parser = argparse.ArgumentParser()

    parser.add_argument("trace", type=argparse.FileType("r"),
                        help="The trace file to analyze")

    return parser.parse_args()


def main():
    """
    The main function.

    Returns:
        int: The process return code.
    """
    args = parse_command_line()

    data = json.load(args.trace)
    db = generate_database(data)
    filter_database(db)

    generate_feature_statistics(db)

    return 0


if __name__ == "__main__":
    sys.exit(main())