-
-
visibility Watch View
-
Monitor and interact with variables in real-time.
-
-
-
show_chart Scope View
-
Visualize variable data over time with oscilloscope-style charts.
-
-
-
dashboard Dashboard
-
Create custom layouts with interactive widgets and gauges.
+
+
upload_file Export Variables
+
Use the export icon in the header to save the variables currently used in Watch View, Scope View, and Dashboard as YML or PKL.
+
+
+
Available Views
+
+
visibility Watch View
+
Monitor and interact with variables in real-time.
+
+
+
show_chart Scope View
+
Visualize variable data over time with oscilloscope-style charts.
+
+
+
dashboard Dashboard
+
Create custom layouts with interactive widgets and gauges.
+
+
+
code Scripting
+
Execute Python scripts with access to x2cscope.
+
+
+
Tips
+
+ - qr_code_2 Click QR code icons to generate scannable links for mobile access
+ - open_in_new Click open icons to launch views in separate windows
+ - help_outline Click the help icon in the header to toggle this card
+ - upload_file Click the export icon in the header to download variables used by the active views
+
-
-
code Scripting
-
Execute Python scripts with access to x2cscope.
+
+
+
+
Software Versions
+
+
+
Application:
+
+ - pyX2Cscope: {{ version }}
+
+
Web Interface Dependencies:
+
+ - Flask: Loading...
+ - Flask-SocketIO: Loading...
+ - mchplnet: Loading...
+ - python-can: Loading...
+
+
Core Dependencies:
+
+ - numpy: Loading...
+ - matplotlib: Loading...
+ - pyserial: Loading...
+ - pyelftools: Loading...
+
+
+
-
Tips
-
- - qr_code_2 Click QR code icons to generate scannable links for mobile access
- - open_in_new Click open icons to launch views in separate windows
- - help_outline Click the help icon in the header to toggle this card
-
+
diff --git a/pyx2cscope/gui/web/views/scope_view.py b/pyx2cscope/gui/web/views/scope_view.py
index 651fe648..6bba8b71 100644
--- a/pyx2cscope/gui/web/views/scope_view.py
+++ b/pyx2cscope/gui/web/views/scope_view.py
@@ -62,8 +62,7 @@ def load():
if isinstance(data, list) and data and isinstance(data[0], dict) and "variable" in data[0].keys():
web_scope.clear_scope_var()
for item in data:
- web_scope.clear_scope_var()
- var = web_scope.add_scope_var(item["variable"])
+ var = web_scope.add_scope_var(item["variable"], sfr=item.get("sfr", False))
if var is None:
msg = "Variable " + item["variable"] + " is not available."
else:
diff --git a/pyx2cscope/gui/web/views/watch_view.py b/pyx2cscope/gui/web/views/watch_view.py
index 0dab9dda..ff59fdbc 100644
--- a/pyx2cscope/gui/web/views/watch_view.py
+++ b/pyx2cscope/gui/web/views/watch_view.py
@@ -41,7 +41,7 @@ def load():
if isinstance(data, list) and data and isinstance(data[0], dict) and "variable" in data[0].keys():
web_scope.clear_watch_var()
for item in data:
- var = web_scope.add_watch_var(item["variable"])
+ var = web_scope.add_watch_var(item["variable"], sfr=item.get("sfr", False))
if var is None:
msg = "Variable " + item["variable"] + " is not available."
else:
diff --git a/pyx2cscope/gui/web/ws_handlers.py b/pyx2cscope/gui/web/ws_handlers.py
index e536743d..ad44a6ee 100644
--- a/pyx2cscope/gui/web/ws_handlers.py
+++ b/pyx2cscope/gui/web/ws_handlers.py
@@ -169,7 +169,7 @@ def handle_update_sample_control(data):
"""
data = {k: v[0] if v else '' for k, v in parse_qs(data).items()}
trigger_action = data.get('triggerAction', 'off')
- sample_time = int(data.get('sampleTime', 0))
+ sample_time = max(int(data.get('sampleTime', 1)), 1)
sample_freq = float(data.get('sampleFreq', 20))
web_scope.scope_set_sample(trigger_action, sample_time, sample_freq)
emit("sample_control_updated", {
@@ -220,8 +220,9 @@ def handle_add_dashboard_var(data):
data (dict): Dictionary containing the variable name.
"""
var = data.get("var")
+ sfr = bool(data.get("sfr", False))
if var:
- web_scope.add_dashboard_var(var)
+ web_scope.add_dashboard_var(var, sfr=sfr)
@socketio.on("remove_dashboard_var", namespace="/dashboard")
def handle_remove_dashboard_var(data):
diff --git a/pyx2cscope/parser/elf_parser.py b/pyx2cscope/parser/elf_parser.py
index 618c27f1..fcfde3a5 100644
--- a/pyx2cscope/parser/elf_parser.py
+++ b/pyx2cscope/parser/elf_parser.py
@@ -11,6 +11,12 @@
from pyx2cscope.variable.variable import VariableInfo
+ELF_MACHINE_TO_FAMILY = {
+ "EM_ARM": "arm",
+ "EM_DSPIC30F": "dspic",
+ "EM_MIPS": "pic32",
+}
+
class ElfParser(ABC):
"""Abstract base class for parsing ELF files.
@@ -40,9 +46,12 @@ def __init__(self, elf_path: str):
self.elf_path = elf_path
self.dwarf_info = {}
self.elf_file = None
+ self.elf_machine = None
+ self.target_signature = None
self.variable_map = {}
self.register_map = {}
self.symbol_table = {}
+ self.absolute_symbol_table = {}
self._load_elf_file()
self._load_symbol_table()
@@ -89,6 +98,17 @@ def get_var_list(self) -> List[str]:
"""
return sorted(self.variable_map.keys(), key=lambda x: x.lower())
+ def get_target_family(self) -> Optional[str]:
+ """Return the MCU family inferred from the ELF machine type, if known."""
+ elf_machine = self.elf_machine
+ if elf_machine is None:
+ return None
+ return ELF_MACHINE_TO_FAMILY.get(elf_machine)
+
+ def get_target_signature(self) -> Optional[str]:
+ """Return the best available ELF target signature for compatibility checks."""
+ return self.target_signature or self.get_target_family()
+
@abstractmethod
def _load_elf_file(self):
"""Load the ELF file according to the specific hardware architecture.
diff --git a/pyx2cscope/parser/generic_parser.py b/pyx2cscope/parser/generic_parser.py
index 6a3772ff..61385b69 100644
--- a/pyx2cscope/parser/generic_parser.py
+++ b/pyx2cscope/parser/generic_parser.py
@@ -5,6 +5,7 @@
import logging
import math
+import re
from itertools import product
from elftools.construct.lib import ListContainer
@@ -15,6 +16,20 @@
from pyx2cscope.parser.elf_parser import ElfParser
from pyx2cscope.variable.variable import VariableInfo
+TARGET_SIGNATURE_PATTERNS = (
+ ("dspic33a", ("__DSPIC33A", "__33AK")),
+ ("arm", ("__PIC32C", "PIC32C/", "SAME", "__GENERIC_ARM_", "ARMV6", "ARMV7")),
+ ("pic32", ("__PIC32",)),
+ ("dspic", ("__DSPIC", "__PIC24", "__33CK", "__33CH", "__33EP", "__33FJ")),
+)
+REGISTER_SYMBOL_PATTERN = re.compile(r"^[A-Z][A-Z0-9]*$")
+FAMILY_REGISTER_BYTE_SIZE = {
+ "arm": 4,
+ "pic32": 4,
+ "dspic33a": 4,
+ "dspic": 2,
+}
+
class GenericParser(ElfParser):
"""Class for parsing ELF files compatible with 32-bit architectures."""
@@ -33,6 +48,7 @@ def _load_elf_file(self):
try:
self.stream = open(self.elf_path, "rb")
self.elf_file = ELFFile(self.stream)
+ self.elf_machine = self.elf_file["e_machine"]
self.dwarf_info = self.elf_file.get_dwarf_info()
except IOError:
raise Exception(f"Error loading ELF file: {self.elf_path}")
@@ -69,7 +85,7 @@ def _get_die_variable(self, die_struct):
elif die_struct.attributes.get("DW_AT_external") and die_struct.attributes.get("DW_AT_name") is not None:
if die_struct.tag != "DW_TAG_variable":
return
- self.die_variable = die_struct
+ self.die_variable = die_struct
self.is_sfr = True
else:
return
@@ -104,6 +120,40 @@ def _process_die(self, die):
valid_values=member_data["valid_values"],
)
+ if self.is_sfr:
+ self._add_sfr_aliases(self.var_name, target_map)
+
+ @staticmethod
+ def _get_sfr_alias_names(register_name: str, member_name: str) -> list[str]:
+ """Return alternate names for SFR bitfield members."""
+ aliases = []
+ if "." not in member_name:
+ return aliases
+
+ member_leaf = member_name.split(".")[-1]
+ aliases.append(member_leaf)
+
+ return aliases
+
+ def _add_sfr_aliases(self, register_name: str, target_map: dict[str, VariableInfo]):
+ """Add convenience aliases for parsed SFR bitfield members."""
+ register_entries = list(target_map.items())
+ for member_name, variable_info in register_entries:
+ if not member_name.startswith(register_name + "."):
+ continue
+ for alias_name in self._get_sfr_alias_names(register_name, member_name):
+ if alias_name not in target_map:
+ target_map[alias_name] = VariableInfo(
+ name=alias_name,
+ type=variable_info.type,
+ byte_size=variable_info.byte_size,
+ bit_size=variable_info.bit_size,
+ bit_offset=variable_info.bit_offset,
+ address=variable_info.address,
+ array_size=variable_info.array_size,
+ valid_values=variable_info.valid_values,
+ )
+
def _get_base_type_die(self, current_die):
"""Find the base type die regarding the current selected die, i.e. array_type."""
type_attr = current_die.attributes.get("DW_AT_type")
@@ -170,15 +220,42 @@ def _extract_address(self, die_variable):
def _load_symbol_table(self):
"""Loads symbol table entries into a dictionary for fast access."""
+ all_symbol_names = []
for section in self.elf_file.iter_sections():
if isinstance(section, SymbolTableSection):
for symbol in section.iter_symbols():
- if symbol["st_info"].type == "STT_OBJECT" or symbol["st_info"].bind == "STB_GLOBAL":
+ all_symbol_names.append(symbol.name)
+ if symbol.name:
self.symbol_table[symbol.name] = symbol["st_value"]
+ if symbol["st_shndx"] == "SHN_ABS":
+ self.absolute_symbol_table[symbol.name] = {
+ "address": symbol["st_value"],
+ "size": symbol["st_size"],
+ "type": symbol["st_info"].type,
+ "bind": symbol["st_info"].bind,
+ }
+ self.target_signature = self._infer_target_signature(all_symbol_names)
+
+ def _infer_target_signature(self, symbol_names):
+ """Infer a more specific target signature from the ELF symbol table."""
+ symbol_names = "\n".join(symbol_names).upper()
+ for signature, markers in TARGET_SIGNATURE_PATTERNS:
+ if any(marker in symbol_names for marker in markers):
+ return signature
+ return self.get_target_family()
def _fetch_address_from_symtab(self, variable_name):
"""Fetches the address of a variable from the preloaded symbol table."""
- return self.symbol_table.get(variable_name, None)
+ candidates = [variable_name]
+ if not variable_name.startswith("_"):
+ candidates.append("_" + variable_name)
+ else:
+ candidates.append(variable_name[1:])
+
+ for candidate in candidates:
+ if candidate in self.symbol_table:
+ return self.symbol_table[candidate]
+ return None
def _find_actual_declaration(self, die_variable):
"""Find the actual declaration of an extern variable."""
@@ -402,6 +479,42 @@ def _map_registers(self) -> dict[str, VariableInfo]:
"""
return self.register_map
+ def _get_symbol_only_register_byte_size(self) -> int:
+ """Return a best-effort byte size for symbol-only SFR entries."""
+ signature = self.get_target_signature()
+ family = self.get_target_family()
+ return FAMILY_REGISTER_BYTE_SIZE.get(signature, FAMILY_REGISTER_BYTE_SIZE.get(family, 2))
+
+ @staticmethod
+ def _get_symbol_only_register_type(byte_size: int) -> str:
+ """Return the variable type string for a symbol-only SFR entry."""
+ type_by_size = {
+ 1: "unsigned char",
+ 2: "unsigned int",
+ 4: "unsigned long",
+ 8: "unsigned long long",
+ }
+ return type_by_size.get(byte_size, "unsigned int")
+
+ def _map_symbol_only_registers(self):
+ """Populate missing SFRs from absolute symbol-table entries when DWARF lacks variables."""
+ byte_size = self._get_symbol_only_register_byte_size()
+ for symbol_name, symbol_data in self.absolute_symbol_table.items():
+ if not REGISTER_SYMBOL_PATTERN.fullmatch(symbol_name):
+ continue
+ if symbol_name in self.register_map:
+ continue
+ self.register_map[symbol_name] = VariableInfo(
+ name=symbol_name,
+ type=self._get_symbol_only_register_type(byte_size),
+ byte_size=byte_size,
+ bit_size=0,
+ bit_offset=0,
+ address=symbol_data["address"],
+ array_size=0,
+ valid_values={},
+ )
+
def _map_variables(self) -> dict[str, VariableInfo]:
"""Maps all variables in the ELF file."""
self.variable_map.clear()
@@ -411,11 +524,13 @@ def _map_variables(self) -> dict[str, VariableInfo]:
self.expression_parser = DWARFExprParser(die.cu.structs)
self._process_die(die)
+ self._map_symbol_only_registers()
+
return self.variable_map
if __name__ == "__main__":
-
+
# elf_file = r"..\..\tests\data\qspin_foc_same54.elf"
elf_file = r"..\..\..\tests\data\dsPIC33ak128mc106_foc.elf"
elf_reader = GenericParser(elf_file)
@@ -425,13 +540,13 @@ def _map_variables(self) -> dict[str, VariableInfo]:
print(variable_map)
print(len(variable_map))
print("'''''''''''''''''''''''''''''''''''''''' ")
-
+
print("Array variables:")
for var_info in variable_map.values():
if var_info.array_size > 0:
print(var_info)
print("'''''''''''''''''''''''''''''''''''''''' ")
-
+
print("register variables:")
print(register_map)
print(len(register_map))
diff --git a/pyx2cscope/utils.py b/pyx2cscope/utils.py
index a0b3302e..9e410c20 100644
--- a/pyx2cscope/utils.py
+++ b/pyx2cscope/utils.py
@@ -4,6 +4,8 @@
get_config_file() -> ConfigParser: Retrieves the configuration file.
get_elf_file_path(key="path") -> str: Gets the path to the ELF file from the configuration.
get_com_port(key="com_port") -> str: Gets the COM port from the configuration.
+ get_host_address(key="host_ip") -> str: Gets the host IP address from the configuration.
+ get_can_config() -> dict: Gets the CAN interface configuration parameters.
"""
import logging
import os
@@ -20,6 +22,14 @@ def get_config_file() -> ConfigParser:
default_path = {"path": "path_to_your_elf_file"}
default_com = {"com_port": "your_com_port, ex:COM3"}
default_host_ip = {"host_ip": "your_host_ip, ex:192.168.1.100"}
+ default_can = {
+ "bustype": "pcan_usb",
+ "channel": "1",
+ "baud_rate": "500000",
+ "id_tx": "0x110",
+ "id_rx": "0x100",
+ "mode": "standard"
+ }
config = ConfigParser()
if os.path.exists(config_file):
config.read(config_file)
@@ -28,6 +38,7 @@ def get_config_file() -> ConfigParser:
config["ELF_FILE"] = default_path
config["COM_PORT"] = default_com
config["HOST_IP"] = default_host_ip
+ config["CAN"] = default_can
with open(config_file, "w") as configfile:
config.write(configfile)
logging.debug(f"Config file '{config_file}' created with default values")
@@ -76,3 +87,79 @@ def get_host_address(key="host_ip") -> str:
if not config["HOST_IP"][key] or "your" in config["HOST_IP"][key]:
return ""
return config["HOST_IP"][key]
+
+
+def get_can_config() -> dict:
+ """Gets the CAN interface configuration parameters from the configuration file.
+
+ Returns a dictionary with all CAN interface parameters including bustype, channel,
+ baud_rate, id_tx, id_rx, and mode. If values are not properly configured,
+ returns default values.
+
+ Returns:
+ dict: Dictionary containing CAN configuration parameters:
+ - bustype (str): CAN interface type (default: 'pcan_usb')
+ - channel (int): CAN channel number (default: 1)
+ - baud_rate (int): CAN baud rate in bps (default: 500000)
+ - id_tx (int): TX arbitration ID (default: 0x110)
+ - id_rx (int): RX arbitration ID (default: 0x100)
+ - mode (str): CAN ID mode 'standard' or 'extended' (default: 'standard')
+
+ Example:
+ >>> can_config = get_can_config()
+ >>> x2c = X2CScope(elf_file="firmware.elf", **can_config)
+ """
+ config = get_config_file()
+
+ # Default values
+ defaults = {
+ "bustype": "pcan_usb",
+ "channel": 1,
+ "baud_rate": 500000,
+ "id_tx": 0x110,
+ "id_rx": 0x100,
+ "mode": "standard"
+ }
+
+ # Check if CAN section exists
+ if "CAN" not in config:
+ return defaults
+
+ can_section = config["CAN"]
+ result = {}
+
+ # Get bustype
+ bustype = can_section.get("bustype", defaults["bustype"])
+ result["bustype"] = bustype if bustype and "your" not in bustype else defaults["bustype"]
+
+ # Get channel (convert to int)
+ try:
+ result["channel"] = int(can_section.get("channel", defaults["channel"]))
+ except (ValueError, TypeError):
+ result["channel"] = defaults["channel"]
+
+ # Get baud_rate (convert to int)
+ try:
+ result["baud_rate"] = int(can_section.get("baud_rate", defaults["baud_rate"]))
+ except (ValueError, TypeError):
+ result["baud_rate"] = defaults["baud_rate"]
+
+ # Get id_tx (convert hex string to int)
+ try:
+ id_tx_str = can_section.get("id_tx", hex(defaults["id_tx"]))
+ result["id_tx"] = int(id_tx_str, 16) if "0x" in id_tx_str.lower() else int(id_tx_str)
+ except (ValueError, TypeError):
+ result["id_tx"] = defaults["id_tx"]
+
+ # Get id_rx (convert hex string to int)
+ try:
+ id_rx_str = can_section.get("id_rx", hex(defaults["id_rx"]))
+ result["id_rx"] = int(id_rx_str, 16) if "0x" in id_rx_str.lower() else int(id_rx_str)
+ except (ValueError, TypeError):
+ result["id_rx"] = defaults["id_rx"]
+
+ # Get mode
+ mode = can_section.get("mode", defaults["mode"])
+ result["mode"] = mode if mode and mode.lower() in ["standard", "extended", "ext", "29bit"] else defaults["mode"]
+
+ return result
diff --git a/pyx2cscope/variable/variable.py b/pyx2cscope/variable/variable.py
index b4cad58b..02b87194 100644
--- a/pyx2cscope/variable/variable.py
+++ b/pyx2cscope/variable/variable.py
@@ -166,7 +166,7 @@ def _set_bit_value(self, value: Number):
shift = (8 * self.info.byte_size) - (self.info.bit_offset + self.info.bit_size)
mask = ((1 << self.info.bit_size) - 1) << shift
current_data &= ~mask
- return current_data | ((value << shift) & mask)
+ return current_data | ((int(value) << shift) & mask)
def get_value(self):
"""Get the stored value from the MCU.
diff --git a/pyx2cscope/variable/variable_factory.py b/pyx2cscope/variable/variable_factory.py
index 1fa11699..519fe98a 100644
--- a/pyx2cscope/variable/variable_factory.py
+++ b/pyx2cscope/variable/variable_factory.py
@@ -3,8 +3,10 @@
import logging
import os
import pickle
+import warnings
from dataclasses import asdict
from enum import Enum
+from typing import Optional
import yaml
@@ -64,6 +66,19 @@ class VariableFactory:
_get_variable_instance: Creates a Variable instance from provided information.
"""
+ _NAME_SFR_PAIR_LEN = 2
+ _DEVICE_FAMILY_KEYWORDS = {
+ "arm": ("ARM",),
+ "pic32": ("PIC32",),
+ "dspic": ("DSPIC", "PIC24"),
+ }
+ _DEVICE_SIGNATURE_KEYWORDS = {
+ "dspic33a": ("DSPIC33A", "33AK"),
+ "arm": ("ARM",),
+ "pic32": ("PIC32",),
+ "dspic": ("DSPIC", "PIC24"),
+ }
+
def __init__(self, l_net: LNet, elf_path=None):
"""Initialize the VariableFactory with LNet instance and path to the ELF file.
@@ -91,6 +106,7 @@ def set_elf_file(self, elf_path: str):
"""
parser = GenericParser
self.parser = parser(elf_path)
+ self._warn_if_incompatible(elf_path)
def set_lnet_interface(self, lnet: LNet):
"""Set the LNet interface to be used for data communication.
@@ -99,8 +115,61 @@ def set_lnet_interface(self, lnet: LNet):
lnet (LNet): the LNet interface
"""
self.l_net = lnet
+ self.device_info = self.l_net.get_device_info()
+
+ def _get_device_family(self) -> Optional[str]:
+ processor_id = str(getattr(self.device_info, "processor_id", "") or "")
+ for family, keywords in self._DEVICE_FAMILY_KEYWORDS.items():
+ if any(keyword in processor_id for keyword in keywords):
+ return family
+ return None
+
+ def _get_device_signature(self) -> Optional[str]:
+ processor_id = str(getattr(self.device_info, "processor_id", "") or "")
+ for signature, keywords in self._DEVICE_SIGNATURE_KEYWORDS.items():
+ if any(keyword in processor_id for keyword in keywords):
+ return signature
+ return None
+
+ def check_device_compatibility(self) -> dict:
+ """Check whether the loaded ELF appears compatible with the connected target."""
+ file_family = self.parser.get_target_family() if hasattr(self.parser, "get_target_family") else None
+ file_signature = self.parser.get_target_signature() if hasattr(self.parser, "get_target_signature") else file_family
+ device_family = self._get_device_family()
+ device_signature = self._get_device_signature() or device_family
+ checked = bool(file_signature and device_signature)
+ compatible = (file_signature == device_signature) if checked else None
+ if compatible is False:
+ reason = "ELF file and connected target appear to describe different MCU targets."
+ elif checked:
+ reason = "ELF file appears compatible with the connected target."
+ else:
+ reason = "Compatibility could not be determined."
+ return {
+ "checked": checked,
+ "compatible": compatible,
+ "device_family": device_family,
+ "file_family": file_family,
+ "device_signature": device_signature,
+ "file_signature": file_signature,
+ "processor_id": str(getattr(self.device_info, "processor_id", "") or ""),
+ "elf_file": getattr(self.parser, "elf_path", ""),
+ "reason": reason,
+ }
+
+ def _warn_if_incompatible(self, elf_path: str):
+ """Warn when the loaded ELF appears incompatible with the connected target."""
+ compatibility = self.check_device_compatibility()
+ if compatibility["compatible"] is False:
+ warnings.warn(
+ (
+ f"Loaded ELF '{elf_path}' appears incompatible with the connected target "
+ f"({compatibility['processor_id']})."
+ ),
+ stacklevel=2,
+ )
- def _build_export_file_name(self, filename: str = None, ext: FileType= FileType.YAML):
+ def _build_export_file_name(self, filename: Optional[str] = None, ext: FileType= FileType.YAML):
if filename is None:
if self.parser.elf_path is not None:
# get the elf_file name without extension
@@ -110,7 +179,38 @@ def _build_export_file_name(self, filename: str = None, ext: FileType= FileType.
return os.path.splitext(filename)[0] + ext.value
- def export_variables(self, filename: str = None, ext: FileType = FileType.YAML, items=None):
+ def _resolve_export_item(self, item):
+ """Resolve an export item to VariableInfo and its target map kind."""
+ variable_info = None
+ is_register = False
+
+ if isinstance(item, tuple) and len(item) == self._NAME_SFR_PAIR_LEN:
+ name, sfr = item
+ is_register = bool(sfr)
+ if isinstance(name, str):
+ variable_info = self.parser.get_var_info(name, sfr=is_register)
+ elif isinstance(item, Variable):
+ item = item.info.name
+
+ if isinstance(item, VariableInfo):
+ if self.parser.register_map.get(item.name) == item:
+ variable_info = item
+ is_register = True
+ elif self.parser.variable_map.get(item.name) == item:
+ variable_info = item
+ else:
+ variable_info = item
+ is_register = item.name in self.parser.register_map
+ elif isinstance(item, str):
+ if item in self.parser.variable_map:
+ variable_info = self.parser.variable_map.get(item)
+ elif item in self.parser.register_map:
+ variable_info = self.parser.register_map.get(item)
+ is_register = True
+
+ return variable_info, is_register
+
+ def export_variables(self, filename: Optional[str] = None, ext: FileType = FileType.YAML, items=None):
"""Store the variables registered on the elf file to a pickle file.
Args:
@@ -122,13 +222,17 @@ def export_variables(self, filename: str = None, ext: FileType = FileType.YAML,
raise ValueError("Elf file is not yet supported as export format...")
filename = self._build_export_file_name(filename, ext)
- export_dict = {}
+ export_dict = {"variables": {}, "registers": {}}
if items:
for item in items:
- variable_name = item.info.name if isinstance(item, Variable) else item
- export_dict[variable_name] = self.parser.variable_map.get(variable_name)
+ variable_info, is_register = self._resolve_export_item(item)
+ if variable_info is None:
+ continue
+ target_key = "registers" if is_register else "variables"
+ export_dict[target_key][variable_info.name] = variable_info
else:
- export_dict = self.parser.variable_map
+ export_dict["variables"] = dict(self.parser.variable_map)
+ export_dict["registers"] = dict(self.parser.register_map)
if ext is FileType.PICKLE:
with open(filename, 'wb') as file:
@@ -156,15 +260,25 @@ def import_variables(self, filename: str):
# clear any previous loaded variable
self.parser.variable_map.clear()
+ self.parser.register_map.clear()
+ imported_data = None
if ext is FileType.ELF:
self.parser = GenericParser(filename)
+ self._warn_if_incompatible(filename)
if ext is FileType.PICKLE:
with open(filename, 'rb') as file:
- self.parser.variable_map = pickle.load(file)
+ imported_data = pickle.loads(file.read())
if ext is FileType.YAML:
with open(filename, 'r') as file:
- self.parser.variable_map = yaml.load(file, Loader=yaml.FullLoader)
+ imported_data = yaml.load(file.read(), Loader=yaml.FullLoader)
+
+ if ext is not FileType.ELF:
+ if isinstance(imported_data, dict) and "variables" in imported_data:
+ self.parser.variable_map = imported_data.get("variables", {}) or {}
+ self.parser.register_map = imported_data.get("registers", {}) or {}
+ else:
+ self.parser.variable_map = imported_data or {}
logging.debug(f"Variables loaded from {filename}")
@@ -196,6 +310,9 @@ def get_variable(self, name: str, sfr: bool = False) -> Variable | None:
"""
try:
variable_info = self.parser.get_var_info(name, sfr=sfr)
+ if variable_info is None:
+ logging.error(f"Variable '{name}' not found!")
+ return None
if variable_info is None:
logging.error(f"Variable '{name}' not found!")
return None
diff --git a/pyx2cscope/x2cscope.py b/pyx2cscope/x2cscope.py
index 293425ca..d089c26a 100644
--- a/pyx2cscope/x2cscope.py
+++ b/pyx2cscope/x2cscope.py
@@ -9,7 +9,7 @@
import logging
from dataclasses import dataclass
from numbers import Number
-from typing import Dict, List
+from typing import Dict, List, Optional
from mchplnet.interfaces.abstract_interface import Interface
from mchplnet.interfaces.factory import InterfaceFactory, InterfaceType
@@ -156,7 +156,7 @@ def get_variable_raw(self, variable_info: VariableInfo) -> Variable:
"""
return self.variable_factory.get_variable_raw(variable_info)
- def export_variables(self, filename: str = None, ext: FileType = FileType.YAML, items=None):
+ def export_variables(self, filename: Optional[str] = None, ext: FileType = FileType.YAML, items=None):
"""Store the variables registered on the elf file to a pickle file.
Args:
@@ -177,6 +177,10 @@ def import_variables(self, filename: str):
"""
self.variable_factory.import_variables(filename)
+ def check_compatibility(self) -> dict:
+ """Check whether the currently loaded ELF appears compatible with the connected target."""
+ return self.variable_factory.check_device_compatibility()
+
def add_scope_channel(self, variable: Variable, trigger: bool = False) -> int:
"""Add a variable as a scope channel.
@@ -256,13 +260,14 @@ def set_sample_time(self, sample_time: int):
"""Define the resolution how the samples will be buffered at the internal buffer.
This can be used to extend total sampling time at the cost of resolution.
- 0 = every sample, 1 = every 2nd sample, 2 = every 3rd sample …
+ In the pyX2Cscope API, values start at 1, while LNET uses a 0-based pre-scaler.
+ 1 = every sample, 2 = every 2nd sample, 3 = every 3rd sample ...
Args:
sample_time (int): The sample time factor.
"""
- sample_time = 0 if sample_time < 0 else sample_time
- self.scope_setup.set_sample_time_factor(sample_time)
+ sample_time = 1 if sample_time < 1 else sample_time
+ self.scope_setup.set_sample_time_factor(sample_time - 1)
def set_scope_state(self, scope_state: int):
"""Set the state of the scope.
@@ -343,6 +348,7 @@ def _read_array_chunks(self) -> List[bytearray]:
current_address = self.lnet.scope_data.data_array_address + i * chunk_size
data_size = chunk_size if i < int(num_chunks) else int(chunk_rest)
try:
+ # Read the chunk of data
data = self.lnet.get_ram_array(current_address, data_size, data_type)
chunk_data.extend(data)
except Exception as e:
diff --git a/quality.txt b/quality.txt
deleted file mode 100644
index 27746401..00000000
Binary files a/quality.txt and /dev/null differ
diff --git a/requirements.txt b/requirements.txt
deleted file mode 100644
index d9a6c953..00000000
Binary files a/requirements.txt and /dev/null differ
diff --git a/tests/test_can_reconnection.py b/tests/test_can_reconnection.py
new file mode 100644
index 00000000..deb46c87
--- /dev/null
+++ b/tests/test_can_reconnection.py
@@ -0,0 +1,252 @@
+"""Tests for CAN interface reconnection bug fix.
+
+This test suite verifies that the CAN interface can be properly disconnected
+and reconnected multiple times without errors, preventing the regression of
+the "PCAN Channel has not been initialized" bug.
+"""
+
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from mchplnet.interfaces.can import LNetCan
+from pyx2cscope.x2cscope import X2CScope
+from tests import data
+
+EXPECTED_BUS_CALLS_AFTER_RECONNECT = 2
+RECONNECT_CYCLES = 3
+EXPECTED_BUS_CALLS_AFTER_CYCLES = RECONNECT_CYCLES + 1
+
+
+@pytest.fixture
+def mock_can_bus():
+ """Create a mock CAN bus for testing."""
+ with patch('can.interface.Bus') as mock_bus_class:
+ mock_bus_instance = MagicMock()
+ mock_bus_instance._is_shutdown = False
+ mock_bus_instance.recv = MagicMock(return_value=None)
+ mock_bus_instance.send = MagicMock()
+ mock_bus_instance.shutdown = MagicMock()
+ mock_bus_class.return_value = mock_bus_instance
+ yield mock_bus_class, mock_bus_instance
+
+
+class TestCANReconnection:
+ """Test CAN interface reconnection scenarios."""
+
+ elf_file = os.path.join(
+ os.path.dirname(data.__file__), "mc_foc_sl_fip_dspic33ck_mclv48v300w.elf"
+ )
+
+ def test_can_interface_start_cleans_up_existing_bus(self, mock_can_bus):
+ """Test that start() properly cleans up an existing bus before creating a new one."""
+ mock_bus_class, mock_bus_instance = mock_can_bus
+
+ # Create CAN interface
+ can_interface = LNetCan(
+ bustype="pcan_usb",
+ channel=1,
+ baud_rate=500000,
+ id_tx=0x110,
+ id_rx=0x100,
+ )
+
+ # Start the interface (first time)
+ can_interface.start()
+ assert mock_bus_class.call_count == 1
+ assert can_interface.bus is not None
+
+ # Start again without stopping (simulates reconnection attempt)
+ can_interface.start()
+
+ # Verify shutdown was called on the old bus
+ assert mock_bus_instance.shutdown.called
+ # Verify a new bus was created (total 2 calls)
+ assert mock_bus_class.call_count == EXPECTED_BUS_CALLS_AFTER_RECONNECT
+
+ def test_can_interface_stop_clears_bus_reference(self, mock_can_bus):
+ """Test that stop() properly clears the bus reference."""
+ mock_bus_class, mock_bus_instance = mock_can_bus
+
+ can_interface = LNetCan(
+ bustype="pcan_usb",
+ channel=1,
+ baud_rate=500000,
+ )
+
+ # Start and stop
+ can_interface.start()
+ assert can_interface.bus is not None
+
+ can_interface.stop()
+
+ # Verify shutdown was called and bus is None
+ assert mock_bus_instance.shutdown.called
+ assert can_interface.bus is None
+
+ def test_can_interface_multiple_reconnections(self, mock_can_bus):
+ """Test multiple connect-disconnect-reconnect cycles."""
+ mock_bus_class, mock_bus_instance = mock_can_bus
+
+ can_interface = LNetCan(
+ bustype="pcan_usb",
+ channel=1,
+ baud_rate=500000,
+ )
+
+ # Perform 3 connect-disconnect cycles
+ for i in range(3):
+ # Start
+ can_interface.start()
+ assert can_interface.bus is not None
+ assert mock_bus_class.call_count == i + 1
+
+ # Stop
+ can_interface.stop()
+ assert can_interface.bus is None
+ assert mock_bus_instance.shutdown.call_count == i + 1
+
+ # Verify we can start again after all cycles
+ can_interface.start()
+ assert can_interface.bus is not None
+ assert mock_bus_class.call_count == EXPECTED_BUS_CALLS_AFTER_CYCLES
+
+ def test_can_interface_stop_handles_shutdown_error(self, mock_can_bus):
+ """Test that stop() handles errors during bus shutdown gracefully."""
+ mock_bus_class, mock_bus_instance = mock_can_bus
+
+ # Make shutdown raise an exception
+ mock_bus_instance.shutdown.side_effect = Exception("Shutdown error")
+
+ can_interface = LNetCan(
+ bustype="pcan_usb",
+ channel=1,
+ baud_rate=500000,
+ )
+
+ can_interface.start()
+ assert can_interface.bus is not None
+
+ # Stop should not raise exception even if shutdown fails
+ can_interface.stop()
+
+ # Bus should still be set to None
+ assert can_interface.bus is None
+
+ def test_x2cscope_disconnect_calls_interface_stop(self, mock_can_bus):
+ """Test that X2CScope.disconnect() properly calls interface.stop()."""
+ mock_bus_class, mock_bus_instance = mock_can_bus
+
+ # Mock the read/write methods to avoid actual CAN communication
+ with patch('mchplnet.interfaces.can.LNetCan.read') as mock_read, \
+ patch('mchplnet.interfaces.can.LNetCan.write'):
+
+ # Mock read to return valid LNet DeviceInfo frame
+ # Frame structure (49+ bytes):
+ # SYN(0x55), SIZE, NODE, SERVICE_ID, STATUS,
+ # MONITOR_VER(2), APP_VER(2), PROCESSOR_ID(2),
+ # MONITOR_DATE(9), MONITOR_TIME(4), APP_DATE(9), APP_TIME(4),
+ # DSP_STATE(1), EVENT_TYPE(2), EVENT_ID(4), TABLE_STRUCT_ADD(4)
+ device_info_frame = bytearray(
+ b'\x55\x2E\x01\x11\x00' # SYN, SIZE(46 data bytes), NODE, SERVICE_ID, STATUS
+ b'\x01\x00' # Monitor version (little-endian)
+ b'\x01\x00' # App version (little-endian)
+ b'\x10\x82' # Processor ID 0x8210 (16-bit generic dsPIC)
+ b'01/01/2024' # Monitor date (9 bytes)
+ b'1200' # Monitor time (4 bytes)
+ b'01/01/2024' # App date (9 bytes)
+ b'1200' # App time (4 bytes)
+ b'\x01' # DSP state (0x01 = Application runs on target)
+ b'\x00\x00' # Event type (2 bytes)
+ b'\x00\x00\x00\x00' # Event ID (4 bytes)
+ b'\x00\x00\x00\x00' # Table struct address (4 bytes)
+ b'\x00' # CRC
+ )
+ # Mock LoadParameters frame (simpler, just status response)
+ load_param_frame = bytearray(b'\x55\x00\x01\x12\x00\x00')
+ # Mock LoadScopeData frame
+ load_scope_frame = bytearray(b'\x55\x08\x01\x13\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00')
+
+ # Return different frames for each call
+ mock_read.side_effect = [device_info_frame, load_param_frame, load_scope_frame]
+
+ # Create X2CScope with CAN interface
+ x2cscope = X2CScope(
+ elf_file=self.elf_file,
+ bustype="pcan_usb",
+ channel=1,
+ baud_rate=500000,
+ )
+
+ # Verify interface was started
+ assert mock_bus_class.called
+
+ # Disconnect
+ x2cscope.disconnect()
+
+ # Verify shutdown was called
+ assert mock_bus_instance.shutdown.called
+
+
+
+class TestCANInterfaceStartupCleanup:
+ """Test CAN interface cleanup on startup."""
+
+ def test_start_with_none_bus_creates_new_bus(self, mock_can_bus):
+ """Test that start() creates a bus when bus is None."""
+ mock_bus_class, _ = mock_can_bus
+
+ can_interface = LNetCan(bustype="pcan_usb", channel=1)
+ assert can_interface.bus is None
+
+ can_interface.start()
+
+ assert can_interface.bus is not None
+ assert mock_bus_class.call_count == 1
+
+ def test_start_with_existing_bus_stops_first(self, mock_can_bus):
+ """Test that start() stops existing bus before creating new one."""
+ mock_bus_class, mock_bus_instance = mock_can_bus
+
+ can_interface = LNetCan(bustype="pcan_usb", channel=1)
+
+ # First start
+ can_interface.start()
+ first_bus = can_interface.bus
+ assert first_bus is not None
+
+ # Second start without stopping
+ can_interface.start()
+
+ # Verify old bus was shut down
+ assert mock_bus_instance.shutdown.called
+ # Verify new bus was created
+ assert mock_bus_class.call_count == EXPECTED_BUS_CALLS_AFTER_RECONNECT
+
+ def test_is_open_returns_false_when_bus_is_none(self):
+ """Test that is_open() returns False when bus is None."""
+ can_interface = LNetCan(bustype="pcan_usb", channel=1)
+ assert can_interface.bus is None
+ assert can_interface.is_open() is False
+
+ def test_is_open_returns_true_when_bus_is_active(self, mock_can_bus):
+ """Test that is_open() returns True when bus is active."""
+ mock_bus_class, mock_bus_instance = mock_can_bus
+
+ can_interface = LNetCan(bustype="pcan_usb", channel=1)
+ can_interface.start()
+
+ assert can_interface.is_open() is True
+
+ def test_is_open_returns_false_when_bus_is_shutdown(self, mock_can_bus):
+ """Test that is_open() returns False when bus is shutdown."""
+ mock_bus_class, mock_bus_instance = mock_can_bus
+
+ can_interface = LNetCan(bustype="pcan_usb", channel=1)
+ can_interface.start()
+
+ # Simulate bus shutdown
+ mock_bus_instance._is_shutdown = True
+
+ assert can_interface.is_open() is False
diff --git a/tests/test_install.py b/tests/test_install.py
index 95e36364..4923e61d 100644
--- a/tests/test_install.py
+++ b/tests/test_install.py
@@ -1,13 +1,15 @@
"""Integration test to check if after install, PyX2CScope outputs the expected behavior."""
import os
+import subprocess
+import sys
import pyx2cscope
from mchplnet.services.frame_device_info import DeviceInfo
from mchplnet.services.frame_load_parameter import LoadScopeData
from pyx2cscope.x2cscope import X2CScope
from tests import data
-from tests.utils.serial_stub import fake_serial
+from tests.utils.serial_stub import DEVICE_PROFILE_ARM, fake_serial
elf_file_16 = os.path.join(
os.path.dirname(data.__file__), "mc_foc_sl_fip_dspic33ck_mclv48v300w.elf"
@@ -34,12 +36,12 @@ def test_x2cscope_install_serial_16(mocker):
def test_x2cscope_install_serial_32(mocker):
"""Using a fake serial interface with a 32 bit device, check the device_info and scope_data frames."""
- fake_serial(mocker, 32)
+ fake_serial(mocker, 32, device_profile=DEVICE_PROFILE_ARM)
pyx2cscope = X2CScope(elf_file=elf_file_32, port="COM11")
device_info: DeviceInfo = pyx2cscope.lnet.device_info
assert device_info.uc_width == PROCESSOR_32_BIT_LENGTH, "wrong processor bit length"
assert (
- device_info.processor_id == "__GENERIC_MICROCHIP_PIC32__"
+ device_info.processor_id == "__GENERIC_ARM_ARMV7__"
), "unknown processor"
scope_data: LoadScopeData = pyx2cscope.lnet.scope_data
assert scope_data.scope_state == 0, "wrong scope state value"
@@ -48,8 +50,13 @@ def test_x2cscope_install_serial_32(mocker):
def test_x2cscope_install_script_bin():
"""Test if the script scripts/pyx2cscope executable is installed correctly."""
- result = os.popen("pyx2cscope -v").read().split(" ")[1].strip("\n")
- assert result == pyx2cscope.__version__, "x2cscope script not working"
+ result = subprocess.run(
+ [sys.executable, "-m", "pyx2cscope", "-v"],
+ capture_output=True,
+ text=True, check=False,
+ )
+ version = result.stdout.strip().split(" ")[-1].strip()
+ assert version == pyx2cscope.__version__, "x2cscope script not working"
# def test_web_x2cscope_install():
diff --git a/tests/test_parser.py b/tests/test_parser.py
index 8be032e0..3c285055 100644
--- a/tests/test_parser.py
+++ b/tests/test_parser.py
@@ -5,12 +5,17 @@
from pyx2cscope.variable.variable_factory import FileType
from pyx2cscope.x2cscope import X2CScope
from tests import data
-from tests.utils.serial_stub import fake_serial
+from tests.utils.serial_stub import DEVICE_PROFILE_ARM, DEVICE_PROFILE_DSPIC33A, fake_serial
class TestParser:
"""Parser related unit tests."""
+ LATE3_BIT_MASK = 0x0008
+ DMA0CH_ADDRESS = 8976
+ DMA0STAT_ADDRESS = 8984
+ DMA0CH_BYTE_SIZE = 4
+
elf_file_16 = os.path.join(
os.path.dirname(data.__file__), "MCAF_ZSMT_dsPIC33CK.elf"
)
@@ -22,6 +27,52 @@ class TestParser:
os.path.dirname(data.__file__), "dsPIC33ak128mc106_foc.elf"
)
+ def test_sfr_bitfield_aliases_dspic33ck(self, mocker):
+ """Check dsPIC33CK SFR bitfields are exposed with nested and flat aliases."""
+ serial_stub = fake_serial(mocker, 16)
+ x2c_scope = X2CScope(port="COM14", elf_file=self.elf_file_16)
+ late_bit = x2c_scope.get_variable("LATE3", sfr=True)
+ late_bit_nested = x2c_scope.get_variable("LATEbits.LATE3", sfr=True)
+
+ assert late_bit is not None
+ assert late_bit_nested is not None
+ assert late_bit.info.address == late_bit_nested.info.address
+ assert late_bit.info.bit_size == 1
+
+ serial_stub.mock_memory[late_bit.info.address] = 0
+ late_bit.set_value(int("1"))
+ assert serial_stub.mock_memory[late_bit.info.address] == self.LATE3_BIT_MASK
+ assert late_bit.get_value() == 1
+
+ def test_sfr_bitfield_aliases_dspic33a(self, mocker):
+ """Check dsPIC33A SFR bitfields are exposed with nested and flat aliases."""
+ fake_serial(mocker, 32, device_profile=DEVICE_PROFILE_DSPIC33A)
+ x2c_scope = X2CScope(port="COM14")
+ x2c_scope.import_variables(self.elf_file_dspic33ak)
+
+ ansela_bit = x2c_scope.get_variable("ANSELA0", sfr=True)
+ ansela_bit_nested = x2c_scope.get_variable("ANSELAbits.ANSELA0", sfr=True)
+
+ assert ansela_bit is not None
+ assert ansela_bit_nested is not None
+ assert ansela_bit.info.address == ansela_bit_nested.info.address
+ assert ansela_bit.info.bit_size == 1
+
+ def test_symbol_only_sfr_is_listed_with_address(self, mocker):
+ """Check symbol-only SFRs like DMA0CH are exposed with their absolute address."""
+ fake_serial(mocker, 32, device_profile=DEVICE_PROFILE_DSPIC33A)
+ x2c_scope = X2CScope(port="COM14")
+ x2c_scope.import_variables(self.elf_file_dspic33ak)
+
+ dma0ch = x2c_scope.get_variable("DMA0CH", sfr=True)
+ dma0stat = x2c_scope.get_variable("DMA0STAT", sfr=True)
+
+ assert dma0ch is not None
+ assert dma0stat is not None
+ assert dma0ch.info.address == self.DMA0CH_ADDRESS
+ assert dma0stat.info.address == self.DMA0STAT_ADDRESS
+ assert dma0ch.info.byte_size == self.DMA0CH_BYTE_SIZE
+
def test_variable_16_does_not_exist(self, mocker):
"""Given a valid 16 bit elf file, check if an invalid variable outputs the expected behavior."""
fake_serial(mocker, 16)
@@ -31,7 +82,7 @@ def test_variable_16_does_not_exist(self, mocker):
def test_variable_32_does_not_exist(self, mocker):
"""Given a valid 32 bit elf file, check if an invalid variable outputs the expected behavior."""
- fake_serial(mocker, 16)
+ fake_serial(mocker, 32, device_profile=DEVICE_PROFILE_ARM)
x2c_scope = X2CScope(port="COM14", elf_file=self.elf_file_32)
variable = x2c_scope.get_variable("wrong_variable_name")
assert variable is None
@@ -47,7 +98,7 @@ def test_array_variable_16(self, mocker, array_size_test=4):
def test_array_variable_32(self, mocker, array_size_test=255):
"""Given a valid 32 bit elf file, check if an array variable is read correctly."""
- fake_serial(mocker, 32)
+ fake_serial(mocker, 32, device_profile=DEVICE_PROFILE_ARM)
x2c_scope = X2CScope(port="COM14", elf_file=self.elf_file_32)
variable = x2c_scope.get_variable("bufferLNet")
assert variable is not None, "variable name not found"
@@ -69,7 +120,7 @@ def test_union_variable_16(self, mocker):
def test_variable_dspic33ak(self, mocker, array_size_test=4900, address=22122):
"""Given a valid dspic33ak elf file, check if an array variable is read correctly."""
- fake_serial(mocker, 32)
+ fake_serial(mocker, 32, device_profile=DEVICE_PROFILE_DSPIC33A)
x2c_scope = X2CScope(port="COM14")
x2c_scope.import_variables(self.elf_file_dspic33ak)
variable = x2c_scope.get_variable("measureInputs.current.Ia")
@@ -84,7 +135,7 @@ def test_variable_dspic33ak(self, mocker, array_size_test=4900, address=22122):
def test_nested_array_variable_32(self, mocker, array_size_test=3):
"""Given a valid 32 bit elf file, check if an array variable is read correctly."""
- fake_serial(mocker, 32)
+ fake_serial(mocker, 32, device_profile=DEVICE_PROFILE_ARM)
x2c_scope = X2CScope(port="COM14", elf_file=self.elf_file_32)
variable = x2c_scope.get_variable("mcFocI_ModuleData_gds.dOutput.duty")
assert variable is not None, "variable name not found"
@@ -93,7 +144,7 @@ def test_nested_array_variable_32(self, mocker, array_size_test=3):
def test_variable_enum_32(self, mocker, size6=6, size3=3):
"""Given a valid dspic33ck elf file, check if an enum variable is read correctly."""
- fake_serial(mocker, 32)
+ fake_serial(mocker, 32, device_profile=DEVICE_PROFILE_ARM)
x2c_scope = X2CScope(port="COM14", elf_file=self.elf_file_32)
# test simple variable enum
variable = x2c_scope.get_variable("nextGlobalState")
@@ -108,7 +159,7 @@ def test_variable_enum_32(self, mocker, size6=6, size3=3):
def test_variable_enum_16(self, mocker, size6=6):
"""Given a valid dspic33ck elf file, check if an enum variable is read correctly."""
- fake_serial(mocker, 32)
+ fake_serial(mocker, 16)
x2c_scope = X2CScope(port="COM14", elf_file=self.elf_file_16)
# test nested enum inside a structure
variable = x2c_scope.get_variable("motor.apiData.motorStatus")
@@ -116,9 +167,10 @@ def test_variable_enum_16(self, mocker, size6=6):
assert variable.is_array() == False, "variable should not be an array"
assert len(variable.info.valid_values) == size6, "enum size should be 3"
- def test_variable_export_import(self, mocker):
+ def test_variable_export_import(self, mocker, tmp_path, monkeypatch):
"""Given a valid 32 bit elf file, check if export and import functions for variables are working."""
- fake_serial(mocker, 32)
+ monkeypatch.chdir(tmp_path)
+ fake_serial(mocker, 32, device_profile=DEVICE_PROFILE_ARM)
x2c_scope = X2CScope(port="COM14")
# try to import elf file instead of loading directly from the constructor
x2c_scope.import_variables(self.elf_file_32)
@@ -141,6 +193,7 @@ def test_variable_export_import(self, mocker):
assert variable.info.name == variable_reloaded.info.name, "variables don't have the same name"
assert variable.info.address == variable_reloaded.info.address, "variables don't have the same address"
assert variable.info.array_size == variable_reloaded.info.array_size, "variables don't have the same array size"
+ x2c_reloaded.disconnect()
# load generated pickle file with single variable
x2c_reloaded = X2CScope(port="COM14")
@@ -150,8 +203,26 @@ def test_variable_export_import(self, mocker):
assert variable.info.name == variable_reloaded.info.name, "variables don't have the same name"
assert variable.info.address == variable_reloaded.info.address, "variables don't have the same address"
assert variable.info.array_size == variable_reloaded.info.array_size, "variables don't have the same array size"
+ x2c_reloaded.disconnect()
+
+ def test_sfr_export_import(self, mocker, tmp_path, monkeypatch):
+ """Check exported SFR entries can be re-imported and resolved as registers."""
+ monkeypatch.chdir(tmp_path)
+ fake_serial(mocker, 32, device_profile=DEVICE_PROFILE_DSPIC33A)
+ x2c_scope = X2CScope(port="COM14")
+ x2c_scope.import_variables(self.elf_file_dspic33ak)
+
+ sfr_list = x2c_scope.list_sfr()
+ assert sfr_list, "expected at least one SFR in the imported ELF"
+
+ sfr_name = sfr_list[0]
+ x2c_scope.export_variables("my_sfr_variable", items=[(sfr_name, True)])
+ assert os.path.exists("my_sfr_variable.yml") == True, "SFR export yaml file name not found"
+
+ x2c_reloaded = X2CScope(port="COM14")
+ x2c_reloaded.import_variables(filename="my_sfr_variable.yml")
+ sfr_reloaded = x2c_reloaded.get_variable(sfr_name, sfr=True)
- # house keeping -> delete generated files
- os.remove("my_variables.yml")
- os.remove("qspin_foc_same54.yml")
- os.remove("my_single_variable.pkl")
+ assert sfr_reloaded is not None, "reloaded SFR should be available"
+ assert sfr_reloaded.info.name == sfr_name, "reloaded SFR name mismatch"
+ assert len(x2c_reloaded.list_sfr()) == 1, "import loaded more than one SFR"
diff --git a/tests/test_pyx2cscope_class.py b/tests/test_pyx2cscope_class.py
index 4615fd97..be39c319 100644
--- a/tests/test_pyx2cscope_class.py
+++ b/tests/test_pyx2cscope_class.py
@@ -9,6 +9,8 @@
from tests import data
from tests.utils.serial_stub import fake_serial
+LNET_SAMPLE_TIME_THIRD_SAMPLE = 2
+
class TestPyX2CScope:
"""Tests related to the PyX2CScope class."""
@@ -16,6 +18,12 @@ class TestPyX2CScope:
elf_file = os.path.join(
os.path.dirname(data.__file__), "mc_foc_sl_fip_dspic33ck_mclv48v300w.elf"
)
+ arm_elf_file = os.path.join(
+ os.path.dirname(data.__file__), "qspin_foc_same54.elf"
+ )
+ dspic33a_elf_file = os.path.join(
+ os.path.dirname(data.__file__), "dsPIC33ak128mc106_foc.elf"
+ )
def test_missing_elf_file_16(self, mocker):
"""Check if the corresponding exception is raised in case of wrong 16 bit elf path."""
@@ -53,8 +61,8 @@ def test_missing_interface(self, mocker):
def test_missing_com_port(self, mocker):
"""Check class behavior in case of non COM-Port initialization.
- A com-port must not be provided, in this case, the default COM1
- is used but a warning is generated on the console.
+ A com-port must not be provided, in this case, AUTO detection
+ is attempted but a warning is generated on the console.
"""
fake_serial(mocker, 16)
with warnings.catch_warnings(record=True) as w:
@@ -67,14 +75,69 @@ def test_missing_com_port(self, mocker):
# Verify the warning was raised
assert len(w) == 2 # noqa: PLR2004
assert issubclass(w[-1].category, Warning) is True
- assert "No port provided, using default COM1" in str(w[-1].message)
+ assert "No port provided, will attempt auto-detection" in str(w[-1].message)
# Clean up
scope.disconnect()
def test_wrong_com_port(self):
"""Check handling of a non-existent COM-PORT input."""
- with pytest.raises(
- RuntimeError, match=r"Failed to retrieve device information"
- ):
+ with pytest.raises(Exception, match=r"could not open port '?COM0'?"):
X2CScope(elf_file=self.elf_file, port="COM0")
+
+ def test_set_sample_time_uses_user_facing_factor(self, mocker):
+ """Check sample time values are translated from 1-based API to 0-based LNET."""
+ fake_serial(mocker, 16)
+ scope = X2CScope(elf_file=self.elf_file, port="COM1")
+ try:
+ scope.set_sample_time(1)
+ assert scope.scope_setup.sample_time_factor == 0
+
+ scope.set_sample_time(3)
+ assert scope.scope_setup.sample_time_factor == LNET_SAMPLE_TIME_THIRD_SAMPLE
+
+ scope.set_sample_time(0)
+ assert scope.scope_setup.sample_time_factor == 0
+ finally:
+ scope.disconnect()
+
+ def test_incompatible_elf_emits_warning(self, mocker):
+ """Check mismatched ELF and target families generate a warning."""
+ fake_serial(mocker, 16)
+ with warnings.catch_warnings(record=True) as captured:
+ warnings.simplefilter("always")
+ scope = X2CScope(elf_file=self.arm_elf_file, port="COM1")
+ try:
+ assert any("appears incompatible" in str(w.message) for w in captured)
+ finally:
+ scope.disconnect()
+
+ def test_check_compatibility(self, mocker):
+ """Check the compatibility API reports a matching dsPIC ELF correctly."""
+ fake_serial(mocker, 16)
+ scope = X2CScope(elf_file=self.elf_file, port="COM1")
+ try:
+ compatibility = scope.check_compatibility()
+ assert compatibility["checked"] is True
+ assert compatibility["compatible"] is True
+ assert compatibility["file_family"] == "dspic"
+ assert compatibility["device_family"] == "dspic"
+ assert compatibility["file_signature"] == "dspic"
+ finally:
+ scope.disconnect()
+
+ def test_check_compatibility_detects_dspic33a_mismatch(self, mocker):
+ """Check the compatibility API distinguishes dsPIC33A from generic dsPIC targets."""
+ fake_serial(mocker, 16)
+ with warnings.catch_warnings(record=True) as captured:
+ warnings.simplefilter("always")
+ scope = X2CScope(elf_file=self.dspic33a_elf_file, port="COM1")
+ try:
+ compatibility = scope.check_compatibility()
+ assert compatibility["checked"] is True
+ assert compatibility["compatible"] is False
+ assert compatibility["file_signature"] == "dspic33a"
+ assert compatibility["device_signature"] == "dspic"
+ assert any("appears incompatible" in str(w.message) for w in captured)
+ finally:
+ scope.disconnect()
diff --git a/tests/test_qt_gui.py b/tests/test_qt_gui.py
index e2682761..1fd5ddec 100644
--- a/tests/test_qt_gui.py
+++ b/tests/test_qt_gui.py
@@ -10,13 +10,28 @@
"""
import os
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
import pytest
# Set headless mode before importing Qt modules
os.environ["QT_QPA_PLATFORM"] = "offscreen"
+from tests import data
+
+
+@pytest.fixture
+def mock_can_bus():
+ """Create a mock CAN bus for testing."""
+ with patch('can.interface.Bus') as mock_bus_class:
+ mock_bus_instance = MagicMock()
+ mock_bus_instance._is_shutdown = False
+ mock_bus_instance.recv = MagicMock(return_value=None)
+ mock_bus_instance.send = MagicMock()
+ mock_bus_instance.shutdown = MagicMock()
+ mock_bus_class.return_value = mock_bus_instance
+ yield mock_bus_class, mock_bus_instance
+
class TestAppStateModel:
"""Tests for AppState model."""
@@ -259,6 +274,38 @@ def test_connect_uart_creates_x2cscope(
assert result is True
+ def test_connect_uart_imports_selected_file(self, connection_manager, mocker):
+ """Test UART connection always loads variables through import_variables."""
+ mock_x2c = MagicMock()
+ mock_x2c.list_variables.return_value = []
+ mocker.patch(
+ "pyx2cscope.gui.qt.controllers.connection_manager.X2CScope",
+ return_value=mock_x2c,
+ )
+
+ result = connection_manager.connect_uart(
+ port="COM1", baud_rate=115200, elf_file="firmware.elf"
+ )
+
+ assert result is True
+ mock_x2c.import_variables.assert_called_once_with("firmware.elf")
+
+ def test_connect_tcp_imports_yml_after_connect(self, connection_manager, mocker):
+ """Test TCP connection loads YML imports after creating the transport."""
+ mock_x2c = MagicMock()
+ mock_x2c.list_variables.return_value = []
+ mocker.patch(
+ "pyx2cscope.gui.qt.controllers.connection_manager.X2CScope",
+ return_value=mock_x2c,
+ )
+
+ result = connection_manager.connect_tcp(
+ host="127.0.0.1", tcp_port=12666, elf_file="variables.yml"
+ )
+
+ assert result is True
+ mock_x2c.import_variables.assert_called_once_with("variables.yml")
+
def test_disconnect_clears_state(self, connection_manager):
"""Test disconnect clears connection state."""
connection_manager.disconnect()
@@ -322,6 +369,25 @@ def test_setup_tab_creation(self, qt_application):
assert tab is not None
+ def test_app_state_exports_selected_variables(self, qt_application):
+ """Test AppState exports only variables selected in watch and scope views."""
+ from pyx2cscope.gui.qt.models.app_state import AppState, ScopeChannel
+
+ app_state = AppState()
+ mock_x2c = MagicMock()
+
+ mock_x2c.list_variables.return_value = ["watch.var", "scope.var"]
+
+ app_state.set_x2cscope(mock_x2c)
+ app_state.add_live_watch_var()
+ app_state.update_live_watch_var_field(0, "name", "watch.var")
+ app_state.set_scope_channel(0, ScopeChannel(name="scope.var", sfr=True))
+
+ app_state.export_selected_variables("selected.yml")
+
+ exported_items = mock_x2c.export_variables.call_args.kwargs["items"]
+ assert set(exported_items) == {("watch.var", False), ("scope.var", True)}
+
def test_scope_view_tab_creation(self, qt_application):
"""Test ScopeViewTab can be created."""
from pyx2cscope.gui.qt.models.app_state import AppState
@@ -388,6 +454,77 @@ def test_stop_polling(self, data_poller):
assert data_poller._running is False
+class TestCANConnectionManager:
+ """Tests for ConnectionManager CAN connect/disconnect cycle."""
+
+ elf_file = os.path.join(
+ os.path.dirname(data.__file__), "mc_foc_sl_fip_dspic33ck_mclv48v300w.elf"
+ )
+
+ def test_qt_connection_manager_disconnect_calls_x2cscope_disconnect(self, mock_can_bus):
+ """Test that Qt ConnectionManager properly calls x2cscope.disconnect()."""
+ from unittest.mock import patch
+
+ from PyQt5.QtCore import QCoreApplication
+
+ from pyx2cscope.gui.qt.controllers.connection_manager import ConnectionManager
+ from pyx2cscope.gui.qt.models.app_state import AppState
+
+ app = QCoreApplication.instance()
+ if app is None:
+ app = QCoreApplication([])
+
+ try:
+ _, mock_bus_instance = mock_can_bus
+
+ app_state = AppState()
+ conn_manager = ConnectionManager(app_state)
+
+ device_info_frame = bytearray(
+ b'\x55\x2E\x01\x11\x00'
+ b'\x01\x00'
+ b'\x01\x00'
+ b'\x10\x82'
+ b'01/01/2024'
+ b'1200'
+ b'01/01/2024'
+ b'1200'
+ b'\x01'
+ b'\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00\x00\x00\x00'
+ b'\x00'
+ )
+ load_param_frame = bytearray(b'\x55\x00\x01\x12\x00\x00')
+
+ with patch('mchplnet.interfaces.can.LNetCan.read') as mock_read, \
+ patch('mchplnet.interfaces.can.LNetCan.write'):
+ mock_read.side_effect = [device_info_frame, load_param_frame]
+
+ success = conn_manager.connect_can(
+ elf_file=self.elf_file,
+ bus_type="USB",
+ channel=1,
+ baudrate="500K",
+ mode="Standard",
+ tx_id="110",
+ rx_id="100",
+ )
+
+ assert success is True
+ assert app_state.is_connected()
+
+ conn_manager.disconnect()
+
+ assert not app_state.is_connected()
+ assert app_state.x2cscope is None
+ assert mock_bus_instance.shutdown.called
+
+ finally:
+ if app:
+ app.quit()
+
+
class TestSignalSlotConnections:
"""Tests for signal/slot connections."""
diff --git a/tests/test_web_gui.py b/tests/test_web_gui.py
index 69b44683..282709ae 100644
--- a/tests/test_web_gui.py
+++ b/tests/test_web_gui.py
@@ -16,6 +16,7 @@
# HTTP status codes for test assertions
HTTP_OK = 200
+HTTP_BAD_REQUEST = 400
HTTP_NOT_FOUND = 404
@@ -57,6 +58,7 @@ def test_index_route(self, flask_client):
assert response.status_code == HTTP_OK
assert b"html" in response.data.lower()
+ assert b"exportVariablesToggle" in response.data
def test_serial_ports_route(self, flask_client, mocker):
"""Test serial-ports route returns list."""
@@ -112,6 +114,85 @@ def test_variables_route_not_connected(self, flask_client):
assert "items" in data
assert data["items"] == []
+ def test_export_variables_route(self, flask_client, mocker):
+ """Test variable export route returns a downloadable file."""
+ from pyx2cscope.gui.web.scope import web_scope
+
+ mock_x2c = MagicMock()
+ original_x2c = web_scope.x2c_scope
+ original_file = web_scope.variables_file
+ original_watch = web_scope.watch_vars
+ original_scope = web_scope.scope_vars
+ original_dashboard = web_scope.dashboard_vars
+
+ def write_export_file(filename, ext, items=None):
+ mode = "wb" if ext.value == ".pkl" else "w"
+ payload = b"pickle-data" if mode == "wb" else "yaml-data"
+ with open(filename, mode) as file:
+ file.write(payload)
+
+ mock_x2c.export_variables.side_effect = write_export_file
+ mocker.patch.object(web_scope, "is_connected", return_value=True)
+ web_scope.x2c_scope = mock_x2c
+ web_scope.variables_file = "firmware.elf"
+ watch_var = MagicMock()
+ watch_var.info.name = "watch.var"
+ scope_var = MagicMock()
+ scope_var.info.name = "scope.var"
+ dashboard_var = MagicMock()
+ dashboard_var.info.name = "dashboard.var"
+ web_scope.watch_vars = [{"variable": watch_var, "sfr": True}]
+ web_scope.scope_vars = [{"variable": scope_var, "sfr": False}]
+ web_scope.dashboard_vars = {"dashboard.var": dashboard_var}
+
+ try:
+ response = flask_client.get("/variables/export?ext=yml")
+
+ assert response.status_code == HTTP_OK
+ assert response.headers["Content-Disposition"] == "attachment; filename=firmware.yml"
+ assert response.data == b"yaml-data"
+ assert mock_x2c.export_variables.call_args.kwargs["items"] == [
+ ("watch.var", True),
+ ("scope.var", False),
+ ("dashboard.var", False),
+ ]
+ finally:
+ web_scope.x2c_scope = original_x2c
+ web_scope.variables_file = original_file
+ web_scope.watch_vars = original_watch
+ web_scope.scope_vars = original_scope
+ web_scope.dashboard_vars = original_dashboard
+
+ def test_export_variables_route_requires_selected_variables(self, flask_client, mocker):
+ """Test export route fails when no view has selected variables."""
+ from pyx2cscope.gui.web.scope import web_scope
+
+ original_x2c = web_scope.x2c_scope
+ original_file = web_scope.variables_file
+ original_watch = web_scope.watch_vars
+ original_scope = web_scope.scope_vars
+ original_dashboard = web_scope.dashboard_vars
+
+ mocker.patch.object(web_scope, "is_connected", return_value=True)
+ web_scope.x2c_scope = MagicMock()
+ web_scope.variables_file = "firmware.elf"
+ web_scope.watch_vars = []
+ web_scope.scope_vars = []
+ web_scope.dashboard_vars = {}
+
+ try:
+ response = flask_client.get("/variables/export?ext=yml")
+
+ assert response.status_code == HTTP_BAD_REQUEST
+ data = json.loads(response.data)
+ assert "No variables are selected" in data["msg"]
+ finally:
+ web_scope.x2c_scope = original_x2c
+ web_scope.variables_file = original_file
+ web_scope.watch_vars = original_watch
+ web_scope.scope_vars = original_scope
+ web_scope.dashboard_vars = original_dashboard
+
class TestWatchViewRoutes:
"""Tests for watch view routes."""
@@ -313,7 +394,7 @@ def test_scope_trigger_defaults(self, web_scope):
def test_scope_sample_time_default(self, web_scope):
"""Test scope sample time default."""
- assert web_scope.scope_sample_time == 0
+ assert web_scope.scope_sample_time == 1
class TestWebScopeVariableManagement:
@@ -358,6 +439,13 @@ def test_add_watch_var(self, web_scope_connected):
assert result is not None
assert len(web_scope_connected.watch_vars) == 1
+ assert web_scope_connected.watch_vars[0]["sfr"] is False
+
+ def test_add_watch_var_tracks_sfr_flag(self, web_scope_connected):
+ """Test adding watch SFR variable preserves its SFR flag."""
+ web_scope_connected.add_watch_var("test_var", sfr=True)
+
+ assert web_scope_connected.watch_vars[0]["sfr"] is True
def test_add_watch_var_duplicate_prevented(self, web_scope_connected):
"""Test duplicate watch variables are not added."""
@@ -380,6 +468,13 @@ def test_add_scope_var(self, web_scope_connected):
assert result is not None
assert len(web_scope_connected.scope_vars) == 1
+ assert web_scope_connected.scope_vars[0]["sfr"] is False
+
+ def test_add_scope_var_tracks_sfr_flag(self, web_scope_connected):
+ """Test adding scope SFR variable preserves its SFR flag."""
+ web_scope_connected.add_scope_var("test_var", sfr=True)
+
+ assert web_scope_connected.scope_vars[0]["sfr"] is True
def test_add_scope_var_max_limit(self, web_scope_connected):
"""Test scope variables can be added (max limit may not be enforced in WebScope)."""
@@ -405,6 +500,16 @@ def test_remove_scope_var(self, web_scope_connected):
web_scope_connected.remove_scope_var("test_var")
assert len(web_scope_connected.scope_vars) == 0
+ def test_scope_sample_time_uses_user_facing_values(self, web_scope_connected):
+ """Test web scope clamps sample time to user-facing 1-based values."""
+ web_scope_connected.x2c_scope.get_scope_sample_time.return_value = 2.5
+ web_scope_connected.scope_sample_time = 2
+
+ web_scope_connected.scope_set_sample("off", 0, 20)
+
+ assert web_scope_connected.scope_sample_time == 1
+ web_scope_connected.x2c_scope.set_sample_time.assert_called_once_with(1)
+
class TestWebScopeScaledValue:
"""Tests for WebScope scaled value calculation."""
diff --git a/tests/utils/serial_stub.py b/tests/utils/serial_stub.py
index 0e329028..30912067 100644
--- a/tests/utils/serial_stub.py
+++ b/tests/utils/serial_stub.py
@@ -9,6 +9,10 @@
BIT_LENGTH_16 = 16
BIT_LENGTH_32 = 32
+DEVICE_PROFILE_DSPIC = "dspic"
+DEVICE_PROFILE_ARM = "arm"
+DEVICE_PROFILE_PIC32 = "pic32"
+DEVICE_PROFILE_DSPIC33A = "dspic33a"
class FrameBuilder(LNetFrame):
"""FrameBuilder class to build LNet frames."""
@@ -66,13 +70,16 @@ def _check_frame_protocol(self):
class SerialStub:
"""Fakes a serial connection for 16 and 32 bit devices."""
- def __init__(self, uc_width=BIT_LENGTH_16):
+ def __init__(self, uc_width=BIT_LENGTH_16, device_profile=None):
"""Constructor of the SerialStub class.
Expected get_device_info and load_param bytestream are initialized here.
"""
self.delay_seconds = 0.2 # 200ms delay to make concurrency
self.uc_width = uc_width
+ self.device_profile = device_profile or (
+ DEVICE_PROFILE_DSPIC if uc_width == BIT_LENGTH_16 else DEVICE_PROFILE_PIC32
+ )
self.data = bytearray()
self.device_info = bytearray(b"\x55\x01\x01\x00\x57")
self.load_param = bytearray(b"\x55\x03\x01\x11\x01\x00\x6b")
@@ -85,9 +92,9 @@ def __init__(self, uc_width=BIT_LENGTH_16):
def lnet_serial_start(self):
"""Mocker for the start interface function.
- As we are dealing with no real device, we just return.
+ As we are dealing with no real device, we just return True.
"""
- return
+ return True
def _get_ram(self):
"""Handle a get_ram request."""
@@ -146,18 +153,27 @@ def lnet_serial_read(self) -> bytearray:
def _get_device_info(self):
if self.data != self.device_info:
raise ValueError("Wrong Device Info package format!")
- if self.uc_width == BIT_LENGTH_16:
+ if self.device_profile == DEVICE_PROFILE_DSPIC:
return bytearray(
b"\x55.\x01\x00\x00\x05\x00\x01\x00\xff\x10\x82Nov2320231500\x00\x00\x00\x00"
b"\x00\x00\xccJ\x14K\xc0\xff\xff\x01\x00\x00\x00\x00\x00\x00\xf4R\x00\x00\xba"
)
- elif self.uc_width == BIT_LENGTH_32:
+ if self.device_profile == DEVICE_PROFILE_PIC32:
return bytearray(
b"\x55.\x01\x00\x00\x05\x00\x01\x00\xff \x82Mar 320191220Mar 320191220\x01"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00 T"
)
- else:
- raise ValueError("Mocker fake_serial: wrong uC width! should be either 16 or 32 bits")
+ if self.device_profile == DEVICE_PROFILE_ARM:
+ return bytearray(
+ b"\x55.\x01\x00\x00\x05\x00\x01\x00\xff\x10\x83Mar 320191220Mar 320191220\x01"
+ b"\x00\x00\x00\x00\x00\x00\x00\x00\x00 T"
+ )
+ if self.device_profile == DEVICE_PROFILE_DSPIC33A:
+ return bytearray(
+ b"\x55.\x01\x00\x00\x05\x00\x01\x00\xff@\x82Apr 120241200Apr 120241200\x01"
+ b"\x00\x00\x00\x00\x00\x00\x00\x00\x00 T"
+ )
+ raise ValueError("Mocker fake_serial: unsupported device profile")
def _get_load_param(self) -> bytearray:
if self.data != self.load_param:
@@ -175,7 +191,7 @@ def _get_load_param(self) -> bytearray:
else:
raise ValueError("Mocker fake_serial: wrong uC width! should be either 16 or 32 bits")
-def fake_serial(mocker, uc_width=BIT_LENGTH_16):
+def fake_serial(mocker, uc_width=BIT_LENGTH_16, device_profile=None):
"""Fakes a serial port for 16/32 bit devices.
The methods being faked by this function are start, read and write.
@@ -183,10 +199,12 @@ def fake_serial(mocker, uc_width=BIT_LENGTH_16):
Args:
mocker: Mocker inheritance
uc_width: bit size of the device to be mocked
+ device_profile: target profile to be mocked
Return:
- None
+ SerialStub: the configured serial stub instance
"""
- serial_stub = SerialStub(uc_width=uc_width)
+ serial_stub = SerialStub(uc_width=uc_width, device_profile=device_profile)
mocker.patch.object(LNetSerial, "start", serial_stub.lnet_serial_start)
mocker.patch.object(LNetSerial, "write", serial_stub.lnet_serial_write)
mocker.patch.object(LNetSerial, "read", serial_stub.lnet_serial_read)
+ return serial_stub