diff options
Diffstat (limited to 'agl_service_voiceagent/utils')
| -rw-r--r-- | agl_service_voiceagent/utils/config.py | 30 | ||||
| -rw-r--r-- | agl_service_voiceagent/utils/kuksa_interface.py | 35 | ||||
| -rw-r--r-- | agl_service_voiceagent/utils/mapper.py | 10 |
3 files changed, 63 insertions, 12 deletions
diff --git a/agl_service_voiceagent/utils/config.py b/agl_service_voiceagent/utils/config.py index 7295c7f..e0b053e 100644 --- a/agl_service_voiceagent/utils/config.py +++ b/agl_service_voiceagent/utils/config.py @@ -14,10 +14,13 @@ # See the License for the specific language governing permissions and # limitations under the License. +import os +import logging import configparser config = configparser.ConfigParser() config_path = None +logger = None def set_config_path(path): """ @@ -25,14 +28,25 @@ def set_config_path(path): """ global config_path config_path = path - config.read(config_path) def load_config(): """ - Loads the config file. + Loads the config file and initializes the logger. + + Also creates logging directory if it doesn't already exist. """ if config_path is not None: + global logger config.read(config_path) + + # create the base log dir if not exists + if not os.path.exists(get_config_value('BASE_LOG_DIR')): + os.makedirs(get_config_value('BASE_LOG_DIR')) + + logging.basicConfig(filename=get_config_value('BASE_LOG_DIR')+'voiceagent_server.log', level=logging.DEBUG, format='[%(asctime)s] [%(name)s] [%(levelname)s]: (%(filename)s:%(funcName)s) %(message)s', filemode='a') + logger = logging.getLogger() + logger.info("-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-") + else: raise Exception("Config file path not provided.") @@ -52,3 +66,15 @@ def get_config_value(key, group="General"): Gets a value from the config file. """ return config.get(group, key) + +def get_logger(): + """ + Gets the initialized logger. + """ + if logger is not None: + return logger + + else: + logging.basicConfig(level=logging.DEBUG) + print("[-] Error: Failed to get logger. Logger is not initialized!") + logging.error("Failed to get logger. Logger is not initialized!") diff --git a/agl_service_voiceagent/utils/kuksa_interface.py b/agl_service_voiceagent/utils/kuksa_interface.py index 9270379..0881660 100644 --- a/agl_service_voiceagent/utils/kuksa_interface.py +++ b/agl_service_voiceagent/utils/kuksa_interface.py @@ -18,7 +18,7 @@ import time import json import threading from kuksa_client import KuksaClientThread -from agl_service_voiceagent.utils.config import get_config_value +from agl_service_voiceagent.utils.config import get_config_value, get_logger class KuksaInterface: """ @@ -55,6 +55,7 @@ class KuksaInterface: self.insecure = get_config_value("insecure", "Kuksa") self.protocol = get_config_value("protocol", "Kuksa") self.token = get_config_value("token", "Kuksa") + self.logger = get_logger() print(self.ip, self.port, self.insecure, self.protocol, self.token) @@ -102,11 +103,14 @@ class KuksaInterface: if not self.get_kuksa_status(): print("[-] Error: Connection to Kuksa server failed.") + self.logger.error("Connection to Kuksa server failed.") else: print("[+] Connection to Kuksa established.") + self.logger.info("Connection to Kuksa established.") except Exception as e: print("[-] Error: Connection to Kuksa server failed. ", str(e)) + self.logger.error(f"Connection to Kuksa server failed. {str(e)}") def authorize_kuksa_client(self): @@ -119,10 +123,13 @@ class KuksaInterface: if "error" in response: error_message = response.get("error", "Unknown error") print(f"[-] Error: Authorization failed. {error_message}") + self.logger.error(f"Authorization failed. {error_message}") else: print("[+] Kuksa client authorized successfully.") + self.logger.info("Kuksa client authorized successfully.") else: print("[-] Error: Kuksa client is not initialized. Call `connect_kuksa_client` first.") + self.logger.error("Kuksa client is not initialized. Call `connect_kuksa_client` first.") def send_values(self, path=None, value=None): @@ -137,7 +144,8 @@ class KuksaInterface: """ result = False if self.kuksa_client is None: - print("[-] Error: Kuksa client is not initialized.") + print(f"[-] Error: Failed to send value '{value}' to Kuksa. Kuksa client is not initialized.") + self.logger.error(f"Failed to send value '{value}' to Kuksa. Kuksa client is not initialized.") return if self.get_kuksa_status(): @@ -150,12 +158,15 @@ class KuksaInterface: else: error_message = response.get("error", "Unknown error") print(f"[-] Error: Failed to send value '{value}' to Kuksa. {error_message}") + self.logger.error(f"Failed to send value '{value}' to Kuksa. {error_message}") except Exception as e: - print("[-] Error: Failed to send values to Kuksa. ", str(e)) + print(f"[-] Error: Failed to send value '{value}' to Kuksa. ", str(e)) + self.logger.error(f"Failed to send value '{value}' to Kuksa. {str(e)}") else: - print("[-] Error: Connection to Kuksa failed.") + print(f"[-] Error: Failed to send value '{value}' to Kuksa. Connection to Kuksa failed.") + self.logger.error(f"Failed to send value '{value}' to Kuksa. Connection to Kuksa failed.") return result @@ -171,7 +182,8 @@ class KuksaInterface: """ result = None if self.kuksa_client is None: - print("[-] Error: Kuksa client is not initialized.") + print(f"[-] Error: Failed to get value at path '{path}' from Kuksa. Kuksa client is not initialized.") + self.logger.error(f"Failed to get value at path '{path}' from Kuksa. Kuksa client is not initialized.") return if self.get_kuksa_status(): @@ -185,13 +197,16 @@ class KuksaInterface: else: error_message = response.get("error", "Unknown error") - print(f"[-] Error: Failed to get value from Kuksa. {error_message}") + print(f"[-] Error: Failed to get value at path '{path}' from Kuksa. {error_message}") + self.logger.error(f"Failed to get value at path '{path}' from Kuksa. {error_message}") except Exception as e: - print("[-] Error: Failed to get values from Kuksa. ", str(e)) + print(f"[-] Error: Failed to get value at path '{path}' from Kuksa. ", str(e)) + self.logger.error(f"Failed to get value at path '{path}' from Kuksa. {str(e)}") else: - print("[-] Error: Connection to Kuksa failed.") + print(f"[-] Error: Failed to get value at path '{path}' from Kuksa. Connection to Kuksa failed.") + self.logger.error(f"Failed to get value at path '{path}' from Kuksa. Connection to Kuksa failed.") return result @@ -206,5 +221,7 @@ class KuksaInterface: self.kuksa_client.stop() self.kuksa_client = None print("[+] Kuksa client stopped.") + self.logger.info("Kuksa client stopped.") except Exception as e: - print("[-] Error: Failed to close Kuksa client. ", str(e))
\ No newline at end of file + print("[-] Error: Failed to close Kuksa client. ", str(e)) + self.logger.error(f"Failed to close Kuksa client. {str(e)}")
\ No newline at end of file diff --git a/agl_service_voiceagent/utils/mapper.py b/agl_service_voiceagent/utils/mapper.py index 7529645..f24f44f 100644 --- a/agl_service_voiceagent/utils/mapper.py +++ b/agl_service_voiceagent/utils/mapper.py @@ -14,7 +14,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -from agl_service_voiceagent.utils.config import get_config_value +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 @@ -33,6 +33,7 @@ class Intent2VSSMapper: 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.") @@ -49,6 +50,7 @@ class Intent2VSSMapper: # 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'] @@ -56,12 +58,14 @@ class Intent2VSSMapper: # 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 @@ -69,6 +73,7 @@ class Intent2VSSMapper: 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'] @@ -76,11 +81,13 @@ class Intent2VSSMapper: # 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'] @@ -88,6 +95,7 @@ class Intent2VSSMapper: # 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 |