zephyr/scripts/logging/dictionary/dictionary_parser/log_database.py

#!/usr/bin/env python3
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0

"""
Class for Dictionary-based Logging Database
"""

import base64
import copy
import json

from .mipi_syst import gen_syst_xml_file
from .utils import extract_one_string_in_section
from .utils import find_string_in_mappings


ARCHS = {
    "arc" : {
        "kconfig": "CONFIG_ARC",
    },
    "arm" : {
        "kconfig": "CONFIG_ARM",
    },
    "arm64" : {
        "kconfig": "CONFIG_ARM64",
    },
    "mips" : {
        "kconfig": "CONFIG_MIPS",
    },
    "sparc" : {
        "kconfig": "CONFIG_SPARC",
    },
    "x86" : {
        "kconfig": "CONFIG_X86",
    },
    "nios2" : {
        "kconfig": "CONFIG_NIOS2",

        # Small static strings are put into section "datas"
        # so we need to include them also.
        #
        # See include/arch/nios2/linker.ld on .sdata.*
        # for explanation.
        "extra_string_section": ['datas'],
    },
    "riscv" : {
        "kconfig": "CONFIG_RISCV",
    },
    "xtensa" : {
        "kconfig": "CONFIG_XTENSA",
    },
}


class LogDatabase():
    """Class of log database"""
    # Update this if database format of dictionary based logging
    # has changed
    ZEPHYR_DICT_LOG_VER = 2

    LITTLE_ENDIAN = True
    BIG_ENDIAN = False

    def __init__(self):
        new_db = {}

        new_db['version'] = self.ZEPHYR_DICT_LOG_VER
        new_db['target'] = {}
        new_db['log_subsys'] = {}
        new_db['log_subsys']['log_instances'] = {}
        new_db['build_id'] = None
        new_db['arch'] = None
        new_db['kconfigs'] = {}

        self.database = new_db


    def get_version(self):
        """Get Database Version"""
        return self.database['version']


    def get_build_id(self):
        """Get Build ID"""
        return self.database['build_id']


    def set_build_id(self, build_id):
        """Set Build ID in Database"""
        self.database['build_id'] = build_id


    def get_arch(self):
        """Get the Target Architecture"""
        return self.database['arch']


    def set_arch(self, arch):
        """Set the Target Architecture"""
        self.database['arch'] = arch


    def get_tgt_bits(self):
        """Get Target Bitness: 32 or 64"""
        if 'bits' in self.database['target']:
            return self.database['target']['bits']

        return None


    def set_tgt_bits(self, bits):
        """Set Target Bitness: 32 or 64"""
        self.database['target']['bits'] = bits


    def is_tgt_64bit(self):
        """Return True if target is 64-bit, False if 32-bit.
        None if error."""
        if 'bits' not in self.database['target']:
            return None

        if self.database['target']['bits'] == 32:
            return False

        if self.database['target']['bits'] == 64:
            return True

        return None


    def get_tgt_endianness(self):
        """
        Get Target Endianness.

        Return True if little endian, False if big.
        """
        if 'little_endianness' in self.database['target']:
            return self.database['target']['little_endianness']

        return None


    def set_tgt_endianness(self, endianness):
        """
        Set Target Endianness

        True if little endian, False if big.
        """
        self.database['target']['little_endianness'] = endianness


    def is_tgt_little_endian(self):
        """Return True if target is little endian"""
        if 'little_endianness' not in self.database['target']:
            return None

        return self.database['target']['little_endianness'] == self.LITTLE_ENDIAN


    def get_string_mappings(self):
        """Get string mappings to database"""
        return self.database['string_mappings']


    def set_string_mappings(self, database):
        """Add string mappings to database"""
        self.database['string_mappings'] = database


    def has_string_mappings(self):
        """Return True if there are string mappings in database"""
        if 'string_mappings' in self.database:
            return True

        return False


    def has_string_sections(self):
        """Return True if there are any static string sections"""
        if 'sections' not in self.database:
            return False

        return len(self.database['sections']) != 0


    def __find_string_in_mappings(self, string_ptr):
        """
        Find string pointed by string_ptr in the string mapping
        list. Return None if not found.
        """
        return find_string_in_mappings(self.database['string_mappings'], string_ptr)


    def __find_string_in_sections(self, string_ptr):
        """
        Find string pointed by string_ptr in the binary data
        sections. Return None if not found.
        """
        for _, sect in self.database['sections'].items():
            one_str = extract_one_string_in_section(sect, string_ptr)

            if one_str is not None:
                return one_str

        return None


    def find_string(self, string_ptr):
        """Find string pointed by string_ptr in the database.
        Return None if not found."""
        one_str = None

        if self.has_string_mappings():
            one_str = self.__find_string_in_mappings(string_ptr)

        if one_str is None and self.has_string_sections():
            one_str = self.__find_string_in_sections(string_ptr)

        return one_str


    def add_log_instance(self, source_id, name, level, address):
        """Add one log instance into database"""
        self.database['log_subsys']['log_instances'][source_id] = {
            'source_id' : source_id,
            'name'      : name,
            'level'     : level,
            'addr'      : address,
        }


    def get_log_source_string(self, domain_id, source_id):
        """Get the source string based on source ID"""
        # JSON stores key as string, so we need to convert
        src_id = str(source_id)

        if src_id in self.database['log_subsys']['log_instances']:
            return self.database['log_subsys']['log_instances'][src_id]['name']

        return f"unknown<{domain_id}:{source_id}>"


    def add_kconfig(self, name, val):
        """Add a kconfig name-value pair into database"""
        self.database['kconfigs'][name] = val


    def get_kconfigs(self):
        """Return kconfig name-value pairs"""
        return self.database['kconfigs']


    @staticmethod
    def read_json_database(db_file_name):
        """Read database from file and return a LogDatabase object"""
        try:
            with open(db_file_name, "r", encoding="iso-8859-1") as db_fd:
                json_db = json.load(db_fd)
        except (OSError, json.JSONDecodeError):
            return None

        # Decode data in JSON back into binary data
        if 'sections' in json_db:
            for _, sect in json_db['sections'].items():
                sect['data'] = base64.b64decode(sect['data_b64'])

        database = LogDatabase()
        database.database = json_db

        # JSON encodes the addresses in string mappings as literal strings.
        # So convert them back to integers, as this is needed for partial
        # matchings.
        if database.has_string_mappings():
            new_str_map = {}

            for addr, one_str in database.get_string_mappings().items():
                new_str_map[int(addr)] = one_str

            database.set_string_mappings(new_str_map)

        return database


    @staticmethod
    def write_json_database(db_file_name, database):
        """Write the database into file"""
        json_db = copy.deepcopy(database.database)

        # Make database object into something JSON can dump
        if 'sections' in json_db:
            for _, sect in json_db['sections'].items():
                encoded = base64.b64encode(sect['data'])
                sect['data_b64'] = encoded.decode('ascii')
                del sect['data']

        try:
            with open(db_file_name, "w", encoding="iso-8859-1") as db_fd:
                db_fd.write(json.dumps(json_db))
        except OSError:
            return False

        return True

    @staticmethod
    def write_syst_database(db_file_name, database):
        """
        Write the database into MIPI Sys-T Collateral XML file
        """

        try:
            with open(db_file_name, "w", encoding="iso-8859-1") as db_fd:
                xml = gen_syst_xml_file(database)
                db_fd.write(xml)
        except OSError:
            return False

        return True