| # Copyright (c) 2023 Nordic Semiconductor ASA |
| # |
| # SPDX-License-Identifier: Apache-2.0 |
| |
| from __future__ import annotations |
| |
| import abc |
| import logging |
| import os |
| import queue |
| import re |
| import shutil |
| import threading |
| import time |
| from datetime import datetime |
| from pathlib import Path |
| |
| from twister_harness.exceptions import ( |
| TwisterHarnessException, |
| TwisterHarnessTimeoutException, |
| ) |
| from twister_harness.twister_harness_config import DeviceConfig |
| |
| logger = logging.getLogger(__name__) |
| |
| |
| class DeviceAdapter(abc.ABC): |
| """ |
| This class defines a common interface for all device types (hardware, |
| simulator, QEMU) used in tests to gathering device output and send data to |
| it. |
| """ |
| |
| def __init__(self, device_config: DeviceConfig) -> None: |
| """ |
| :param device_config: device configuration |
| """ |
| self.device_config: DeviceConfig = device_config |
| self.base_timeout: float = device_config.base_timeout |
| self._device_read_queue: queue.Queue = queue.Queue() |
| self._reader_thread: threading.Thread | None = None |
| self._device_run: threading.Event = threading.Event() |
| self._device_connected: threading.Event = threading.Event() |
| self.command: list[str] = [] |
| self._west: str | None = None |
| |
| self.handler_log_path: Path = device_config.build_dir / 'handler.log' |
| self._log_files: list[Path] = [self.handler_log_path] |
| |
| def __repr__(self) -> str: |
| return f'{self.__class__.__name__}()' |
| |
| @property |
| def env(self) -> dict[str, str]: |
| env = os.environ.copy() |
| return env |
| |
| def launch(self) -> None: |
| """ |
| Start by closing previously running application (no effect if not |
| needed). Then, flash and run test application. Finally, start an |
| internal reader thread capturing an output from a device. |
| """ |
| self.close() |
| self._clear_internal_resources() |
| |
| if not self.command: |
| self.generate_command() |
| |
| if self.device_config.type != 'hardware': |
| self._flash_and_run() |
| |
| self._device_run.set() |
| self._start_reader_thread() |
| self.connect() |
| |
| if self.device_config.type == 'hardware': |
| # On hardware, flash after connecting to COM port, otherwise some messages |
| # from target can be lost. |
| self._flash_and_run() |
| |
| def close(self) -> None: |
| """Disconnect, close device and close reader thread.""" |
| if not self._device_run.is_set(): |
| # device already closed |
| return |
| self.disconnect() |
| self._close_device() |
| self._device_run.clear() |
| self._join_reader_thread() |
| |
| def connect(self) -> None: |
| """Connect to device - allow for output gathering.""" |
| if self.is_device_connected(): |
| logger.debug('Device already connected') |
| return |
| if not self.is_device_running(): |
| msg = 'Cannot connect to not working device' |
| logger.error(msg) |
| raise TwisterHarnessException(msg) |
| self._connect_device() |
| self._device_connected.set() |
| |
| def disconnect(self) -> None: |
| """Disconnect device - block output gathering.""" |
| if not self.is_device_connected(): |
| logger.debug("Device already disconnected") |
| return |
| self._disconnect_device() |
| self._device_connected.clear() |
| |
| def readline(self, timeout: float | None = None, print_output: bool = True) -> str: |
| """ |
| Read line from device output. If timeout is not provided, then use |
| base_timeout. |
| """ |
| timeout = timeout or self.base_timeout |
| if self.is_device_connected() or not self._device_read_queue.empty(): |
| data = self._read_from_queue(timeout) |
| else: |
| msg = 'No connection to the device and no more data to read.' |
| logger.error(msg) |
| raise TwisterHarnessException('No connection to the device and no more data to read.') |
| if print_output: |
| logger.debug('#: %s', data) |
| return data |
| |
| def readlines_until( |
| self, |
| regex: str | None = None, |
| num_of_lines: int | None = None, |
| timeout: float | None = None, |
| print_output: bool = True, |
| ) -> list[str]: |
| """ |
| Read available output lines produced by device from internal buffer |
| until following conditions: |
| |
| 1. If regex is provided - read until regex regex is found in read |
| line (or until timeout). |
| 2. If num_of_lines is provided - read until number of read lines is |
| equal to num_of_lines (or until timeout). |
| 3. If none of above is provided - return immediately lines collected so |
| far in internal buffer. |
| |
| If timeout is not provided, then use base_timeout. |
| """ |
| timeout = timeout or self.base_timeout |
| if regex: |
| regex_compiled = re.compile(regex) |
| lines: list[str] = [] |
| if regex or num_of_lines: |
| timeout_time: float = time.time() + timeout |
| while time.time() < timeout_time: |
| try: |
| line = self.readline(0.1, print_output) |
| except TwisterHarnessTimeoutException: |
| continue |
| lines.append(line) |
| if regex and regex_compiled.search(line): |
| break |
| if num_of_lines and len(lines) == num_of_lines: |
| break |
| else: |
| msg = 'Read from device timeout occurred' |
| logger.error(msg) |
| raise TwisterHarnessTimeoutException(msg) |
| else: |
| lines = self.readlines(print_output) |
| return lines |
| |
| def readlines(self, print_output: bool = True) -> list[str]: |
| """ |
| Read all available output lines produced by device from internal buffer. |
| """ |
| lines: list[str] = [] |
| while not self._device_read_queue.empty(): |
| line = self.readline(0.1, print_output) |
| lines.append(line) |
| return lines |
| |
| def clear_buffer(self) -> None: |
| """ |
| Remove all available output produced by device from internal buffer |
| (queue). |
| """ |
| self.readlines(print_output=False) |
| |
| def write(self, data: bytes) -> None: |
| """Write data bytes to device.""" |
| if not self.is_device_connected(): |
| msg = 'No connection to the device' |
| logger.error(msg) |
| raise TwisterHarnessException(msg) |
| self._write_to_device(data) |
| |
| def initialize_log_files(self, test_name: str = '') -> None: |
| """ |
| Initialize log files (e.g. handler.log) by adding header with |
| information about performed test and current time. |
| """ |
| for log_file_path in self._log_files: |
| with open(log_file_path, 'a+') as log_file: |
| log_file.write(f'\n==== Test {test_name} started at {datetime.now()} ====\n') |
| |
| def _start_reader_thread(self) -> None: |
| self._reader_thread = threading.Thread(target=self._handle_device_output, daemon=True) |
| self._reader_thread.start() |
| |
| def _handle_device_output(self) -> None: |
| """ |
| This method is dedicated to run it in separate thread to read output |
| from device and put them into internal queue and save to log file. |
| """ |
| with open(self.handler_log_path, 'a+') as log_file: |
| while self.is_device_running(): |
| if self.is_device_connected(): |
| output = self._read_device_output().decode(errors='replace').strip() |
| if output: |
| self._device_read_queue.put(output) |
| log_file.write(f'{output}\n') |
| log_file.flush() |
| else: |
| # ignore output from device |
| self._flush_device_output() |
| time.sleep(0.1) |
| |
| def _read_from_queue(self, timeout: float) -> str: |
| """Read data from internal queue""" |
| try: |
| data: str | object = self._device_read_queue.get(timeout=timeout) |
| except queue.Empty as exc: |
| raise TwisterHarnessTimeoutException(f'Read from device timeout occurred ({timeout}s)') from exc |
| return data |
| |
| def _join_reader_thread(self) -> None: |
| if self._reader_thread is not None: |
| self._reader_thread.join(self.base_timeout) |
| self._reader_thread = None |
| |
| def _clear_internal_resources(self) -> None: |
| self._reader_thread = None |
| self._device_read_queue = queue.Queue() |
| self._device_run.clear() |
| self._device_connected.clear() |
| |
| @property |
| def west(self) -> str: |
| """ |
| Return a path to west or if not found - raise an error. Once found |
| west path is stored as internal property to save time of looking for it |
| in the next time. |
| """ |
| if self._west is None: |
| self._west = shutil.which('west') |
| if self._west is None: |
| msg = 'west not found' |
| logger.error(msg) |
| raise TwisterHarnessException(msg) |
| return self._west |
| |
| @abc.abstractmethod |
| def generate_command(self) -> None: |
| """ |
| Generate and set command which will be used during flashing or running |
| device. |
| """ |
| |
| @abc.abstractmethod |
| def _flash_and_run(self) -> None: |
| """Flash and run application on a device.""" |
| |
| @abc.abstractmethod |
| def _connect_device(self) -> None: |
| """Connect with the device (e.g. via serial port).""" |
| |
| @abc.abstractmethod |
| def _disconnect_device(self) -> None: |
| """Disconnect from the device (e.g. from serial port).""" |
| |
| @abc.abstractmethod |
| def _close_device(self) -> None: |
| """Stop application""" |
| |
| @abc.abstractmethod |
| def _read_device_output(self) -> bytes: |
| """ |
| Read device output directly through serial, subprocess, FIFO, etc. |
| Even if device is not connected, this method has to return something |
| (e.g. empty bytes string). This assumption is made to maintain |
| compatibility between various adapters and their reading technique. |
| """ |
| |
| @abc.abstractmethod |
| def _write_to_device(self, data: bytes) -> None: |
| """Write to device directly through serial, subprocess, FIFO, etc.""" |
| |
| @abc.abstractmethod |
| def _flush_device_output(self) -> None: |
| """Flush device connection (serial, subprocess output, FIFO, etc.)""" |
| |
| @abc.abstractmethod |
| def is_device_running(self) -> bool: |
| """Return true if application is running on device.""" |
| |
| @abc.abstractmethod |
| def is_device_connected(self) -> bool: |
| """Return true if device is connected.""" |
