zephyr/tests/drivers/can/host/pytest/can_shell.py

# Copyright (c) 2024 Vestas Wind Systems A/S
#
# SPDX-License-Identifier: Apache-2.0

"""
Zephyr CAN shell module support for providing a python-can bus interface for testing.
"""

import re
import logging
from typing import Optional, Tuple

from can import BusABC, CanProtocol, Message
from can.exceptions import CanInitializationError, CanOperationError
from can.typechecking import CanFilters

from twister_harness import DeviceAdapter, Shell

logger = logging.getLogger(__name__)

class CanShellBus(BusABC): # pylint: disable=abstract-method
    """
    A CAN interface using the Zephyr CAN shell module.
    """

    def __init__(self, dut: DeviceAdapter, shell: Shell, channel: str,
                 can_filters: Optional[CanFilters] = None, **kwargs) -> None:
        self._dut = dut
        self._shell = shell
        self._device = channel
        self._is_filtered = False
        self._filter_ids = []

        self.channel_info = f'Zephyr CAN shell, device "{self._device}"'

        mode = 'normal'
        if 'fd' in self._get_capabilities():
            self._can_protocol = CanProtocol.CAN_FD
            mode += ' fd'
        else:
            self._can_protocol = CanProtocol.CAN_20

        self._set_mode(mode)
        self._start()

        super().__init__(channel=channel, can_filters=can_filters, **kwargs)

    def _retval(self):
        """Get return value of last shell command."""
        return int(self._shell.get_filtered_output(self._shell.exec_command('retval'))[0])

    def _get_capabilities(self) -> list[str]:
        cmd = f'can show {self._device}'

        lines = self._shell.get_filtered_output(self._shell.exec_command(cmd))
        regex_compiled = re.compile(r'capabilities:\s+(?P<caps>.*)')
        for line in lines:
            m = regex_compiled.match(line)
            if m:
                return m.group('caps').split()

        raise CanOperationError('capabilities not found')

    def _set_mode(self, mode: str) -> None:
        self._shell.exec_command(f'can mode {self._device} {mode}')
        retval = self._retval()
        if retval != 0:
            raise CanOperationError(f'failed to set mode "{mode}" (err {retval})')

    def _start(self):
        self._shell.exec_command(f'can start {self._device}')
        retval = self._retval()
        if retval != 0:
            raise CanInitializationError(f'failed to start (err {retval})')

    def _stop(self):
        self._shell.exec_command(f'can stop {self._device}')

    def send(self, msg: Message, timeout: Optional[float] = None) -> None:
        logger.debug('sending: %s', msg)

        cmd = f'can send {self._device}'
        cmd += ' -e' if msg.is_extended_id else ''
        cmd += ' -r' if msg.is_remote_frame else ''
        cmd += ' -f' if msg.is_fd else ''
        cmd += ' -b' if msg.bitrate_switch else ''

        if msg.is_extended_id:
            cmd += f' {msg.arbitration_id:08x}'
        else:
            cmd += f' {msg.arbitration_id:03x}'

        if msg.data:
            cmd += ' ' + msg.data.hex(' ', 1)

        lines = self._shell.exec_command(cmd)
        regex_compiled = re.compile(r'enqueuing\s+CAN\s+frame\s+#(?P<id>\d+)')
        frame_num = None
        for line in lines:
            m = regex_compiled.match(line)
            if m:
                frame_num = m.group('id')
                break

        if frame_num is None:
            raise CanOperationError('frame not enqueued')

        tx_regex = r'CAN\s+frame\s+#' + frame_num + r'\s+successfully\s+sent'
        self._dut.readlines_until(regex=tx_regex, timeout=timeout)

    def _add_filter(self, can_id: int, can_mask: int, extended: bool) -> None:
        """Add RX filter."""
        cmd = f'can filter add {self._device}'
        cmd += ' -e' if extended else ''

        if extended:
            cmd += f' {can_id:08x}'
            cmd += f' {can_mask:08x}'
        else:
            cmd += f' {can_id:03x}'
            cmd += f' {can_mask:03x}'

        lines = self._shell.exec_command(cmd)
        regex_compiled = re.compile(r'filter\s+ID:\s+(?P<id>\d+)')
        for line in lines:
            m = regex_compiled.match(line)
            if m:
                filter_id = int(m.group('id'))
                self._filter_ids.append(filter_id)
                return

        raise CanOperationError('filter_id not found')

    def _remove_filter(self, filter_id: int) -> None:
        """Remove RX filter."""
        if filter_id in self._filter_ids:
            self._filter_ids.remove(filter_id)

        self._shell.exec_command(f'can filter remove {self._device} {filter_id}')
        retval = self._retval()
        if retval != 0:
            raise CanOperationError(f'failed to remove filter ID {filter_id} (err {retval})')

    def _remove_all_filters(self) -> None:
        """Remove all RX filters."""
        for filter_id in self._filter_ids[:]:
            self._remove_filter(filter_id)

    def _apply_filters(self, filters: Optional[CanFilters]) -> None:
        self._remove_all_filters()

        if filters:
            self._is_filtered = True
        else:
            # Accept all frames if no hardware filters provided
            filters = [
                {'can_id': 0x0, 'can_mask': 0x0},
                {'can_id': 0x0, 'can_mask': 0x0, 'extended': True}
            ]
            self._is_filtered = False

        for can_filter in filters:
            can_id = can_filter['can_id']
            can_mask = can_filter['can_mask']
            extended = can_filter['extended'] if 'extended' in can_filter else False
            self._add_filter(can_id, can_mask, extended)

    def _recv_internal(self, timeout: Optional[float]) -> Tuple[Optional[Message], bool]:
        frame_regex = r'.*' + re.escape(self._device) + \
            r'\s+(?P<brs>\S)(?P<esi>\S)\s+(?P<can_id>\d+)\s+\[(?P<dlc>\d+)\]\s*(?P<data>[a-z0-9 ]*)'
        lines = self._dut.readlines_until(regex=frame_regex, timeout=timeout)
        msg = None

        regex_compiled = re.compile(frame_regex)
        for line in lines:
            m = regex_compiled.match(line)
            if m:
                can_id = int(m.group('can_id'), 16)
                ext = len(m.group('can_id')) == 8
                dlc = int(m.group('dlc'))
                fd = len(m.group('dlc')) == 2
                brs = m.group('brs') == 'B'
                esi = m.group('esi') == 'P'
                data = bytearray.fromhex(m.group('data'))
                msg = Message(arbitration_id=can_id,is_extended_id=ext,
                              data=data, dlc=dlc,
                              is_fd=fd, bitrate_switch=brs, error_state_indicator=esi,
                              channel=self._device, check=True)
                logger.debug('received: %s', msg)

        return msg, self._is_filtered

    def shutdown(self) -> None:
        if not self._is_shutdown:
            super().shutdown()
            self._stop()
            self._remove_all_filters()