path: root/scripts/convert2conformance/convert2conformance.py

#!/usr/bin/env python3
# Copyright (c) 2021-2023, ARM Limited.
# SPDX-License-Identifier: Apache-2.0
"""This script converts generated tests into conformance tests.

It can convert a framework unit test or a reference model unit test.
It expects the tests have been already run on the reference model
so it can capture the result as the expected result.
"""
import argparse
import json
import logging
import os
import shutil
from pathlib import Path
from typing import Optional

from json2fbbin.json2fbbin import fbbin_to_json
from json2numpy.json2numpy import npy_to_json
from schemavalidation.schemavalidation import TestDescSchemaValidator

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("convert2conformance")


NAME_FLATBUFFER_DIR = ["flatbuffer-", "_FW_"]
NAME_DESC_FILENAME = "desc.json"
NAME_CONFORMANCE_RESULT_PREFIX = "Conformance-"
NAME_REFMODEL_RUN_RESULT_SUFFIX = ".runner.tosa_refmodel_sut_run.npy"

PROFILES_LIST = ["tosa-bi", "tosa-mi"]

OUTPUT_TYPE_JSON = "json"
OUTPUT_TYPE_BINARY = "binary"
OUTPUT_TYPE_BOTH = "both"
OUTPUT_TYPES = (OUTPUT_TYPE_JSON, OUTPUT_TYPE_BINARY, OUTPUT_TYPE_BOTH)
OUTPUT_TYPE_DEFAULT = OUTPUT_TYPE_JSON


def parse_args(argv):
    """Parse the arguments."""
    # Set prog for when we are called via tosa_verif_conformance_generator
    parser = argparse.ArgumentParser(prog="convert2conformance")
    parser.add_argument(
        "test_dir",
        default=Path.cwd(),
        type=Path,
        nargs="?",
        help="The test directory to convert (default is CWD)",
    )
    parser.add_argument(
        "--schema-path",
        "--operator-fbs",
        dest="schema_path",
        type=Path,
        required=True,
        help="Path to reference model schema.",
    )
    parser.add_argument(
        "--flatc-path",
        dest="flatc_path",
        type=Path,
        required=True,
        help="Path to flatc executable.",
    )
    parser.add_argument(
        "--output-directory",
        dest="output_dir",
        type=Path,
        default=Path.cwd() / "conformance",
        help="Output directory (default is conformance in CWD)",
    )
    parser.add_argument(
        "--output-type",
        dest="output_type",
        choices=OUTPUT_TYPES,
        default=OUTPUT_TYPE_DEFAULT,
        help=f"Output file type produced (default is {OUTPUT_TYPE_DEFAULT})",
    )
    parser.add_argument(
        "--framework",
        dest="framework",
        choices=["tflite"],
        default="tflite",
        help="Framework to convert (default tflite)",
    )
    parser.add_argument(
        "--framework-schema",
        dest="framework_schema",
        type=Path,
        help="Framework schema needed to convert framework models",
    )
    parser.add_argument(
        "--profile",
        dest="profile",
        choices=PROFILES_LIST,
        action="append",
        required=True,
        help="Profiles this test is suitable for. May be repeated",
    )
    parser.add_argument(
        "--tag",
        dest="tags",
        action="append",
        type=str,
        help="Optional string tag to mark this test with. May be repeated",
    )
    parser.add_argument(
        "--strict",
        dest="strict",
        action="store_true",
        help="Output directory must not contain the same test directory",
    )
    parser.add_argument(
        "--lazy-data-generation",
        action="store_true",
        help="Enable lazy data generation (only for tosa-mi)",
    )
    parser.add_argument(
        "-v", "--verbose", dest="verbose", action="store_true", help="Verbose operation"
    )
    args = parser.parse_args(argv)

    return args


def find_framework_artifacts(framework: str, schema_path: Path, desc_file: Path):
    """Check that any required schema has been supplied for conversion."""
    if framework == "tflite":
        if not schema_path:
            raise Exception("the following arguments are required: --framework-schema")
        elif not schema_path.is_file():
            raise Exception(f"framework schema not found at {schema_path}")
        model = desc_file.parent.parent / "model.tflite"
        if not model.is_file():
            raise Exception(f"Model file not found at {model}")
        return schema_path, model
    return None, None


def get_framework_name(name_array: list, framework: str):
    """Get the framework conversion directory name."""
    name = ""
    for part in name_array:
        if part == "_FW_":
            part = framework
        name = f"{name}{part}"
    return name


