import contextlib
import logging
import os
import re
from dataclasses import dataclass
from logging import Formatter
from pathlib import Path
from typing import overload
from esptool.loader import ESPLoader
import esptest.common.compat_typing as t
from ...common.data_monitor import DataMonitor
from ...common.timestamp import timestamp_slug
from ...interface.dut import DutInterface
from ...interface.port import PortInterface
from ...logger import get_logger
from ...utility.parse_bin_path import ParseBinPath, get_baud_from_bin_path
from ..port.base_port import BasePort, RawPort
from ..port.data_monitor_mixin import DataMonitorMixin
from ..port.serial_port import SerialPort
logger = get_logger('dut')
DEFAULT_SERIAL_CONFIGS = {'timeout': 0.005}
[docs]
@dataclass
class DutConfig:
name: str = '' # default = dut name / port name
device: str = '' # log serial device, eg: '/dev/ttyUSB0', 'COM3', etc.
baudrate: int = 0 # console baudrate, 0: get from bin path or 115200
baudrate_from_bin_path: bool = True # always get baudrate from bin path if bin_path is set
serial_configs: t.Optional[t.Dict[str, t.Any]] = None # serial configs, eg: {'bytesize': 8, 'timeout': 0.1}
# capabilities
support_esptool: bool = False # esp port or serial port
esptool_stub: bool = True # Try to get stub from bin_path if bin_path was set
esptool_chip: str = 'auto' # Try to get chip from bin_path if bin_path was set
create_redirect_thread: bool = True # create redirect thread or not
# create dut from open serial/raw_port/others
opened_port: t.Any = None
# download bin
bin_path: t.Union[str, Path] = ''
use_esptool: str = '' # For FPGA, use specific esptool
download_device: str = '' # default = base port device
download_baudrate: int = 0 # default ESPBAUD
# dut (uart) log
log_path: t.Union[str, Path] = '' # default
log_file: str = '' # default: auto-generate log_path + <name>[_<second>]_<timestamp>.log
rx_log_formater: t.Optional[Formatter] = None
tx_log_formater: t.Optional[Formatter] = None
logger: t.Optional[logging.Logger] = None
# callback (data, dut_name)
tx_log_callback: t.Optional[t.Callable[[str, bytes, str], None]] = None
rx_log_callback: t.Optional[t.Callable[[str, bytes, str], None]] = None
# monitor
monitors: t.Optional[object] = None
# checkers
pexpect_timeout: float = 30 # default pexpect timeout]
# extra / customer args
kwargs: t.Optional[t.Dict[str, t.Any]] = None
def __post_init__(self) -> None:
self._auto_gen_name()
if not self.log_file:
_log_path = self.log_path or './dut_logs'
_file_name = f'{self.name}_{timestamp_slug()}.log'.replace(':', '-')
self.log_file = str(Path(_log_path) / _file_name)
# serial configs
_serial_configs = DEFAULT_SERIAL_CONFIGS.copy()
if self.serial_configs:
_serial_configs.update(self.serial_configs)
self.serial_configs = _serial_configs
# bin_path and get variables from bin path
if self.bin_path:
self.bin_path = Path(self.bin_path).expanduser().resolve()
parsed_bin = ParseBinPath(self.bin_path)
self.esptool_stub = parsed_bin.stub
self.esptool_chip = parsed_bin.chip
if self.baudrate_from_bin_path:
# baudrate from bin path is much reliable than the specified one for uart0
self.baudrate = get_baud_from_bin_path(self.bin_path) or self.baudrate or 115200
if not self.baudrate:
self.baudrate = 115200 # set default baudrate to 115200
# download device
if not self.download_device:
self.download_device = self.device
def _auto_gen_name(self) -> None:
if self.opened_port:
if isinstance(self.opened_port, SerialPort):
self.device = self.opened_port.raw_port.device
self.name = self.opened_port.name
elif isinstance(self.opened_port, BasePort):
self.name = self.opened_port.name
if not self.name:
if self.device:
self.name = os.path.basename(self.device)
assert self.name, 'DutConfig "name" must be set'
@property
def serial_read_timeout(self) -> float:
assert isinstance(self.serial_configs, dict)
assert 'timeout' in self.serial_configs
return float(self.serial_configs['timeout']) # type: ignore
[docs]
class VariablesMixin:
def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
self._variables: t.Dict[str, t.Any] = {}
self._dynamic_variables: t.Dict[str, t.Any] = {}
super().__init__(*args, **kwargs)
[docs]
def get_variable_by_name(self, name: str, default: t.Any = None) -> t.Any:
if name in self._dynamic_variables:
return self._dynamic_variables[name]
if name in self._variables:
return self._variables[name]
return default
[docs]
def add_variable(self, name: str, value: t.Any) -> None:
self._variables[name] = value
[docs]
def remove_variable(self, name: str) -> None:
if name in self._variables:
self._variables.pop(name)
[docs]
def add_dynamic_variable(self, name: str, value: t.Any) -> None:
self._dynamic_variables[name] = value
[docs]
def remove_dynamic_variable(self, name: str) -> None:
if name in self._dynamic_variables:
self._dynamic_variables.pop(name)
class _DutBase(DutInterface):
def __init__(self, *args: t.Any, **_kwargs: t.Any) -> None: # pylint: disable=unused-argument
# kwargs are kept for BasePort creation and should not be forwarded
# to the end of MRO where object.__init__ rejects extra arguments.
super().__init__()
[docs]
class DutBase(VariablesMixin, DataMonitorMixin, _DutBase): # pylint: disable=too-many-public-methods
"""A base Dut class"""
BASE_PORT_PROXY_METHODS = list(PortInterface.__abstractmethods__)
def __init__(self, *, dut_config: DutConfig, **kwargs: t.Any) -> None:
# pass extra parameters to base port
self._kwargs = kwargs
if dut_config.monitors:
self._kwargs['monitors'] = dut_config.monitors
if dut_config.rx_log_callback:
self._kwargs['rx_log_callback'] = dut_config.rx_log_callback
# kwargs are kept for BasePort creation and should not be forwarded
# to the end of MRO where object.__init__ rejects extra arguments.
super().__init__(**kwargs)
# args and kwargs may be used by mixins
self._dut_config: DutConfig = dut_config
self._raw_port: t.Optional[RawPort] = None
self._base_port_proxy: t.Optional[BasePort] = None
self._dut_logger: logging.Logger = self._create_dut_logger()
# update variables / fields, open ports, logging
self._post_init()
# flash thread
self.setup_dut()
# redirect thread
self._start()
def _create_dut_logger(self) -> logging.Logger:
return logger.getChild(self.name)
def _post_init(self) -> None:
"""Update variables"""
pass # pylint: disable=unnecessary-pass
[docs]
def setup_dut(self) -> None:
"""Flash / hard reset"""
pass # pylint: disable=unnecessary-pass
def _start(self) -> None:
pass
@property
def dut_config(self) -> DutConfig:
return self._dut_config
@property
def dut_logger(self) -> logging.Logger:
return self._dut_logger
@property
def target(self) -> str:
# child class should implement this method
return 'unknown'
@property
def esp(self) -> ESPLoader:
"""Not all Dut support this method"""
raise NotImplementedError()
[docs]
def close(self) -> None:
pass
def __enter__(self) -> 't.Self':
return self
def __exit__(self, exc_type, exc_value, trace) -> None: # type: ignore
self.close()
# bin path related methods
@property
def bin_path(self) -> t.Union[str, Path]:
return self.dut_config.bin_path
@property
def sdkconfig(self) -> t.Dict[str, t.Any]:
if not self.bin_path:
raise FileNotFoundError('Can not get sdkconfig, bin_path is not set.')
return ParseBinPath(self.bin_path).sdkconfig
# esptool related methods
[docs]
def hard_reset(self) -> None:
raise NotImplementedError()
[docs]
def flash(self, erase_nvs: bool = True) -> None:
raise NotImplementedError()
[docs]
def flash_partition(self, part: t.Union[int, str], bin_file: str = '') -> None:
raise NotImplementedError()
[docs]
def dump_flash(self, part: t.Union[int, str], bin_file: str, size: int = 0) -> None:
raise NotImplementedError()
# port base methods, use the proxy method if possible, otherwise, child class should implement them
# def __getattribute__(self, name: str) -> t.Any:
# if object.__getattribute__(self, '_base_port_proxy'):
# if name in object.__getattribute__(self, 'BASE_PORT_PROXY_METHODS'):
# return getattr(self._base_port_proxy, name)
# return object.__getattribute__(self, name)
@property
def raw_port(self) -> t.Any:
if self._base_port_proxy:
return self._base_port_proxy.raw_port
raise NotImplementedError()
@property
def log_file(self) -> t.Any:
if self._base_port_proxy:
return self._base_port_proxy.log_file
return None
@property
def rx_log_callback(self) -> t.Optional[t.Callable[[str, bytes], None]]:
"""Get Current dut log file."""
if self._base_port_proxy:
return self._base_port_proxy.rx_log_callback
return t.cast(t.Optional[t.Callable[[str, bytes], None]], self._kwargs.get('rx_log_callback', None))
[docs]
def set_rx_log_callback(self, new_callback: t.Optional[t.Callable[[str, bytes], None]]) -> None:
self._kwargs['rx_log_callback'] = new_callback
if self._base_port_proxy:
self._base_port_proxy.set_rx_log_callback(new_callback)
@property
def monitors(self) -> t.List[DataMonitor]:
if self._base_port_proxy:
return self._base_port_proxy.monitors
return t.cast(t.List[DataMonitor], self._kwargs.setdefault('monitors', []))
@monitors.setter
def monitors(self, new_monitors: t.List[DataMonitor]) -> None:
synced_monitors = list(new_monitors)
self._kwargs['monitors'] = synced_monitors
if self._base_port_proxy:
self._base_port_proxy.monitors = synced_monitors
@property
def name(self) -> t.Any:
return self.dut_config.name
@name.setter
def name(self, value: str) -> None:
raise NotImplementedError()
[docs]
def write(self, data: t.AnyStr) -> None:
if self._base_port_proxy:
return self._base_port_proxy.write(data)
raise NotImplementedError()
[docs]
def write_line(self, data: t.AnyStr, end: str = '\n') -> None:
if self._base_port_proxy:
return self._base_port_proxy.write_line(data, end)
raise NotImplementedError()
@overload
def expect(self, pattern: str, timeout: float = 30) -> None: ...
@overload
def expect(self, pattern: bytes, timeout: float = 30) -> None: ...
@overload
def expect(self, pattern: 're.Pattern[str]', timeout: float = 30) -> 're.Match[str]': ...
@overload
def expect(self, pattern: 're.Pattern[bytes]', timeout: float = 30) -> 're.Match[bytes]': ...
[docs]
def expect(self, pattern, timeout=30): # type: ignore
if self._base_port_proxy:
return self._base_port_proxy.expect(pattern, timeout)
raise NotImplementedError()
@property
def data_cache(self) -> str:
return self.read_all_data(flush=False)
[docs]
def flush_data(self) -> str:
return self.read_all_data(flush=True)
[docs]
def read_all_data(self, flush: bool = True) -> str:
if self._base_port_proxy:
return self._base_port_proxy.read_all_data(flush)
raise NotImplementedError()
[docs]
def read_all_bytes(self, flush: bool = False) -> bytes:
if self._base_port_proxy:
return self._base_port_proxy.read_all_bytes(flush)
raise NotImplementedError()
[docs]
def start_redirect_thread(self) -> None:
"""Start a new thread to read data from port and save to data cache."""
if self._base_port_proxy:
return self._base_port_proxy.start_redirect_thread()
raise NotImplementedError()
[docs]
def stop_redirect_thread(self) -> bool:
"""Stop the redirect thread and pexpect process."""
if self._base_port_proxy:
return self._base_port_proxy.stop_redirect_thread()
raise NotImplementedError()
[docs]
@contextlib.contextmanager
def disable_redirect_thread(self) -> t.Generator[None, None, None]:
stopped = self.stop_redirect_thread()
yield
if stopped:
self.start_redirect_thread()