path: root/agl_service_voiceagent/utils/mapper.py

# SPDX-License-Identifier: Apache-2.0
#
# Copyright (c) 2023 Malik Talha
#
# 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.

from agl_service_voiceagent.utils.config import get_config_value, get_logger
from agl_service_voiceagent.utils.common import load_json_file, words_to_number


class Intent2VSSMapper:
    """
    Intent2VSSMapper is a class that facilitates the mapping of natural language intent to
    corresponding vehicle signal specifications (VSS) for automated vehicle control systems.
    """

    def __init__(self):
        """
        Initializes the Intent2VSSMapper class by loading Intent-to-VSS signal mappings
        and VSS signal specifications from external configuration files.
        """
        intents_vss_map_file = get_config_value("intents_vss_map", "Mapper")
        vss_signals_spec_file = get_config_value("vss_signals_spec", "Mapper")
        self.intents_vss_map = load_json_file(intents_vss_map_file).get("intents", {})
        self.vss_signals_spec = load_json_file(vss_signals_spec_file).get("signals", {})
        self.logger = get_logger()

        if not self.validate_signal_spec_structure():
            raise ValueError("[-] Invalid VSS signal specification structure.")

    def validate_signal_spec_structure(self):
        """
        Validates the structure of the VSS signal specification data.
        """

        signals = self.vss_signals_spec

        # Iterate over each signal in the 'signals' dictionary
        for signal_name, signal_data in signals.items():
            # Check if the required keys are present in the signal data
            if not all(key in signal_data for key in ['default_value', 'default_change_factor', 'actions', 'values', 'default_fallback', 'value_set_intents']):
                print(f"[-] {signal_name}: Missing required keys in signal data.")
                self.logger.error(f"{signal_name}: Missing required keys in signal data.")
                return False

            actions = signal_data['actions']

            # Check if 'actions' is a dictionary with at least one action
            if not isinstance(actions, dict) or not actions:
                print(f"[-] {signal_name}: Invalid 'actions' key in signal data. Must be an object with at least one action.")
                self.logger.error(f"{signal_name}: Invalid 'actions' key in signal data. Must be an object with at least one action.")
                return False

            # Check if the actions match the allowed actions ["set", "increase", "decrease"]
            for action in actions.keys():
                if action not in ["set", "increase", "decrease"]:
                    print(f"[-] {signal_name}: Invalid action in signal data. Allowed actions: ['set', 'increase', 'decrease']")
                    self.logger.error(f"{signal_name}: Invalid action in signal data. Allowed actions: ['set', 'increase', 'decrease']")
                    return False

            # Check if the 'synonyms' list is present for each action and is either a list or None
            for action_data in actions.values():
                synonyms = action_data.get('synonyms')
                if synonyms is not None and (not isinstance(synonyms, list) or not all(isinstance(synonym, str) for synonym in synonyms)):
                    print(f"[-] {signal_name}: Invalid 'synonyms' value in signal data. Must be a list of strings.")
                    self.logger.error(f"{signal_name}: Invalid 'synonyms' value in signal data. Must be a list of strings.")
                    return False

            values = signal_data['values']

            # Check if 'values' is a dictionary with the required keys
            if not isinstance(values, dict) or not all(key in values for key in ['ranged', 'start', 'end', 'ignore', 'additional']):
                print(f"[-] {signal_name}: Invalid 'values' key in signal data. Required keys: ['ranged', 'start', 'end', 'ignore', 'additional']")
                self.logger.error(f"{signal_name}: Invalid 'values' key in signal data. Required keys: ['ranged', 'start', 'end', 'ignore', 'additional']")
                return False

            # Check if 'ranged' is a boolean
            if not isinstance(values['ranged'], bool):
                print(f"[-] {signal_name}: Invalid 'ranged' value in signal data. Allowed values: [true, false]")
                self.logger.error(f"{signal_name}: Invalid 'ranged' value in signal data. Allowed values: [true, false]")
                return False

            default_fallback = signal_data['default_fallback']

            # Check if 'default_fallback' is a boolean
            if not isinstance(default_fallback, bool):
                print(f"[-] {signal_name}: Invalid 'default_fallback' value in signal data. Allowed values: [true, false]")
                self.logger.error(f"{signal_name}: Invalid 'default_fallback' value in signal data. Allowed values: [true, false]")
                return False

        # If all checks pass, the self.vss_signals_spec structure is valid
        return True


    def map_intent_to_signal(self, intent_name):
        """
        Maps an intent name to the corresponding VSS signals and their specifications.

        Args:
            intent_name (str): The name of the intent to be mapped.

        Returns:
            dict: A dictionary containing VSS signals as keys and their specifications as values.
        """
    
        intent_data = self.intents_vss_map.get(intent_name, None)
        result = {}
        if intent_data:
            signals = intent_data.get("signals", [])

            for signal in signals:
                signal_info = self.vss_signals_spec.get(signal, {})
                if signal_info:
                    result.update({signal: signal_info})
        
        return result


    def parse_intent(self, intent_name, intent_slots = []):
        """
        Parses an intent, extracting relevant VSS signals, actions, modifiers, and values
        based on the intent and its associated slots.

        Args:
            intent_name (str): The name of the intent to be parsed.
            intent_slots (list): A list of dictionaries representing intent slots.

        Returns:
            list: A list of dictionaries describing actions and signal-related details for execution.

        Note:
            - If no relevant VSS signals are found for the intent, an empty list is returned.
            - If no specific action or modifier is determined, default values are used.
        """
        vss_signal_data = self.map_intent_to_signal(intent_name)
        execution_list = []
        for signal_name, signal_data in vss_signal_data.items():
            action = self.determine_action(signal_data, intent_slots)
            modifier = self.determine_modifier(signal_data, intent_slots)
            value = self.determine_value(signal_data, intent_slots)

            if value != None and not self.verify_value(signal_data, value):
                value = None

            change_factor = signal_data["default_change_factor"]

            if action in ["increase", "decrease"]:
                if value and modifier == "to":
                    execution_list.append({"action": action, "signal": signal_name, "value": str(value)})
                
                elif value and modifier == "by":
                    execution_list.append({"action": action, "signal": signal_name, "factor": str(value)})
                
                elif value:
                    execution_list.append({"action": action, "signal": signal_name, "value": str(value)})

                elif signal_data["default_fallback"]:
                    execution_list.append({"action": action, "signal": signal_name, "factor": str(change_factor)})

            # if no value found set the default value
            if value == None and signal_data["default_fallback"]:
                value = signal_data["default_value"]

            if action == "set" and value != None:
                execution_list.append({"action": action, "signal": signal_name, "value": str(value)})
                    
        
        return execution_list
        

    def determine_action(self, signal_data, intent_slots):
        """
        Determines the action (e.g., set, increase, decrease) based on the intent slots
        and VSS signal data.

        Args:
            signal_data (dict): The specification data for a VSS signal.
            intent_slots (list): A list of dictionaries representing intent slots.

        Returns:
            str: The determined action or None if no action can be determined.
        """
        action_res = None
        for intent_slot in intent_slots:
           for action, action_data in signal_data["actions"].items():
                if intent_slot["name"] in action_data["intents"] and intent_slot["value"] in action_data["synonyms"]:
                    action_res = action
                    break
        
        return action_res
    

    def determine_modifier(self, signal_data, intent_slots):
        """
        Determines the modifier (e.g., 'to' or 'by') based on the intent slots
        and VSS signal data.

        Args:
            signal_data (dict): The specification data for a VSS signal.
            intent_slots (list): A list of dictionaries representing intent slots.

        Returns:
            str: The determined modifier or None if no modifier can be determined.
        """
        modifier_res = None
        for intent_slot in intent_slots:
           for _, action_data in signal_data["actions"].items():
                intent_val = intent_slot["value"]
                if "modifier_intents" in action_data and intent_slot["name"] in action_data["modifier_intents"] and ("to" in intent_val or "by" in intent_val):
                    modifier_res = "to" if "to" in intent_val else "by" if "by" in intent_val else None
                    break
        
        return modifier_res


    def determine_value(self, signal_data, intent_slots):
        """
        Determines the value associated with the intent slot, considering the data type
        and converting it to a numeric string representation if necessary.

        Args:
            signal_data (dict): The specification data for a VSS signal.
            intent_slots (list): A list of dictionaries representing intent slots.

        Returns:
            str: The determined value or None if no value can be determined.
        """
        result  = None
        for intent_slot in intent_slots:
           for value, value_data in signal_data["value_set_intents"].items():
                if intent_slot["name"] == value:
                    result = intent_slot["value"]

                    if value_data["datatype"] == "number":
                        result = words_to_number(result) # we assume our model will always return a number in words
        
        # the value should always returned as str because Kuksa expects str values
        return str(result) if result != None else None
    

    def verify_value(self, signal_data, value):
        """
        Verifies that the value is valid based on the VSS signal data.

        Args:
            signal_data (dict): The specification data for a VSS signal.
            value (str): The value to be verified.

        Returns:
            bool: True if the value is valid, False otherwise.
        """
        if value in signal_data["values"]["ignore"]:
            return False
        
        elif signal_data["values"]["ranged"] and isinstance(value, (int, float)):
            return value >= signal_data["values"]["start"] and value <= signal_data["values"]["end"]
        
        else:
            return value in signal_data["values"]["additional"]