def convert_flatbuffer_file(
    output_type: str, flatc: Path, schema: Path, model_file: Path, output: Path
):
    """Convert and/or copy the flatbuffer binary."""
    if output_type in (OUTPUT_TYPE_JSON, OUTPUT_TYPE_BOTH):
        try:
            fbbin_to_json(flatc, schema, model_file, output)
        except Exception as e:
            logger.error(f"Failed to convert flatbuffer binary:\n{e}")
            return None

        if model_file.name == "model.tflite":
            file_name = "model-tflite.json"
            os.rename(output / "model.json", output / file_name)
        else:
            file_name = model_file.stem + ".json"
    if output_type in (OUTPUT_TYPE_BINARY, OUTPUT_TYPE_BOTH):
        try:
            shutil.copy(model_file, output)
        except Exception as e:
            logger.error(f"Failed to copy flatbuffer binary:\n{e}")
            return None
        # By default return the binary name (if we have created both)
        file_name = model_file.name

    return output / file_name


def convert_numpy_file(
    output_type: str, npy_file: Path, output: Path, outstem: Optional[str] = None
):
    """Convert and/or copy the numpy file."""
    if output_type in (OUTPUT_TYPE_JSON, OUTPUT_TYPE_BOTH):
        new_file = output / ((outstem if outstem else npy_file.stem) + ".json")
        npy_to_json(npy_file, new_file)
    if output_type in (OUTPUT_TYPE_BINARY, OUTPUT_TYPE_BOTH):
        new_file = output / ((outstem + ".npy") if outstem else npy_file.name)
        shutil.copy(npy_file, new_file)


def update_desc_json(
    output_type: str,
    test_dir: Path,
    test_desc,
    output_dir: Optional[Path] = None,
    record_result=True,
    profiles=None,
    tags=None,
):
    """Update the desc.json format for conformance and optionally record result."""
    ofm_files = []
    cfm_files = []
    if not output_dir:
        output_dir = test_dir
    for index, ofm in enumerate(test_desc["ofm_file"]):
        ofm_path = test_dir / ofm
        if not test_desc["expected_failure"]:
            cfm = NAME_CONFORMANCE_RESULT_PREFIX + test_desc["ofm_name"][index]
            if record_result:
                if ofm_path.is_file():
                    # Use the desc.json name
                    ofm_refmodel = ofm_path
                else:
                    # Adjust for renaming due to tosa_verif_run_tests
                    ofm_refmodel = ofm_path.with_suffix(NAME_REFMODEL_RUN_RESULT_SUFFIX)
                # Create conformance result
                if ofm_refmodel.is_file():
                    convert_numpy_file(
                        output_type, ofm_refmodel, output_dir, outstem=cfm
                    )
                else:
                    logger.error(f"Missing result file {ofm_path}")
                    return None
                cfm_files.append(cfm + ".npy")
        # Remove path and "ref-"/"ref_model_" from output filenames
        ofm_files.append(strip_ref_output_name(ofm_path.name))

    # Rewrite output file names as they can be relative, but keep them npys
    test_desc["ofm_file"] = ofm_files
    if not test_desc["expected_failure"] and cfm_files:
        # Output expected result file for conformance if expected pass and we
        # have some files!
        test_desc["expected_result_file"] = cfm_files

    # Add supported profiles
    if profiles is None:
        # Assume base profile
        profiles = [PROFILES_LIST[0]]
    test_desc["profile"] = profiles

    # Add tags (if any)
    if tags is not None:
        test_desc["tag"] = tags

    return test_desc


def strip_ref_output_name(name):
    """Remove mentions of reference from output files."""
    if name.startswith("ref-"):
        name = name[4:]
    if name.startswith("ref_model_"):
        name = name[10:]
    return name


