| # Copyright: (c) 2025, Intel Corporation |
| # Author: Arkadiusz Cholewinski <arkadiuszx.cholewinski@intel.com> |
| |
| import csv |
| import logging |
| import queue |
| import re |
| import threading |
| import time |
| |
| import utils.UtilityFunctions as UtilityFunctions |
| from abstract.PowerMonitor import PowerMonitor |
| from stm32l562e_dk.PowerShieldConfig import PowerShieldConf |
| from stm32l562e_dk.PowerShieldData import PowerShieldData |
| from stm32l562e_dk.SerialHandler import SerialHandler |
| |
| |
| class PowerShield(PowerMonitor): |
| def __init__(self): |
| """ |
| Initializes the PowerShield. |
| """ |
| self.handler = None |
| self.dataQueue = queue.Queue() |
| self.acqComplete = False |
| self.acqStart = False |
| self.target_voltage = None |
| self.target_temperature = None |
| self.acqTimeoutThread = None |
| self.power_shield_conf = PowerShieldConf() |
| self.power_shield_data = PowerShieldData() |
| |
| def init(self): |
| """ |
| Initializes the power monitor. |
| """ |
| self.__take_control() |
| self.__set_voltage(self.power_shield_conf.target_voltage) |
| self.__set_format(self.power_shield_conf.data_format) |
| self.__set_func_mode(self.power_shield_conf.function_mode) |
| |
| def connect(self, power_device_path: str): |
| """Opens the connection using the SerialHandler.""" |
| self.handler = SerialHandler(power_device_path, 3686400) |
| self.handler.open() |
| |
| def disconnect(self): |
| """Closes the connection using the SerialHandler.""" |
| self.handler.close() |
| |
| def __send_command(self, command: str, expected_ack: str = None, ack: bool = False) -> str: |
| """ |
| Sends a command to the device, retrieves the response, |
| and optionally verifies the acknowledgment. |
| |
| :param command: The command to send. |
| :param expected_ack: The expected acknowledgment response (e.g., "ack htc"). |
| :return: The response received from the device. |
| """ |
| if not self.handler.is_open(): |
| logging.info(f"Error: Connection is not open. Cannot send command: {command}") |
| return "" |
| |
| logging.debug(f"Sending command: {command}") |
| self.handler.send_cmd(command) |
| if ack: |
| response = self.handler.receive_cmd() |
| logging.debug(f"Response: {response}") |
| |
| # Check if the response contains the expected acknowledgment |
| if expected_ack and expected_ack not in response: |
| logging.error(f"Error: Expected acknowledgment '{expected_ack}' not found.") |
| return "" |
| |
| return response |
| return 0 |
| |
| def __test_communication(self): |
| """ |
| Sends a version command to the device. |
| """ |
| if not self.handler.is_open(): |
| logging.error("Error: Connection is not open. Cannot send version command.") |
| return "" |
| command = 'version' |
| logging.info(f"Sending command: {command}") |
| self.handler.send_cmd(command) |
| response = self.handler.receive_cmd() |
| logging.info(f"Response: {response}") |
| return response |
| |
| def __reset(self): |
| """ |
| Sends the reset command ('PSRST') to the power monitor device, |
| closes the connection, waits for the reset process to complete, |
| and repeatedly attempts to reconnect until successful. |
| """ |
| command = "psrst" |
| |
| if not self.handler.is_open(): |
| logging.error("Error: Connection is not open. Cannot reset the device.") |
| return |
| |
| logging.info(f"Sending reset command: {command}") |
| self.handler.send_cmd(command) |
| |
| # Close the connection |
| self.handler.close() |
| self.handler.serial_connection = None |
| |
| time.sleep(5) |
| # Attempt to reopen the connection |
| try: |
| self.handler.open() |
| logging.info("Connection reopened after reset.") |
| except Exception as e: |
| logging.error(f"Failed to reopen connection after reset: {e}") |
| |
| def __get_voltage_level(self) -> float: |
| """ |
| Sends the 'volt get' command and returns the voltage value as a float. |
| |
| :return: The voltage level as a float, in volts (V). |
| """ |
| command = 'volt get' |
| response = self.__send_command(command, expected_ack="ack volt get", ack=True) |
| |
| # If response contains the expected acknowledgment, extract and return the voltage |
| if response: |
| parts = response.split() |
| try: |
| if len(parts) >= 5: |
| # Use regex to find a string that matches the pattern, e.g., "3292-03" |
| match = re.search(r'(\d+)-(\d+)', parts[5]) |
| if match: |
| # Extract the base (3292) and exponent (03) |
| base = match.group(1) |
| exponent = match.group(2) |
| |
| # Construct the scientific notation string (e.g., 3292e-03) |
| voltage_str = f"{base}e-{exponent}" |
| |
| # Convert the string into a float |
| voltage = float(voltage_str) |
| |
| # Return the voltage as a float |
| self.target_voltage = round(voltage, 3) |
| return self.target_voltage |
| except ValueError: |
| logging.error("Error: Could not convert temperature value.") |
| return float('nan') |
| else: |
| logging.error("Error: No response for voltage command.") |
| return float('nan') |
| |
| def __get_temperature(self, unit: str = PowerShieldConf.TemperatureUnit.CELSIUS) -> float: |
| """ |
| Sends the temperature command and returns the temperature as a float. |
| |
| :param unit: The unit to request the temperature in, either 'degc' or 'degf'. |
| :return: The temperature value as a float, in the specified unit (°C or °F). |
| """ |
| # Send the temp command with the unit |
| response = self.__send_command(f"temp {unit}", expected_ack=f"ack temp {unit}", ack=True) |
| |
| # If response contains the expected acknowledgment, extract the temperature |
| if response: |
| try: |
| # Example response format: "PowerShield > ack temp degc 28.0" |
| parts = response.split() |
| if len(parts) >= 5 and parts[5].replace('.', '', 1).isdigit(): |
| # Extract temperature and convert to float |
| self.target_temetarute = float(parts[5]) |
| logging.info(f"Temperature: {self.target_temetarute} {unit}") |
| return self.target_temetarute |
| else: |
| print("Error: Temperature value not found in response.") |
| return None |
| except ValueError: |
| logging.error("Error: Could not convert temperature value.") |
| return None |
| else: |
| logging.error("Error: No response for temp command.") |
| return None |
| |
| def __take_control(self) -> str: |
| """ |
| Sends the 'htc' command and verifies the acknowledgment. |
| |
| :return: The acknowledgment response or error message. |
| """ |
| return self.__send_command("htc", expected_ack="ack htc", ack=True) |
| |
| def __set_format(self, data_format: str = PowerShieldConf.DataFormat.ASCII_DEC): |
| """ |
| Sets the measurement data format. |
| The format can be either ASCII (decimal) or Binary (hexadecimal). |
| |
| :param data_format: The data format to set. |
| Options are 'ascii_dec' or 'bin_hexa'. |
| :return: None |
| """ |
| # Validate the input format |
| if data_format not in vars(PowerShieldConf.DataFormat).values(): |
| logging.error( |
| f"Error: Invalid format '{data_format}'. " |
| "Valid options are 'ascii_dec' or 'bin_hexa'." |
| ) |
| return |
| |
| command = f"format {data_format}" |
| response = self.__send_command(command, expected_ack=f"ack format {data_format}", ack=True) |
| |
| # If response contains the expected acknowledgment, the format was set successfully |
| if response: |
| logging.info(f"Data format set to {data_format}.") |
| else: |
| logging.error(f"Error: Failed to set data format to {data_format}.") |
| |
| def __set_frequency(self, frequency: enumerate): |
| """ |
| Sets the sampling frequency for the measurement. |
| The frequency can be any valid value from the list. |
| |
| :param frequency: The sampling frequency to set. |
| Valid options include: |
| {100k, 50k, 20k, 10k, 5k, 2k, 1k, 500, 200, 100, 50, 20, 10, 5, 2, 1}. |
| |
| :return: None |
| """ |
| # Validate the input frequency |
| if frequency not in vars(PowerShieldConf.SamplingFrequency).values(): |
| logging.error( |
| f"Error: Invalid frequency '{frequency}'." |
| "Valid options are:" |
| "100k, 50k, 20k, 10k, 5k, 2k, 1k, 500, 200, 100, 50, 20, 10, 5, 2, 1." |
| ) |
| return |
| |
| command = f"freq {frequency}" |
| response = self.__send_command(command, expected_ack=f"ack freq {frequency}", ack=True) |
| |
| if response: |
| logging.info(f"Sampling frequency set to {frequency}.") |
| else: |
| logging.error(f"Error: Failed to set sampling frequency to {frequency}.") |
| |
| def __set_acquisition_time(self, acquisition_time: str = '0'): |
| command = f"acqtime {acquisition_time}" |
| response = self.__send_command( |
| command, expected_ack=f"ack acqtime {acquisition_time}", ack=True |
| ) |
| |
| if response: |
| logging.info(f"Acquisition time set to {acquisition_time}.") |
| else: |
| logging.error(f"Error: Failed to set acquisition time to {acquisition_time}.") |
| |
| def __set_voltage(self, voltage: enumerate): |
| command = f"volt {voltage}" |
| response = self.__send_command(command, expected_ack=f"ack volt {voltage}", ack=True) |
| |
| if response: |
| logging.info(f"Voltage set to {voltage}.") |
| else: |
| logging.error(f"Error: Failed to set voltage to {voltage}.") |
| |
| def __set_func_mode(self, function_mode: str = PowerShieldConf.FunctionMode.HIGH): |
| """ |
| Sets the acquisition mode for current measurement. |
| The function_mode can be either 'optim' or 'high'. |
| |
| - 'optim': Priority on current resolution (100 nA - 10 mA) with max freq at 100 kHz. |
| - 'high': High current (30 µA - 10 mA), high frequency (50-100 kHz), high resolution. |
| |
| :param mode: The acquisition mode. Must be either 'optim' or 'high'. |
| :return: None |
| """ |
| # Validate the input format |
| if function_mode not in vars(PowerShieldConf.FunctionMode).values(): |
| logging.error( |
| f"Error: Invalid format '{function_mode}'." |
| "Valid options are 'ascii_dec' or 'bin_hexa'." |
| ) |
| return |
| |
| command = f"funcmode {function_mode}" |
| response = self.__send_command( |
| command, expected_ack=f"ack funcmode {function_mode}", ack=True |
| ) |
| |
| if response: |
| logging.info(f"Data format set to {function_mode}.") |
| else: |
| logging.error(f"Error: Failed to set data format to {function_mode}.") |
| |
| def __acq_data(self): |
| """ |
| Continuously reads data from the serial port and puts it |
| into a queue until acquisition is complete. |
| """ |
| logging.info("Started data acquisition...") |
| while True: |
| # Read the first byte |
| first_byte = self.handler.read_bytes(1) |
| if len(first_byte) < 1 or self.acqComplete: # Exit conditions |
| logging.info("Stopping data acquisition...") |
| return |
| |
| # Check if it's metadata |
| if first_byte == b'\xf0': # Metadata marker |
| second_byte = self.handler.read_bytes(1) |
| # Handle metadata types |
| metadata_type = second_byte[0] |
| self.__handle_metadata(metadata_type) |
| else: |
| # Not metadata, treat as data |
| if self.acqStart: |
| second_byte = self.handler.read_bytes(1) |
| data = [] |
| data.append(first_byte) |
| if len(second_byte) < 1 or self.acqComplete: |
| logging.info("Stopping data acquisition...") |
| return |
| data.append(second_byte) |
| amps = UtilityFunctions.convert_to_amps( |
| UtilityFunctions.bytes_to_twobyte_values(data) |
| ) |
| self.dataQueue.put([amps]) |
| |
| def __handle_metadata(self, metadata_type): |
| if metadata_type == 0xF1: |
| logging.info("Received Metadata: ASCII error message.") |
| # self.handle_metadata_error() |
| elif metadata_type == 0xF2: |
| logging.info("Received Metadata: ASCII information message.") |
| # self.handle_metadata_info() |
| elif metadata_type == 0xF3: |
| logging.info("Received Metadata: Timestamp message.") |
| self.__handle_metadata_timestamp() |
| self.acqStart = True |
| elif metadata_type == 0xF4: |
| logging.info("Received Metadata: End of acquisition tag.") |
| self.__handle_metadata_end() |
| self.__handle_summary() |
| elif metadata_type == 0xF5: |
| logging.info("Received Metadata: Overcurrent detected.") |
| # self.handle_metadata_overcurrent() |
| else: |
| logging.error(f"Error: Unknown Metadata Type: {metadata_type:#04x}") |
| |
| def __handle_summary(self): |
| s = "" |
| while True: |
| # Read the first byte |
| x = self.handler.read_bytes(1) |
| if len(x) < 1 or x == 0xF0: |
| self.acqComplete = True |
| return s.replace("\0", "").strip().replace("\r", "").replace("\n\n\n", "\n") |
| s += str(x, encoding='ascii', errors='ignore') |
| |
| def __handle_metadata_end(self): |
| """ |
| Handle metadata end of acquisition message. |
| """ |
| # Read the next 2 bytes |
| metadata_bytes = self.handler.read_bytes(2) |
| if len(metadata_bytes) < 2: |
| logging.error("Error: Incomplete end of acquisition metadata reveived.") |
| return |
| # Check for end tags (last 2 bytes) |
| end_tag_1 = metadata_bytes[0] |
| end_tag_2 = metadata_bytes[1] |
| if end_tag_1 != 0xFF or end_tag_2 != 0xFF: |
| logging.error("Error: Invalid metadata end tags received.") |
| return |
| |
| def __handle_metadata_timestamp(self): |
| """ |
| Handle metadata timestamp message. Parses and displays the timestamp and buffer load. |
| """ |
| # Read the next 7 bytes (timestamp + buffer load + end tags) |
| metadata_bytes = self.handler.read_bytes(7) |
| if len(metadata_bytes) < 7: |
| logging.error("Error: Incomplete timestamp metadata received.") |
| return |
| |
| # Parse the timestamp (4 bytes, big-endian) |
| timestamp_ms = int.from_bytes(metadata_bytes[0:4], byteorder='big', signed=False) |
| # Parse the buffer Tx load value (1 byte) |
| buffer_load = metadata_bytes[4] |
| # Check for end tags (last 2 bytes) |
| end_tag_1 = metadata_bytes[5] |
| end_tag_2 = metadata_bytes[6] |
| if end_tag_1 != 0xFF or end_tag_2 != 0xFF: |
| logging.error("Error: Invalid metadata end tags received.") |
| return |
| |
| # Display parsed values |
| logging.info(f"Metadata Timestamp: {timestamp_ms} ms") |
| logging.info(f"Buffer Tx Load: {buffer_load}%") |
| |
| def __start_measurement(self): |
| """ |
| Starts the measurement by sending the 'start' command. Once the measurement starts, |
| data can be received continuously until the 'stop' command is sent. |
| |
| :return: None |
| """ |
| command = "start" |
| self.acqComplete = False |
| self.__send_command(command) |
| |
| raw_to_file_Thread = threading.Thread( |
| target=self.__raw_to_file, args=(self.power_shield_conf.output_file,) |
| ) |
| raw_to_file_Thread.start() |
| logging.info("Measurement started. Receiving data...") |
| self.__acq_data() |
| raw_to_file_Thread.join() |
| |
| def __raw_to_file(self, outputFilePath: str): |
| # Open a CSV file for writing |
| with open(outputFilePath, 'w', newline='') as outputFile: |
| writer = csv.writer(outputFile) |
| while True: |
| if self.dataQueue.empty() and bool(self.acqComplete): |
| outputFile.close() |
| break |
| if not self.dataQueue.empty(): |
| data = self.dataQueue.get() |
| writer.writerow(data) |
| outputFile.flush() |
| else: |
| time.sleep(0.1) |
| |
| def measure(self, time: int, freq: str = None, reset: bool = False): |
| self.power_shield_conf.acquisition_time = time |
| _time, self.power_shield_conf.acquisition_time_unit = UtilityFunctions.convert_acq_time( |
| time |
| ) |
| |
| if reset: |
| self.__reset() |
| self.__take_control() |
| self.__set_format(self.power_shield_conf.data_format) |
| if freq is not None: |
| self.__set_frequency(freq) |
| else: |
| self.__set_frequency(self.power_shield_conf.sampling_frequency) |
| self.__set_acquisition_time( |
| UtilityFunctions.convert_to_scientific_notation( |
| time=_time, unit=self.power_shield_conf.acquisition_time_unit |
| ) |
| ) |
| self.__start_measurement() |
| |
| def get_data(self, unit: str = PowerShieldConf.MeasureUnit.RAW_DATA): |
| if self.acqComplete: |
| # Open the CSV file |
| with open(self.power_shield_conf.output_file) as file: |
| csv_reader = csv.reader(file) |
| for row in csv_reader: |
| self.power_shield_data.data.append(row[0]) |
| if unit == PowerShieldConf.MeasureUnit.CURRENT_RMS: |
| self.power_shield_data.current_RMS = UtilityFunctions.calculate_rms( |
| self.power_shield_data.data |
| ) |
| return self.power_shield_data.current_RMS |
| elif unit == PowerShieldConf.MeasureUnit.POWER: |
| _delta_time = self.power_shield_conf.acquisition_time |
| self.power_shield_data.power = 0 |
| for data in self.power_shield_data.data: |
| self.power_shield_data.power += float( |
| float(data) * float(_delta_time) * float(self.target_voltage) |
| ) |
| return self.power_shield_data.power |
| elif unit == PowerShieldConf.MeasureUnit.RAW_DATA: |
| return self.power_shield_data.data |
| else: |
| logging.error("Error: Unknown unit of requested data") |
| else: |
| logging.info("Acquisition not complete.") |
| return None |