def main(argv=None):
    """Convert the given directory to a conformance test."""
    args = parse_args(argv)
    # Verbosity
    if args.verbose:
        logger.setLevel(logging.DEBUG)

    # Check we can get the files we need
    if not args.flatc_path.is_file():
        logger.error("flatc not found at %s", args.flatc_path)
        return 2
    if not args.schema_path.is_file():
        logger.error("TOSA schema not found at %s", args.schema_path)
        return 2

    # Work out where the desc.json file is
    desc_filename = args.test_dir / NAME_DESC_FILENAME
    framework_conversion = False
    test_type_desc = "unknown"
    if desc_filename.is_file():
        logger.debug("Found TOSA operator unit test")
        test_type_desc = "TOSA operator"
    else:
        desc_filename = (
            args.test_dir
            / get_framework_name(NAME_FLATBUFFER_DIR, args.framework)
            / NAME_DESC_FILENAME
        )
        if desc_filename.is_file():
            logger.debug(f"Found framework unit test for {args.framework}")
            test_type_desc = f"{args.framework}"
            framework_conversion = True
        else:
            logger.error(f"Could not find {NAME_DESC_FILENAME} in {args.test_dir}")
            return 2
    logger.debug(f"desc.json file: {desc_filename}")

    # Check for required files for framework conversion
    if framework_conversion:
        try:
            framework_schema, framework_filename = find_framework_artifacts(
                args.framework, args.framework_schema, desc_filename
            )
        except Exception as err:
            logger.error(err)
            return 2
    else:
        framework_schema, framework_filename = None, None

    # Open the meta desc.json file
    with open(desc_filename, mode="r") as fd:
        test_desc = json.load(fd)

    if "tosa_file" not in test_desc:
        logger.error(f"Unsupported desc.json file found {desc_filename}")
        return 2

    # Dictionary fix
    if "ifm_name" not in test_desc:
        logger.warn("Old format desc.json file found - attempting to fix up")
        test_desc["ifm_name"] = test_desc["ifm_placeholder"]
        del test_desc["ifm_placeholder"]

    # Make the output directory if needed
    try:
        args.output_dir.mkdir(parents=True, exist_ok=(not args.strict))
    except FileExistsError:
        if args.strict:
            logger.error(f"{args.output_dir} already exists")
        else:
            logger.error(f"{args.output_dir} is not a directory")
        return 2

    # Convert the TOSA flatbuffer binary
    tosa_filename = desc_filename.parent / test_desc["tosa_file"]
    tosa_filename = convert_flatbuffer_file(
        args.output_type,
        args.flatc_path,
        args.schema_path,
        tosa_filename,
        args.output_dir,
    )
    if not tosa_filename:
        # Failed to convert the file, json2fbbin will have printed an error
        return 1
    else:
        # Replace binary with JSON name
        test_desc["tosa_file"] = tosa_filename.name

    if framework_conversion and framework_filename:
        # Convert the framework flatbuffer binary
        framework_filename = convert_flatbuffer_file(
            args.output_type,
            args.flatc_path,
            framework_schema,
            framework_filename,
            args.output_dir,
        )
        if not framework_filename:
            # Failed to convert the file, json2fbbin will have printed an error
            return 1

    # Convert input files to JSON
    ifm_files = []
    for file in test_desc["ifm_file"]:
        if file:
            path = desc_filename.parent / file
            ifm_files.append(path.name)
            if path.is_file():
                convert_numpy_file(args.output_type, path, args.output_dir)
            else:
                if not args.lazy_data_generation:
                    logger.error(f"Missing input file {path.name}")
                    return 1

    # Rewrite input file names to make sure the paths are correct,
    # but keep them numpys as the test runner will convert them back
    # before giving them to the SUT
    test_desc["ifm_file"] = ifm_files

    # Check for cpp files for data-generator/verifier
    cpp_files = args.test_dir.glob("*.cpp")
    for cpp in cpp_files:
        shutil.copy(str(cpp), str(args.output_dir))

    # Work out if we have a result to record
    record_result = not args.lazy_data_generation
    if "meta" in test_desc and "compliance" in test_desc["meta"]:
        # We don't have pre-generated results for compliance tests
        record_result = False

    # Update desc.json and convert result files to JSON
    test_desc = update_desc_json(
        args.output_type,
        desc_filename.parent,
        test_desc,
        output_dir=args.output_dir,
        record_result=record_result,
        profiles=args.profile,
        tags=args.tags,
    )
    if not test_desc:
        # Error from conversion/update
        return 1

    # Validate the desc.json schema
    try:
        TestDescSchemaValidator().validate_config(test_desc)
    except Exception as e:
        logger.error(e)
        return 1

    # Output new desc.json
    new_desc_filename = args.output_dir / NAME_DESC_FILENAME
    with open(new_desc_filename, "w") as fd:
        json.dump(test_desc, fd, indent=2)

    logger.info(f"Converted {test_type_desc} test to {args.output_dir}")
    return 0


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