diff --git a/LICENSE.txt b/LICENSE.txt index bfce6f06..db5094ed 100644 --- a/LICENSE.txt +++ b/LICENSE.txt @@ -1,6 +1,7 @@ MIT License Copyright (c) 2016 Christian Sandberg +Copyright (c) 2025 Svein Seldal Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal diff --git a/README.rst b/README.rst index e01ac668..ce5b8d6c 100644 --- a/README.rst +++ b/README.rst @@ -8,6 +8,85 @@ automation tasks rather than a standard compliant master implementation. The library supports Python 3.9 or newer. +The library can be run in two modes: regular mode or in async mode. In regular +mode, calls to the library may block until a response is ready. In async mode +it is possible to read and write to more than one node at the same time +without the need of multiple threads. + + +Asyncio port +------------ + +The objective of the library is to provide a canopen implementation in +either async or non-async environment, with suitable API for both. + +To minimize the impact of the async changes, this port is designed to use the +existing synchronous backend of the library. This means that the library +uses :code:`asyncio.to_thread()` for many asynchronous operations. + +This port remains compatible with using it in a regular non-asyncio +environment. This is selected with the `loop` parameter in the +:code:`Network` constructor. If you pass a valid asyncio event loop, the +library will run in async mode. If you pass `loop=None`, it will run in +regular blocking mode. It cannot be used in both modes at the same time. + + +Difference between async and non-async version +---------------------------------------------- + +This port have some differences with the upstream non-async version of canopen. + +* The :code:`Network` accepts additional parameters than upstream. It accepts + :code:`loop` which selects the mode of operation. If :code:`None` it will + run in blocking mode, otherwise it will run in async mode. It supports + providing a custom CAN :code:`notifier` if the CAN bus will be shared by + multiple protocols. + +* The :code:`Network` class can be (and should be) used in an async context + manager. This will ensure the network will be automatically disconnected when + exiting the context. See the example below. + +* Most async functions follow an "a" prefix naming scheme. + E.g. the async variant for :code:`SdoClient.download()` is available + as :code:`SdoClient.adownload()`. + +* Variables in the regular canopen library uses properties for getting and + setting. This is replaced with awaitable methods in the async version. + + var = sdo['Variable'].raw # synchronous + sdo['Variable'].raw = 12 # synchronous + + var = await sdo['Variable'].get_raw() # async + await sdo['Variable'].set_raw(12) # async + +* Installed :code:`ensure_not_async()` sentinel guard in functions which + prevents calling blocking functions in async context. It will raise the + exception :code:`RuntimeError` "Calling a blocking function" when this + happen. If this is encountered, it is likely that the code is not using the + async variants of the library. + +* The mechanism for CAN bus callbacks have been changed. Callbacks might be + async, which means they cannot be called immediately. This affects how + error handling is done in the library. + +* The callbacks to the message handlers have been changed to be handled by + :code:`Network.dispatch_callbacks()`. They are no longer called with any + locks held, as this would not work with async. This affects: + * :code:`PdoMaps.on_message` + * :code:`EmcyConsumer.on_emcy` + * :code:`NtmMaster.on_heartbaet` + +* SDO block upload and download is not yet supported in async mode. + +* :code:`ODVariable.__len__()` returns 64 bits instead of 8 bits to support + truncated 24-bits integers, see #436 + +* :code:`BaseNode402` does not work with async + +* :code:`LssMaster` does not work with async, except :code:`LssMaster.fast_scan()` + +* :code:`Bits` is not working in async + Features -------- @@ -156,6 +235,70 @@ The :code:`n` is the PDO index (normally 1 to 4). The second form of access is f network.disconnect() +Asyncio +------- + +This is the same example as above, but using asyncio + +.. code-block:: python + + import asyncio + import canopen + import can + + async def my_node(network, nodeid, od): + + # Create the node object and load the OD + node = network.add_node(nodeid, od) + + # Read the PDOs from the remote + await node.tpdo.aread() + await node.rpdo.aread() + + # Set the module state + node.nmt.set_state('OPERATIONAL') + + # Set motor speed via SDO + await node.sdo['MotorSpeed'].aset_raw(2) + + while True: + + # Wait for TPDO 1 + t = await node.tpdo[1].await_for_reception(1) + if not t: + continue + + # Get the TPDO 1 value + rpm = node.tpdo[1]['MotorSpeed Actual'].get_raw() + print(f'SPEED on motor {nodeid}:', rpm) + + # Sleep a little + await asyncio.sleep(0.2) + + # Send RPDO 1 with some data + node.rpdo[1]['Some variable'].set_phys(42) + node.rpdo[1].transmit() + + async def main(): + + # Connect to the CAN bus + # Arguments are passed to python-can's can.Bus() constructor + # (see https://python-can.readthedocs.io/en/latest/bus.html). + # Note the loop parameter to enable asyncio operation + loop = asyncio.get_running_loop() + async with canopen.Network(loop=loop).connect( + interface='pcan', bitrate=1000000) as network: + + # Create two independent tasks for two nodes 51 and 52 which will run concurrently + task1 = asyncio.create_task(my_node(network, 51, '/path/to/object_dictionary.eds')) + task2 = asyncio.create_task(my_node(network, 52, '/path/to/object_dictionary.eds')) + + # Wait for both to complete (which will never happen) + await asyncio.gather((task1, task2)) + + asyncio.run(main()) + + Debugging --------- diff --git a/canopen/async_guard.py b/canopen/async_guard.py new file mode 100644 index 00000000..c2c523c8 --- /dev/null +++ b/canopen/async_guard.py @@ -0,0 +1,49 @@ +""" Utils for async """ +import functools +import logging +import threading +import traceback + + +_ASYNC_SENTINELS: dict[int, bool] = {} +"""Per-thread boolean indicating allowance of running blocking functions. + +Any function that is guarded with the :code:`@ensure_not_async` decorator will +fail with a :code:`RuntimeError` if called if True for this thread. The +application sets this value to True using :code:`set_async_sentinel(True)` +when it is running async code. +""" + +logger = logging.getLogger(__name__) + + +def set_async_sentinel(enable: bool): + """ Register a function to validate if async is running """ + _ASYNC_SENTINELS[threading.get_ident()] = enable + + +def ensure_not_async(fn): + """ Decorator that will ensure that the function is not called if async + is running. + """ + @functools.wraps(fn) + def async_guard_wrap(*args, **kwargs): + if _ASYNC_SENTINELS.get(threading.get_ident(), False): + st = "".join(traceback.format_stack()) + logger.debug("Traceback:\n%s", st.rstrip()) + raise RuntimeError(f"Calling a blocking function, {fn.__qualname__}() in {fn.__code__.co_filename}:{fn.__code__.co_firstlineno}, while running async") + return fn(*args, **kwargs) + return async_guard_wrap + + +class AllowBlocking: + """ Context manager to pause async guard """ + def __init__(self): + self._enabled = _ASYNC_SENTINELS.get(threading.get_ident(), False) + + def __enter__(self): + set_async_sentinel(False) + return self + + def __exit__(self, exc_type, exc_value, traceback): + set_async_sentinel(self._enabled) diff --git a/canopen/emcy.py b/canopen/emcy.py index 8b7d3bff..57a1b6b5 100644 --- a/canopen/emcy.py +++ b/canopen/emcy.py @@ -1,11 +1,13 @@ from __future__ import annotations +import asyncio import logging import struct import threading import time from typing import Callable, Optional +from canopen.async_guard import ensure_not_async import canopen.network @@ -24,7 +26,9 @@ def __init__(self): self.active: list[EmcyError] = [] self.callbacks = [] self.emcy_received = threading.Condition() + self.network: canopen.network.Network = canopen.network._UNINITIALIZED_NETWORK + @ensure_not_async def on_emcy(self, can_id, data, timestamp): code, register, data = EMCY_STRUCT.unpack(data) entry = EmcyError(code, register, data, timestamp) @@ -38,11 +42,8 @@ def on_emcy(self, can_id, data, timestamp): self.log.append(entry) self.emcy_received.notify_all() - for callback in self.callbacks: - try: - callback(entry) - except Exception: - logger.exception("Exception in EMCY callback") + # Call all registered callbacks + self.network.dispatch_callbacks(self.callbacks, entry) def add_callback(self, callback: Callable[[EmcyError], None]): """Get notified on EMCY messages from this node. @@ -58,6 +59,7 @@ def reset(self): self.log = [] self.active = [] + @ensure_not_async def wait( self, emcy_code: Optional[int] = None, timeout: float = 10 ) -> Optional[EmcyError]: @@ -86,6 +88,18 @@ def wait( # This is the one we're interested in return emcy + async def async_wait( + self, emcy_code: Optional[int] = None, timeout: float = 10 + ) -> EmcyError: + """Wait for a new EMCY to arrive. + + :param emcy_code: EMCY code to wait for + :param timeout: Max time in seconds to wait + + :return: The EMCY exception object or None if timeout + """ + return await asyncio.to_thread(self.wait, emcy_code, timeout) + class EmcyProducer: diff --git a/canopen/lss.py b/canopen/lss.py index 311f77b5..4dab8748 100644 --- a/canopen/lss.py +++ b/canopen/lss.py @@ -1,8 +1,10 @@ +import asyncio import logging import queue import struct import time +from canopen.async_guard import ensure_not_async import canopen.network @@ -241,6 +243,7 @@ def send_identify_non_configured_remote_slave(self): message[0] = CS_IDENTIFY_NON_CONFIGURED_REMOTE_SLAVE self.__send_command(message) + @ensure_not_async def fast_scan(self): """This command sends a series of fastscan message to find unconfigured slave with lowest number of LSS idenities @@ -282,6 +285,10 @@ def fast_scan(self): return False, None + async def afast_scan(self): + """Asynchronous version of fast_scan""" + return await asyncio.to_thread(self.fast_scan) + def __send_fast_scan_message(self, id_number, bit_checker, lss_sub, lss_next): message = bytearray(8) message[0:8] = struct.pack(' None: @@ -108,7 +120,9 @@ def connect(self, *args, **kwargs) -> Network: self.bus = can.Bus(*args, **kwargs) logger.info("Connected to '%s'", self.bus.channel_info) if self.notifier is None: - self.notifier = can.Notifier(self.bus, self.listeners, self.NOTIFIER_CYCLE) + self.notifier = can.Notifier(self.bus, [], self.NOTIFIER_CYCLE) + for listener in self.listeners: + self.notifier.add_listener(listener) return self def disconnect(self) -> None: @@ -130,12 +144,25 @@ def disconnect(self) -> None: # Release notifier after check self.notifier = None + # Remove the async sentinel + set_async_sentinel(False) + def __enter__(self): return self def __exit__(self, type, value, traceback): self.disconnect() + async def __aenter__(self): + # FIXME: When TaskGroup are available, we should use them to manage the + # tasks. The user must use the `async with` statement with the Network + # to ensure its created. + return self + + async def __aexit__(self, type, value, traceback): + self.disconnect() + + @ensure_not_async def add_node( self, node: Union[int, RemoteNode, LocalNode], @@ -165,6 +192,20 @@ def add_node( self[node.id] = node return node + async def aadd_node( + self, + node: Union[int, RemoteNode, LocalNode], + object_dictionary: Union[str, ObjectDictionary, None] = None, + upload_eds: bool = False, + ) -> RemoteNode: + """Add a remote node to the network, async variant. + + See add_node() for description + """ + # The async variant exists because import_from_node might block + return await asyncio.to_thread(self.add_node, node, + object_dictionary, upload_eds) + def create_node( self, node: Union[int, LocalNode], @@ -247,11 +288,44 @@ def notify(self, can_id: int, data: bytearray, timestamp: float) -> None: Timestamp of the message, preferably as a Unix timestamp """ if can_id in self.subscribers: - callbacks = self.subscribers[can_id] - for callback in callbacks: - callback(can_id, data, timestamp) + self.dispatch_callbacks(self.subscribers[can_id], can_id, data, timestamp) self.scanner.on_message_received(can_id) + def on_error(self, exc: BaseException) -> None: + """This method is called to handle any exception in the callbacks.""" + + # Exceptions in any callbaks should not affect CAN processing + logger.exception("Exception in callback: %s", exc_info=exc) + + def dispatch_callbacks(self, callbacks: list[Callback], *args) -> None: + """Dispatch a list of callbacks with the given arguments. + + :param callbacks: + List of callbacks to call + :param args: + Arguments to pass to the callbacks + """ + def task_done(future: asyncio.Future) -> None: + """Callback to be called when a task is done.""" + self._futures.discard(future) + try: + if (exc := future.exception()) is not None: + self.on_error(exc) + except (asyncio.CancelledError, asyncio.InvalidStateError) as exc: + # Handle cancelled tasks and unfinished tasks gracefully + self.on_error(exc) + + # Run the callbacks + for callback in callbacks: + result = callback(*args) + if result is not None and self.loop is not None and asyncio.iscoroutine(result): + # This function may be called from the rx thread, so it must + # be thread-safe. We cannot use asyncio.create_task() here, since + # it is not thread-safe. + task = asyncio.run_coroutine_threadsafe(result, self.loop) + self._futures.add(task) + task.add_done_callback(task_done) + def check(self) -> None: """Check that no fatal error has occurred in the receiving thread. @@ -264,6 +338,11 @@ def check(self) -> None: logger.error("An error has caused receiving of messages to stop") raise exc + @property + def is_async(self) -> bool: + """Check if canopen has been connected with async""" + return self.loop is not None + def __getitem__(self, node_id: int) -> Union[RemoteNode, LocalNode]: return self.nodes[node_id] @@ -374,7 +453,7 @@ def on_message_received(self, msg): self.network.notify(msg.arbitration_id, msg.data, msg.timestamp) except Exception as e: # Exceptions in any callbaks should not affect CAN processing - logger.error(str(e)) + self.network.on_error(e) def stop(self) -> None: """Override abstract base method to release any resources.""" diff --git a/canopen/nmt.py b/canopen/nmt.py index 77d56910..41a4ff8b 100644 --- a/canopen/nmt.py +++ b/canopen/nmt.py @@ -1,9 +1,11 @@ +import asyncio import logging import struct import threading import time from typing import Callable, Final, Optional, TYPE_CHECKING +from canopen.async_guard import ensure_not_async import canopen.network if TYPE_CHECKING: @@ -119,6 +121,7 @@ def __init__(self, node_id: int): self.state_update = threading.Condition() self._callbacks: list[Callable[[int], None]] = [] + @ensure_not_async def on_heartbeat(self, can_id, data, timestamp): new_state, = struct.unpack_from("B", data) # Mask out toggle bit @@ -135,8 +138,8 @@ def on_heartbeat(self, can_id, data, timestamp): self._state_received = new_state self.state_update.notify_all() - for callback in self._callbacks: - callback(new_state) + # Call all registered callbacks + self.network.dispatch_callbacks(self._callbacks, new_state) def send_command(self, code: int): """Send an NMT command code to the node. @@ -149,6 +152,7 @@ def send_command(self, code: int): "Sending NMT command 0x%X to node %d", code, self.id) self.network.send_message(0, [code, self.id]) + @ensure_not_async def wait_for_heartbeat(self, timeout: float = 10): """Wait until a heartbeat message is received.""" with self.state_update: @@ -158,6 +162,11 @@ def wait_for_heartbeat(self, timeout: float = 10): raise NmtError("No boot-up or heartbeat received") return self.state + async def await_for_heartbeat(self, timeout: float = 10): + """Wait until a heartbeat message is received.""" + return await asyncio.to_thread(self.wait_for_heartbeat, timeout) + + @ensure_not_async def wait_for_bootup(self, timeout: float = 10) -> None: """Wait until a boot-up message is received.""" end_time = time.time() + timeout @@ -171,6 +180,10 @@ def wait_for_bootup(self, timeout: float = 10) -> None: if self._state_received == 0: break + async def await_for_bootup(self, timeout: float = 10) -> None: + """Wait until a boot-up message is received.""" + return await asyncio.to_thread(self.wait_for_bootup, timeout) + def add_heartbeat_callback(self, callback: Callable[[int], None]): """Add function to be called on heartbeat reception. @@ -230,11 +243,12 @@ def send_command(self, code: int) -> None: # The heartbeat service should start on the transition # between INITIALIZING and PRE-OPERATIONAL state if old_state == 0 and self._state == 127: - try: + # FIXME: Document why this was fixed + if self._heartbeat_time_ms == 0: heartbeat_time_ms = self._local_node.sdo[0x1017].raw - self.start_heartbeat(heartbeat_time_ms) - except KeyError: - pass + else: + heartbeat_time_ms = self._heartbeat_time_ms + self.start_heartbeat(heartbeat_time_ms) else: self.update_heartbeat() diff --git a/canopen/node/remote.py b/canopen/node/remote.py index b5e4e6b1..5fec0ae4 100644 --- a/canopen/node/remote.py +++ b/canopen/node/remote.py @@ -59,6 +59,7 @@ def associate_network(self, network: canopen.network.Network): self.tpdo.network = network self.rpdo.network = network self.nmt.network = network + self.emcy.network = network for sdo in self.sdo_channels: network.subscribe(sdo.tx_cobid, sdo.on_response) network.subscribe(0x700 + self.id, self.nmt.on_heartbeat) @@ -79,6 +80,7 @@ def remove_network(self) -> None: self.tpdo.network = canopen.network._UNINITIALIZED_NETWORK self.rpdo.network = canopen.network._UNINITIALIZED_NETWORK self.nmt.network = canopen.network._UNINITIALIZED_NETWORK + self.emcy.network = canopen.network._UNINITIALIZED_NETWORK def add_sdo(self, rx_cobid, tx_cobid): """Add an additional SDO channel. diff --git a/canopen/objectdictionary/eds.py b/canopen/objectdictionary/eds.py index 608024f3..cb6f8827 100644 --- a/canopen/objectdictionary/eds.py +++ b/canopen/objectdictionary/eds.py @@ -1,11 +1,13 @@ from __future__ import annotations +import asyncio import copy import logging import re from configparser import NoOptionError, NoSectionError, RawConfigParser from typing import Any, TYPE_CHECKING +from canopen.async_guard import ensure_not_async from canopen.objectdictionary import ( ODArray, ODRecord, @@ -184,6 +186,7 @@ def import_eds(source, node_id): return od +@ensure_not_async def import_from_node(node_id: int, network: canopen.network.Network): """ Download the configuration from the remote node :param int node_id: Identifier of the node @@ -207,6 +210,14 @@ def import_from_node(node_id: int, network: canopen.network.Network): return od +async def aimport_from_node(node_id: int, network: canopen.network.Network): + """ Download the configuration from the remote node, async variant + :param int node_id: Identifier of the node + :param network: network object + """ + return await asyncio.to_thread(import_from_node, node_id, network) + + def _calc_bit_length(data_type: int) -> int: if data_type in datatypes.INTEGER_TYPES: st = ODVariable.STRUCT_TYPES[data_type] diff --git a/canopen/pdo/base.py b/canopen/pdo/base.py index f9973882..edd5daf7 100644 --- a/canopen/pdo/base.py +++ b/canopen/pdo/base.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import binascii import contextlib import logging @@ -11,6 +12,7 @@ import canopen.network from canopen import objectdictionary from canopen import variable +from canopen.async_guard import ensure_not_async from canopen.sdo import SdoAbortedError if TYPE_CHECKING: @@ -60,16 +62,28 @@ def __getitem__(self, key: Union[int, str]): def __len__(self): return len(self.map) + @ensure_not_async def read(self, from_od=False): """Read PDO configuration from node using SDO.""" for pdo_map in self.map.values(): pdo_map.read(from_od=from_od) + async def aread(self, from_od=False): + """Read PDO configuration from node using SDO, async variant.""" + for pdo_map in self.map.values(): + await pdo_map.aread(from_od=from_od) + + @ensure_not_async def save(self): """Save PDO configuration to node using SDO.""" for pdo_map in self.map.values(): pdo_map.save() + async def asave(self): + """Save PDO configuration to node using SDO, async variant.""" + for pdo_map in self.map.values(): + await pdo_map.asave() + def subscribe(self): """Register the node's PDOs for reception on the network. @@ -328,6 +342,7 @@ def is_periodic(self) -> bool: # Unknown transmission type, assume non-periodic return False + @ensure_not_async def on_message(self, can_id, data, timestamp): is_transmitting = self._task is not None if can_id == self.cob_id and not is_transmitting: @@ -338,8 +353,9 @@ def on_message(self, can_id, data, timestamp): self.period = timestamp - self.timestamp self.timestamp = timestamp self.receive_condition.notify_all() - for callback in self.callbacks: - callback(self) + + # Call all registered callbacks + self.pdo_node.network.dispatch_callbacks(self.callbacks, self) def add_callback(self, callback: Callable[[PdoMap], None]) -> None: """Add a callback which will be called on receive. @@ -350,58 +366,48 @@ def add_callback(self, callback: Callable[[PdoMap], None]) -> None: """ self.callbacks.append(callback) - def read(self, from_od=False) -> None: - """Read PDO configuration for this map. - - :param from_od: - Read using SDO if False, read from object dictionary if True. - When reading from object dictionary, if DCF populated a value, the - DCF value will be used, otherwise the EDS default will be used instead. - """ + def read_generator(self): + """Generator to run through steps for reading the PDO configuration + for this map. - def _raw_from(param): - if from_od: - if param.od.value is not None: - return param.od.value - else: - return param.od.default - return param.raw + This function does not do any io. This must be done by the caller. - cob_id = _raw_from(self.com_record[1]) + """ + cob_id = yield self.com_record[1] self.cob_id = cob_id & 0x1FFFFFFF logger.info("COB-ID is 0x%X", self.cob_id) self.enabled = cob_id & PDO_NOT_VALID == 0 logger.info("PDO is %s", "enabled" if self.enabled else "disabled") self.rtr_allowed = cob_id & RTR_NOT_ALLOWED == 0 logger.info("RTR is %s", "allowed" if self.rtr_allowed else "not allowed") - self.trans_type = _raw_from(self.com_record[2]) + self.trans_type = yield self.com_record[2] logger.info("Transmission type is %d", self.trans_type) if self.trans_type >= 254: try: - self.inhibit_time = _raw_from(self.com_record[3]) + self.inhibit_time = yield self.com_record[3] except (KeyError, SdoAbortedError) as e: logger.info("Could not read inhibit time (%s)", e) else: logger.info("Inhibit time is set to %d ms", self.inhibit_time) try: - self.event_timer = _raw_from(self.com_record[5]) + self.event_timer = yield self.com_record[5] except (KeyError, SdoAbortedError) as e: logger.info("Could not read event timer (%s)", e) else: logger.info("Event timer is set to %d ms", self.event_timer) try: - self.sync_start_value = _raw_from(self.com_record[6]) + self.sync_start_value = yield self.com_record[6] except (KeyError, SdoAbortedError) as e: logger.info("Could not read SYNC start value (%s)", e) else: logger.info("SYNC start value is set to %d ms", self.sync_start_value) self.clear() - nof_entries = _raw_from(self.map_array[0]) + nof_entries = yield self.map_array[0] for subindex in range(1, nof_entries + 1): - value = _raw_from(self.map_array[subindex]) + value = yield self.map_array[subindex] index = value >> 16 subindex = (value >> 8) & 0xFF # Ignore the highest bit, it is never valid for <= 64 PDO length @@ -416,13 +422,70 @@ def _raw_from(param): self.subscribe() - def save(self) -> None: - """Save PDO configuration for this map using SDO.""" + @ensure_not_async + def read(self, from_od=False) -> None: + """Read PDO configuration for this map. + + :param from_od: + Read using SDO if False, read from object dictionary if True. + When reading from object dictionary, if DCF populated a value, the + DCF value will be used, otherwise the EDS default will be used instead. + """ + gen = self.read_generator() + param = next(gen) + while param: + if from_od: + # Use value from OD + if param.od.value is not None: + value = param.od.value + else: + value = param.od.default + else: + # Get value from SDO + value = param.raw + try: + # Deliver value into read_generator and wait for next object + param = gen.send(value) + except StopIteration: + break + + async def aread(self, from_od=False) -> None: + """Read PDO configuration for this map. Async variant. + + :param from_od: + Read using SDO if False, read from object dictionary if True. + When reading from object dictionary, if DCF populated a value, the + DCF value will be used, otherwise the EDS default will be used instead. + """ + gen = self.read_generator() + param = next(gen) + while param: + if from_od: + # Use value from OD + if param.od.value is not None: + value = param.od.value + else: + value = param.od.default + else: + # Get value from SDO + value = await param.aget_raw() + try: + param = gen.send(value) + except StopIteration: + break + + def save_generator(self): + """Generator to run through steps for saving the PDO configuration + using SDO. + + This function does not do any io. This must be done by the caller. + + """ if self.cob_id is None: logger.info("Skip saving %s: COB-ID was never set", self.com_record.od.name) return logger.info("Setting COB-ID 0x%X and temporarily disabling PDO", self.cob_id) - self.com_record[1].raw = ( + yield self.com_record[1], ( self.cob_id | PDO_NOT_VALID | (RTR_NOT_ALLOWED if not self.rtr_allowed else 0) @@ -435,23 +498,31 @@ def _set_com_record( return if self.com_record[subindex].writable: logger.info(f"Setting {log_fmt}", value * log_factor) - self.com_record[subindex].raw = value + return self.com_record[subindex], value else: logger.info(f"Cannot set {log_fmt}, not writable", value * log_factor) - _set_com_record(2, self.trans_type, "transmission type to %d") - _set_com_record(3, self.inhibit_time, "inhibit time to %d us", 100) - _set_com_record(5, self.event_timer, "event timer to %d ms") - _set_com_record(6, self.sync_start_value, "SYNC start value to %d") + for result in ( + _set_com_record(2, self.trans_type, "transmission type to %d"), + _set_com_record(3, self.inhibit_time, "inhibit time to %d us", 100), + _set_com_record(5, self.event_timer, "event timer to %d ms"), + _set_com_record(6, self.sync_start_value, "SYNC start value to %d"), + ): + # Each call might return None, which is not a usable result. + if result is not None: + yield result try: - self.map_array[0].raw = 0 + yield self.map_array[0], 0 except SdoAbortedError: # WORKAROUND for broken implementations: If the array has a # fixed number of entries (count not writable), generate dummy # mappings for an invalid object 0x0000:00 to overwrite any # excess entries with all-zeros. - self._fill_map(self.map_array[0].raw) + # + # The '@@fill_map' yield will run + # self._fill_map(self.map_array[0].raw()) + yield self.map_array[0], '@@fill_map' for var, entry in zip(self.map, self.map_array.values()): if not entry.od.writable: continue @@ -464,11 +535,11 @@ def _set_com_record( ) if getattr(self.pdo_node.node, "curtis_hack", False): # Curtis HACK: mixed up field order - entry.raw = var.index | var.subindex << 16 | var.length << 24 + yield entry, var.index | var.subindex << 16 | var.length << 24 else: - entry.raw = var.index << 16 | var.subindex << 8 | var.length + yield entry, var.index << 16 | var.subindex << 8 | var.length try: - self.map_array[0].raw = len(self.map) + yield self.map_array[0], len(self.map) except SdoAbortedError as e: # WORKAROUND for broken implementations: If the array # number-of-entries parameter is not writable, we have already @@ -482,9 +553,26 @@ def _set_com_record( if self.enabled: cob_id = self.cob_id | (RTR_NOT_ALLOWED if not self.rtr_allowed else 0x0) logger.info("Setting COB-ID 0x%X and re-enabling PDO", cob_id) - self.com_record[1].raw = cob_id + yield self.com_record[1], cob_id self.subscribe() + @ensure_not_async + def save(self) -> None: + """Read PDO configuration for this map using SDO.""" + for sdo, value in self.save_generator(): + if value == '@@fillmap': + self._fill_map(sdo.raw) + else: + sdo.raw = value + + async def asave(self) -> None: + """Read PDO configuration for this map using SDO, async variant.""" + for sdo, value in self.save_generator(): + if value == '@@fillmap': + self._fill_map(await sdo.aget_raw()) + else: + await sdo.aset_raw(value) + def subscribe(self) -> None: """Register the PDO for reception on the network. @@ -592,6 +680,7 @@ def remote_request(self) -> None: if self.enabled and self.rtr_allowed and self.cob_id: self.pdo_node.network.send_message(self.cob_id, bytes(), remote=True) + @ensure_not_async def wait_for_reception(self, timeout: float = 10) -> float: """Wait for the next transmit PDO. @@ -603,6 +692,14 @@ def wait_for_reception(self, timeout: float = 10) -> float: self.receive_condition.wait(timeout) return self.timestamp if self.is_received else None + async def await_for_reception(self, timeout: float = 10) -> float: + """Wait for the next transmit PDO. + + :param float timeout: Max time to wait in seconds. + :return: Timestamp of message received or None if timeout. + """ + return await asyncio.to_thread(self.wait_for_reception, timeout) + class PdoVariable(variable.Variable): """One object dictionary variable mapped to a PDO.""" @@ -642,6 +739,11 @@ def get_data(self) -> bytes: return data + async def aget_data(self) -> bytes: + # Since get_data() is not making any IO, it can be called + # directly with no special async variant + return self.get_data() + def set_data(self, data: bytes): """Set for the given variable the PDO data. @@ -675,6 +777,11 @@ def set_data(self, data: bytes): self.pdo_parent.update() + async def aset_data(self, data: bytes): + # Since get_data() is not making any IO, it can be called + # directly with no special async variant + return self.set_data(data) + # For compatibility Variable = PdoVariable diff --git a/canopen/profiles/p402.py b/canopen/profiles/p402.py index fd4059fe..35a1909a 100644 --- a/canopen/profiles/p402.py +++ b/canopen/profiles/p402.py @@ -2,10 +2,19 @@ import logging import time +from canopen.async_guard import ensure_not_async from canopen.node import RemoteNode from canopen.pdo import PdoMap from canopen.sdo import SdoCommunicationError +""" +NOTE: Async compatibility +This file is not async compatible, as it contains numerous setters and getters +in many of its function. The BaseNode402 class should probably be refactored +and ported to a design which is async compatible. For now, "ensure_not_async" +guard is installed in its init function to warn the user not to use it. +""" + logger = logging.getLogger(__name__) @@ -211,6 +220,8 @@ class BaseNode402(RemoteNode): TIMEOUT_CHECK_TPDO = 0.2 # seconds TIMEOUT_HOMING_DEFAULT = 30 # seconds + # As the class is not async compatible, ensure async is not running + @ensure_not_async def __init__(self, node_id, object_dictionary): super(BaseNode402, self).__init__(node_id, object_dictionary) self.tpdo_values = {} # { index: value from last received TPDO } diff --git a/canopen/sdo/base.py b/canopen/sdo/base.py index d0088ec4..511e1e49 100644 --- a/canopen/sdo/base.py +++ b/canopen/sdo/base.py @@ -7,6 +7,7 @@ import canopen.network from canopen import objectdictionary from canopen import variable +from canopen.async_guard import ensure_not_async from canopen.utils import pretty_index @@ -84,6 +85,9 @@ def get_variable( def upload(self, index: int, subindex: int) -> bytes: raise NotImplementedError() + async def aupload(self, index: int, subindex: int) -> bytes: + raise NotImplementedError() + def download( self, index: int, @@ -93,6 +97,15 @@ def download( ) -> None: raise NotImplementedError() + async def adownload( + self, + index: int, + subindex: int, + data: bytes, + force_segment: bool = False, + ) -> None: + raise NotImplementedError() + class SdoRecord(Mapping): @@ -110,10 +123,20 @@ def __iter__(self) -> Iterator[int]: # Skip the "highest subindex" entry, which is not part of the data return filter(None, iter(self.od)) + async def aiter(self): + for i in iter(self.od): + yield i + + def __aiter__(self): + return self.aiter() + def __len__(self) -> int: # Skip the "highest subindex" entry, which is not part of the data return len(self.od) - int(0 in self.od) + async def alen(self) -> int: + return len(self.od) + def __contains__(self, subindex: object) -> bool: return subindex in self.od @@ -134,12 +157,20 @@ def __iter__(self) -> Iterator[int]: # Skip the "highest subindex" entry, which is not part of the data return iter(range(1, len(self) + 1)) + async def aiter(self): + for i in range(1, await self.alen() + 1): + yield i + + def __aiter__(self): + return self.aiter() + def __len__(self) -> int: return self[0].raw + async def alen(self) -> int: + return await self[0].aget_raw() # type: ignore[return-value] + def __contains__(self, subindex: object) -> bool: - if not isinstance(subindex, int): - return False return 0 <= subindex <= len(self) @@ -150,6 +181,10 @@ def __init__(self, sdo_node: SdoBase, od: objectdictionary.ODVariable): self.sdo_node = sdo_node variable.Variable.__init__(self, od) + def __await__(self): + return self.aget_raw().__await__() + + @ensure_not_async def get_data(self) -> bytes: data = self.sdo_node.upload(self.od.index, self.od.subindex) response_size = len(data) @@ -164,10 +199,18 @@ def get_data(self) -> bytes: data = data[:var_size] return data + async def aget_data(self) -> bytes: + return await self.sdo_node.aupload(self.od.index, self.od.subindex) + + @ensure_not_async def set_data(self, data: bytes): force_segment = self.od.data_type == objectdictionary.DOMAIN self.sdo_node.download(self.od.index, self.od.subindex, data, force_segment) + async def aset_data(self, data: bytes): + force_segment = self.od.data_type == objectdictionary.DOMAIN + await self.sdo_node.adownload(self.od.index, self.od.subindex, data, force_segment) + @property def writable(self) -> bool: return self.od.writable @@ -176,6 +219,7 @@ def writable(self) -> bool: def readable(self) -> bool: return self.od.readable + @ensure_not_async def open(self, mode="rb", encoding="ascii", buffering=1024, size=None, block_transfer=False, request_crc_support=True): """Open the data stream as a file like object. @@ -210,6 +254,13 @@ def open(self, mode="rb", encoding="ascii", buffering=1024, size=None, return self.sdo_node.open(self.od.index, self.od.subindex, mode, encoding, buffering, size, block_transfer, request_crc_support=request_crc_support) + async def aopen(self, mode="rb", encoding="ascii", buffering=1024, size=None, + block_transfer=False, request_crc_support=True): + """Open the data stream as a file like object. See open()""" + return await self.sdo_node.aopen(self.od.index, self.od.subindex, mode, + encoding, buffering, size, block_transfer, + request_crc_support=request_crc_support) + # For compatibility Record = SdoRecord diff --git a/canopen/sdo/client.py b/canopen/sdo/client.py index 82ee0c88..3853fefb 100644 --- a/canopen/sdo/client.py +++ b/canopen/sdo/client.py @@ -1,3 +1,4 @@ +import asyncio import io import logging import queue @@ -6,7 +7,7 @@ from can import CanError -from canopen import objectdictionary +from canopen.async_guard import ensure_not_async from canopen.sdo.base import SdoBase from canopen.sdo.constants import * from canopen.sdo.exceptions import * @@ -42,10 +43,12 @@ def __init__(self, rx_cobid, tx_cobid, od): """ SdoBase.__init__(self, rx_cobid, tx_cobid, od) self.responses = queue.Queue() + self.lock = asyncio.Lock() # For ensuring only one pending SDO request in async def on_response(self, can_id, data, timestamp): self.responses.put(bytes(data)) + @ensure_not_async def send_request(self, request): retries_left = self.MAX_RETRIES if self.PAUSE_BEFORE_SEND: @@ -109,6 +112,11 @@ def abort(self, abort_code=ABORT_GENERAL_ERROR): self.send_request(request) logger.error("Transfer aborted by client with code 0x%08X", abort_code) + async def aabort(self, abort_code=ABORT_GENERAL_ERROR): + """Abort current transfer. Async version.""" + return await asyncio.to_thread(self.abort, abort_code) + + @ensure_not_async def upload(self, index: int, subindex: int) -> bytes: """May be called to make a read operation without an Object Dictionary. @@ -136,6 +144,30 @@ def upload(self, index: int, subindex: int) -> bytes: data = data[:response_size] return data + async def aupload(self, index: int, subindex: int) -> bytes: + """May be called to make a read operation without an Object Dictionary. + Async version. + """ + async with self.lock: # Ensure only one active SDO request per channel + + # Deferring to thread because there are sleeps and queue waits in the call chain + # The call stack is typically: + # upload -> open -> ReadableStream -> request_reponse -> send_request -> network.send_message + # recv -> on_reponse -> queue.put + # request_reponse -> read_response -> queue.get + def _upload(): + with self.open(index, subindex, buffering=0) as fp: + response_size = fp.size + data = fp.read() + return data, response_size + + data, response_size = await asyncio.to_thread(_upload) + + if response_size and response_size < len(data): + data = data[:response_size] + return data + + @ensure_not_async def download( self, index: int, @@ -163,6 +195,27 @@ def download( force_segment=force_segment) as fp: fp.write(data) + async def adownload( + self, + index: int, + subindex: int, + data: bytes, + force_segment: bool = False, + ) -> None: + """May be called to make a write operation without an Object Dictionary. + Async version. + """ + async with self.lock: # Ensure only one active SDO request per channel + # Deferring to thread because there are sleeps in the call chain + + def _download(): + with self.open(index, subindex, "wb", buffering=7, size=len(data), + force_segment=force_segment) as fp: + fp.write(data) + + return await asyncio.to_thread(_download) + + @ensure_not_async def open(self, index, subindex=0, mode="rb", encoding="ascii", buffering=1024, size=None, block_transfer=False, force_segment=False, request_crc_support=True): """Open the data stream as a file like object. diff --git a/canopen/sdo/server.py b/canopen/sdo/server.py index c26dc998..04f9e70f 100644 --- a/canopen/sdo/server.py +++ b/canopen/sdo/server.py @@ -1,5 +1,6 @@ import logging +from canopen.async_guard import ensure_not_async from canopen.sdo.base import SdoBase from canopen.sdo.constants import * from canopen.sdo.exceptions import * @@ -190,6 +191,7 @@ def abort(self, abort_code=ABORT_GENERAL_ERROR): self.send_response(data) # logger.error("Transfer aborted with code 0x%08X", abort_code) + @ensure_not_async def upload(self, index: int, subindex: int) -> bytes: """May be called to make a read operation without an Object Dictionary. @@ -205,6 +207,22 @@ def upload(self, index: int, subindex: int) -> bytes: """ return self._node.get_data(index, subindex) + async def aupload(self, index: int, subindex: int) -> bytes: + """May be called to make a read operation without an Object Dictionary. + + :param index: + Index of object to read. + :param subindex: + Sub-index of object to read. + + :return: A data object. + + :raises canopen.SdoAbortedError: + When node responds with an error. + """ + return self._node.get_data(index, subindex) + + @ensure_not_async def download( self, index: int, @@ -225,3 +243,24 @@ def download( When node responds with an error. """ return self._node.set_data(index, subindex, data) + + async def adownload( + self, + index: int, + subindex: int, + data: bytes, + force_segment: bool = False, + ): + """May be called to make a write operation without an Object Dictionary. + + :param index: + Index of object to write. + :param subindex: + Sub-index of object to write. + :param data: + Data to be written. + + :raises canopen.SdoAbortedError: + When node responds with an error. + """ + return self._node.set_data(index, subindex, data) diff --git a/canopen/variable.py b/canopen/variable.py index 8441ac32..77f54244 100644 --- a/canopen/variable.py +++ b/canopen/variable.py @@ -5,6 +5,7 @@ from typing import Union from canopen import objectdictionary +from canopen.async_guard import ensure_not_async from canopen.utils import pretty_index @@ -35,9 +36,15 @@ def __repr__(self) -> str: def get_data(self) -> bytes: raise NotImplementedError("Variable is not readable") + async def aget_data(self) -> bytes: + raise NotImplementedError("Variable is not readable") + def set_data(self, data: bytes): raise NotImplementedError("Variable is not writable") + async def aset_data(self, data: bytes): + raise NotImplementedError("Variable is not writable") + @property def data(self) -> bytes: """Byte representation of the object as :class:`bytes`.""" @@ -77,7 +84,14 @@ def raw(self) -> Union[int, bool, float, str, bytes]: Data types that this library does not handle yet must be read and written as :class:`bytes`. """ - value = self.od.decode_raw(self.data) + return self._get_raw(self.get_data()) + + async def aget_raw(self) -> Union[int, bool, float, str, bytes]: + """Raw representation of the object, async variant""" + return self._get_raw(await self.aget_data()) + + def _get_raw(self, data: bytes) -> Union[int, bool, float, str, bytes]: + value = self.od.decode_raw(data) text = f"Value of {self.name!r} ({pretty_index(self.index, self.subindex)}) is {value!r}" if ( isinstance(value, int) @@ -89,10 +103,17 @@ def raw(self) -> Union[int, bool, float, str, bytes]: @raw.setter def raw(self, value: Union[int, bool, float, str, bytes]): + self.set_data(self._set_raw(value)) + + async def aset_raw(self, value: Union[int, bool, float, str, bytes]): + """Set the raw value of the object, async variant""" + await self.aset_data(self._set_raw(value)) + + def _set_raw(self, value: Union[int, bool, float, str, bytes]): logger.debug("Writing %r (0x%04X:%02X) = %r", self.name, self.index, self.subindex, value) - self.data = self.od.encode_raw(value) + return self.od.encode_raw(value) @property def phys(self) -> Union[int, bool, float, str, bytes]: @@ -102,7 +123,14 @@ def phys(self) -> Union[int, bool, float, str, bytes]: either a :class:`float` or an :class:`int`. Non integers will be passed as is. """ - value = self.od.decode_phys(self.raw) + return self._get_phys(self.raw) + + async def aget_phys(self) -> Union[int, bool, float, str, bytes]: + """Physical value scaled with some factor (defaults to 1), async variant.""" + return self._get_phys(await self.aget_raw()) + + def _get_phys(self, raw: Union[int, bool, float, str, bytes]): + value = self.od.decode_phys(raw) if self.od.unit: logger.debug("Physical value is %s %s", value, self.od.unit) return value @@ -111,6 +139,10 @@ def phys(self) -> Union[int, bool, float, str, bytes]: def phys(self, value: Union[int, bool, float, str, bytes]): self.raw = self.od.encode_phys(value) + async def aset_phys(self, value: Union[int, bool, float, str, bytes]): + """Set physical value scaled with some factor (defaults to 1). Async variant""" + await self.aset_raw(self.od.encode_phys(value)) + @property def desc(self) -> str: """Convert to and from a description of the value as a string. @@ -124,10 +156,20 @@ def desc(self) -> str: logger.debug("Description is '%s'", value) return value + async def aget_desc(self) -> str: + """Converts to and from a description of the value as a string, async variant.""" + value = self.od.decode_desc(await self.aget_raw()) + logger.debug("Description is '%s'", value) + return value + @desc.setter def desc(self, desc: str): self.raw = self.od.encode_desc(desc) + async def aset_desc(self, desc: str): + """Set variable description, async variant.""" + await self.aset_raw(self.od.encode_desc(desc)) + @property def bits(self) -> Bits: """Access bits using integers, slices, or bit descriptions.""" @@ -156,6 +198,16 @@ def read(self, fmt: str = "raw") -> Union[int, bool, float, str, bytes]: return self.desc raise ValueError(f"Invalid format '{fmt}'") + async def aread(self, fmt: str = "raw") -> Union[int, bool, float, str, bytes]: + """Alternative way of reading using a function instead of attributes. Async variant.""" + if fmt == "raw": + return await self.aget_raw() + elif fmt == "phys": + return await self.aget_phys() + elif fmt == "desc": + return await self.aget_desc() + raise ValueError(f"Unknown format '{fmt}'") + def write( self, value: Union[int, bool, float, str, bytes], @@ -181,12 +233,25 @@ def write( raise TypeError("fmt=desc requires a string value") self.desc = value + async def awrite( + self, value: Union[int, bool, float, str, bytes], fmt: str = "raw" + ) -> None: + """Alternative way of writing using a function instead of attributes. Async variant""" + if fmt == "raw": + await self.aset_raw(value) + elif fmt == "phys": + await self.aset_phys(value) + elif fmt == "desc": + await self.aset_desc(value) # type: ignore[arg-type] + class Bits(Mapping): + @ensure_not_async def __init__(self, variable: Variable): assert variable.od.data_type in objectdictionary.datatypes.INTEGER_TYPES self.variable = variable + # FIXME: This is not compatible with async self.read() self.raw: int @@ -221,3 +286,9 @@ def read(self): def write(self): self.variable.raw = self.raw + + async def aread(self): + self.raw = await self.variable.aget_raw() + + async def awrite(self): + await self.variable.aset_raw(self.raw) diff --git a/examples/canopen_async.py b/examples/canopen_async.py new file mode 100644 index 00000000..147439d7 --- /dev/null +++ b/examples/canopen_async.py @@ -0,0 +1,69 @@ +import asyncio +import logging +import canopen + +# Set logging output +logging.basicConfig(level=logging.INFO) +log = logging.getLogger(__name__) + + +async def do_loop(network: canopen.Network, nodeid): + + # Create the node object and load the OD + node: canopen.RemoteNode = await network.aadd_node(nodeid, 'eds/e35.eds') + + # Get the PDOs from the remote + await node.tpdo.aread(from_od=False) + await node.rpdo.aread(from_od=False) + + # Set the remote state + node.nmt.state = 'OPERATIONAL' + + # Set SDO + await node.sdo['something'].aset_raw(2) + + i = 0 + while True: + i += 1 + + # Wait for PDO + t = await node.tpdo[1].await_for_reception(1) + if not t: + continue + + # Get TPDO value + # PDO values are accessed directly, no await required + state = node.tpdo[1]['state'].raw + + # If state send RPDO to remote + if state == 5: + + await asyncio.sleep(0.2) + + # Set RPDO and transmit + node.rpdo[1]['count'].phys = i + node.rpdo[1].transmit() + + +async def amain(): + + # Create the canopen network and connect it to the CAN bus + loop = asyncio.get_running_loop() + async with canopen.Network(loop=loop).connect( + interface='virtual', bitrate=1000000, receive_own_messages=True + ) as network: + + # Start two instances and run them concurrently + # NOTE: It is better to use asyncio.TaskGroup to manage tasks, but this + # is not available before Python 3.11. + await asyncio.gather( + asyncio.create_task(do_loop(network, 20)), + asyncio.create_task(do_loop(network, 21)), + ) + + +def main(): + asyncio.run(amain()) + +if __name__ == '__main__': + main() diff --git a/pyproject.toml b/pyproject.toml index 986f0fdf..ba9d1488 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,7 @@ authors = [ {name = "Christian Sandberg", email = "christiansandberg@me.com"}, {name = "André Colomb", email = "src@andre.colomb.de"}, {name = "André Filipe Silva", email = "afsilva.work@gmail.com"}, + {name = "Svein Seldal", email = "sveinse@seldal.com"}, ] description = "CANopen stack implementation" readme = "README.rst" @@ -67,3 +68,10 @@ lines-after-imports = 2 [tool.black] line-length = 96 skip-string-normalization = true + +[tool.coverage.run] +branch = true +[tool.coverage.report] +exclude_also = [ + 'if TYPE_CHECKING:', +] diff --git a/test/test_emcy.py b/test/test_emcy.py index b4b54a19..4a79b4fe 100644 --- a/test/test_emcy.py +++ b/test/test_emcy.py @@ -1,3 +1,4 @@ +import asyncio import logging import threading import unittest @@ -22,7 +23,24 @@ def mock_rx_thread(consumer: canopen.emcy.EmcyConsumer, func): t.join(TIMEOUT) -class TestEmcy(unittest.TestCase): +class TestEmcy(unittest.IsolatedAsyncioTestCase): + + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + + def setUp(self): + loop = None + if self.async_test: + loop = asyncio.get_event_loop() + self.loop = loop + + self.net = canopen.Network(loop=loop) + self.net.connect(interface="virtual") + self.emcy = canopen.emcy.EmcyConsumer() + self.emcy.network = self.net + + def tearDown(self): + self.net.disconnect() def check_error(self, err, code, reg, data, ts): self.assertIsInstance(err, canopen.emcy.EmcyError) @@ -32,15 +50,24 @@ def check_error(self, err, code, reg, data, ts): self.assertEqual(err.data, data) self.assertAlmostEqual(err.timestamp, ts) - def test_emcy_consumer_on_emcy(self): + async def dispatch_emcy(self, can_id, data, ts): + # Dispatch an EMCY datagram. + if self.async_test: + await asyncio.to_thread( + self.emcy.on_emcy, can_id, data, ts + ) + else: + self.emcy.on_emcy(can_id, data, ts) + + async def test_emcy_consumer_on_emcy(self): """Make sure multiple callbacks receive the same information.""" - emcy = canopen.emcy.EmcyConsumer() + emcy = self.emcy = canopen.emcy.EmcyConsumer() acc1 = [] acc2 = [] emcy.add_callback(lambda err: acc1.append(err)) emcy.add_callback(lambda err: acc2.append(err)) - emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000) + await self.dispatch_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000) self.assertEqual(len(emcy.log), 1) self.assertEqual(len(emcy.active), 1) @@ -53,7 +80,7 @@ def test_emcy_consumer_on_emcy(self): data=bytes([0, 1, 2, 3, 4]), ts=1000, ) - emcy.on_emcy(0x81, b'\x10\x90\x01\x04\x03\x02\x01\x00', 2000) + await self.dispatch_emcy(0x81, b'\x10\x90\x01\x04\x03\x02\x01\x00', 2000) self.assertEqual(len(emcy.log), 2) self.assertEqual(len(emcy.active), 2) @@ -65,14 +92,14 @@ def test_emcy_consumer_on_emcy(self): data=bytes([4, 3, 2, 1, 0]), ts=2000, ) - emcy.on_emcy(0x81, b'\x00\x00\x00\x00\x00\x00\x00\x00', 2000) + await self.dispatch_emcy(0x81, b'\x00\x00\x00\x00\x00\x00\x00\x00', 2000) self.assertEqual(len(emcy.log), 3) self.assertEqual(len(emcy.active), 0) - def test_emcy_consumer_reset(self): - emcy = canopen.emcy.EmcyConsumer() - emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000) - emcy.on_emcy(0x81, b'\x10\x90\x01\x04\x03\x02\x01\x00', 2000) + async def test_emcy_consumer_reset(self): + emcy = self.emcy = canopen.emcy.EmcyConsumer() + await self.dispatch_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000) + await self.dispatch_emcy(0x81, b'\x10\x90\x01\x04\x03\x02\x01\x00', 2000) self.assertEqual(len(emcy.log), 2) self.assertEqual(len(emcy.active), 2) @@ -80,7 +107,10 @@ def test_emcy_consumer_reset(self): self.assertEqual(len(emcy.log), 0) self.assertEqual(len(emcy.active), 0) - def test_emcy_consumer_wait(self): + async def test_emcy_consumer_wait(self): + if self.async_test: + self.skipTest("Not implemented for async") + emcy = canopen.emcy.EmcyConsumer() def push_err(): @@ -94,7 +124,10 @@ def check_err(err): ) # Check unfiltered wait, on timeout. - self.assertIsNone(emcy.wait(timeout=TIMEOUT)) + if self.async_test: + self.assertIsNone(await emcy.async_wait(timeout=TIMEOUT)) + else: + self.assertIsNone(emcy.wait(timeout=TIMEOUT)) # Check unfiltered wait, on success. with ( @@ -181,6 +214,18 @@ def push_multiple_errors(): self.assertEqual(err.code, 0x2003) +class TestEmcySync(TestEmcy): + """ Run the tests in non-asynchronous mode. """ + __test__ = True + async_test = False + + +class TestEmcyAsync(TestEmcy): + """ Run the tests in asynchronous mode. """ + __test__ = True + async_test = True + + class TestEmcyError(unittest.TestCase): def test_emcy_error(self): @@ -239,12 +284,19 @@ def check(code, expected): check(0xffff, "Device Specific") -class TestEmcyProducer(unittest.TestCase): +class TestEmcyProducer(unittest.IsolatedAsyncioTestCase): + + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool def setUp(self): - self.txbus = can.Bus(interface="virtual") - self.rxbus = can.Bus(interface="virtual") - self.net = canopen.Network(self.txbus) + loop = None + if self.async_test: + loop = asyncio.get_event_loop() + + self.txbus = can.Bus(interface="virtual", loop=loop) + self.rxbus = can.Bus(interface="virtual", loop=loop) + self.net = canopen.Network(self.txbus, loop=loop) self.net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 self.net.connect() self.emcy = canopen.emcy.EmcyProducer(0x80 + 1) @@ -261,7 +313,7 @@ def check_response(self, expected): actual = msg.data self.assertEqual(actual, expected) - def test_emcy_producer_send(self): + async def test_emcy_producer_send(self): def check(*args, res): self.emcy.send(*args) self.check_response(res) @@ -270,7 +322,7 @@ def check(*args, res): check(0x2001, 0x2, res=b'\x01\x20\x02\x00\x00\x00\x00\x00') check(0x2001, 0x2, b'\x2a', res=b'\x01\x20\x02\x2a\x00\x00\x00\x00') - def test_emcy_producer_reset(self): + async def test_emcy_producer_reset(self): def check(*args, res): self.emcy.reset(*args) self.check_response(res) @@ -359,5 +411,17 @@ def test_producer_reset_consumer_integration(self): self.assertEqual(len(self.consumer.log), 2) +class TestEmcyProducerSync(TestEmcyProducer): + """ Run the tests in non-asynchronous mode. """ + __test__ = True + async_test = False + + +class TestEmcyProducerAsync(TestEmcyProducer): + """ Run the tests in asynchronous mode. """ + __test__ = True + async_test = True + + if __name__ == "__main__": unittest.main() diff --git a/test/test_local.py b/test/test_local.py index 6ab94645..f8290385 100644 --- a/test/test_local.py +++ b/test/test_local.py @@ -1,43 +1,56 @@ import time import unittest +import asyncio import canopen +from canopen.async_guard import AllowBlocking from .util import SAMPLE_EDS -class TestSDO(unittest.TestCase): +class TestSDO(unittest.IsolatedAsyncioTestCase): """ Test SDO client and server against each other. """ - @classmethod - def setUpClass(cls): - cls.network1 = canopen.Network() - cls.network1.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 - cls.network1.connect("test", interface="virtual") - cls.remote_node = cls.network1.add_node(2, SAMPLE_EDS) - - cls.network2 = canopen.Network() - cls.network2.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 - cls.network2.connect("test", interface="virtual") - cls.local_node = cls.network2.create_node(2, SAMPLE_EDS) - - cls.remote_node2 = cls.network1.add_node(3, SAMPLE_EDS) - - cls.local_node2 = cls.network2.create_node(3, SAMPLE_EDS) - - @classmethod - def tearDownClass(cls): - cls.network1.disconnect() - cls.network2.disconnect() - - def test_expedited_upload(self): - self.local_node.sdo[0x1400][1].raw = 0x99 - vendor_id = self.remote_node.sdo[0x1400][1].raw + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + + def setUp(self): + loop = None + if self.async_test: + loop = asyncio.get_event_loop() + + self.network1 = canopen.Network(loop=loop) + self.network1.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network1.connect("test", interface="virtual") + with AllowBlocking(): + self.remote_node = self.network1.add_node(2, SAMPLE_EDS) + + self.network2 = canopen.Network(loop=loop) + self.network2.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network2.connect("test", interface="virtual") + self.local_node = self.network2.create_node(2, SAMPLE_EDS) + with AllowBlocking(): + self.remote_node2 = self.network1.add_node(3, SAMPLE_EDS) + self.local_node2 = self.network2.create_node(3, SAMPLE_EDS) + + def tearDown(self): + self.network1.disconnect() + self.network2.disconnect() + + async def test_expedited_upload(self): + if self.async_test: + await self.local_node.sdo[0x1400][1].aset_raw(0x99) + vendor_id = await self.remote_node.sdo[0x1400][1].aget_raw() + else: + self.local_node.sdo[0x1400][1].raw = 0x99 + vendor_id = self.remote_node.sdo[0x1400][1].raw self.assertEqual(vendor_id, 0x99) - def test_block_upload_switch_to_expedite_upload(self): + async def test_block_upload_switch_to_expedite_upload(self): + if self.async_test: + self.skipTest("Block upload not supported in async mode") with self.assertRaises(canopen.SdoCommunicationError) as context: with self.remote_node.sdo[0x1008].open('r', block_transfer=True) as fp: pass @@ -45,7 +58,9 @@ def test_block_upload_switch_to_expedite_upload(self): # from block upload to expedite upload self.assertEqual("Unexpected response 0x41", str(context.exception)) - def test_block_download_not_supported(self): + async def test_block_download_not_supported(self): + if self.async_test: + self.skipTest("Block download not supported in async mode") data = b"TEST DEVICE" with self.assertRaises(canopen.SdoAbortedError) as context: with self.remote_node.sdo[0x1008].open('wb', @@ -54,104 +69,174 @@ def test_block_download_not_supported(self): pass self.assertEqual(context.exception.code, 0x05040001) - def test_expedited_upload_default_value_visible_string(self): - device_name = self.remote_node.sdo["Manufacturer device name"].raw + async def test_expedited_upload_default_value_visible_string(self): + if self.async_test: + device_name = await self.remote_node.sdo["Manufacturer device name"].aget_raw() + else: + device_name = self.remote_node.sdo["Manufacturer device name"].raw self.assertEqual(device_name, "TEST DEVICE") - def test_expedited_upload_default_value_real(self): - sampling_rate = self.remote_node.sdo["Sensor Sampling Rate (Hz)"].raw + async def test_expedited_upload_default_value_real(self): + if self.async_test: + sampling_rate = await self.remote_node.sdo["Sensor Sampling Rate (Hz)"].aget_raw() + else: + sampling_rate = self.remote_node.sdo["Sensor Sampling Rate (Hz)"].raw self.assertAlmostEqual(sampling_rate, 5.2, places=2) - def test_upload_zero_length(self): - self.local_node.sdo["Manufacturer device name"].raw = b"" - with self.assertRaises(canopen.SdoAbortedError) as error: - self.remote_node.sdo["Manufacturer device name"].data + async def test_upload_zero_length(self): + if self.async_test: + await self.local_node.sdo["Manufacturer device name"].aset_raw(b"") + with self.assertRaises(canopen.SdoAbortedError) as error: + await self.remote_node.sdo["Manufacturer device name"].aget_data() + else: + self.local_node.sdo["Manufacturer device name"].raw = b"" + with self.assertRaises(canopen.SdoAbortedError) as error: + self.remote_node.sdo["Manufacturer device name"].data # Should be No data available self.assertEqual(error.exception.code, 0x0800_0024) - def test_segmented_upload(self): - self.local_node.sdo["Manufacturer device name"].raw = "Some cool device" - device_name = self.remote_node.sdo["Manufacturer device name"].data + async def test_segmented_upload(self): + if self.async_test: + await self.local_node.sdo["Manufacturer device name"].aset_raw("Some cool device") + device_name = await self.remote_node.sdo["Manufacturer device name"].aget_data() + else: + self.local_node.sdo["Manufacturer device name"].raw = "Some cool device" + device_name = self.remote_node.sdo["Manufacturer device name"].data self.assertEqual(device_name, b"Some cool device") - def test_expedited_download(self): - self.remote_node.sdo[0x2004].raw = 0xfeff - value = self.local_node.sdo[0x2004].raw + async def test_expedited_download(self): + if self.async_test: + await self.remote_node.sdo[0x2004].aset_raw(0xfeff) + value = await self.local_node.sdo[0x2004].aget_raw() + else: + self.remote_node.sdo[0x2004].raw = 0xfeff + value = self.local_node.sdo[0x2004].raw self.assertEqual(value, 0xfeff) - def test_expedited_download_wrong_datatype(self): + async def test_expedited_download_wrong_datatype(self): # Try to write 32 bit in integer16 type - with self.assertRaises(canopen.SdoAbortedError) as error: - self.remote_node.sdo.download(0x2001, 0x0, bytes([10, 10, 10, 10])) + if self.async_test: + with self.assertRaises(canopen.SdoAbortedError) as error: + await self.remote_node.sdo.adownload(0x2001, 0x0, bytes([10, 10, 10, 10])) + else: + with self.assertRaises(canopen.SdoAbortedError) as error: + self.remote_node.sdo.download(0x2001, 0x0, bytes([10, 10, 10, 10])) self.assertEqual(error.exception.code, 0x06070010) # Try to write normal 16 bit word, should be ok - self.remote_node.sdo.download(0x2001, 0x0, bytes([10, 10])) - value = self.remote_node.sdo.upload(0x2001, 0x0) + if self.async_test: + await self.remote_node.sdo.adownload(0x2001, 0x0, bytes([10, 10])) + value = await self.remote_node.sdo.aupload(0x2001, 0x0) + else: + self.remote_node.sdo.download(0x2001, 0x0, bytes([10, 10])) + value = self.remote_node.sdo.upload(0x2001, 0x0) self.assertEqual(value, bytes([10, 10])) - def test_segmented_download(self): - self.remote_node.sdo[0x2000].raw = "Another cool device" - value = self.local_node.sdo[0x2000].data + async def test_segmented_download(self): + if self.async_test: + await self.remote_node.sdo[0x2000].aset_raw("Another cool device") + value = await self.local_node.sdo[0x2000].aget_data() + else: + self.remote_node.sdo[0x2000].raw = "Another cool device" + value = self.local_node.sdo[0x2000].data self.assertEqual(value, b"Another cool device") - def test_slave_send_heartbeat(self): + async def test_slave_send_heartbeat(self): # Setting the heartbeat time should trigger heartbeating # to start - self.remote_node.sdo["Producer heartbeat time"].raw = 100 - state = self.remote_node.nmt.wait_for_heartbeat() + if self.async_test: + await self.remote_node.sdo["Producer heartbeat time"].aset_raw(100) + state = await self.remote_node.nmt.await_for_heartbeat() + else: + self.remote_node.sdo["Producer heartbeat time"].raw = 100 + state = self.remote_node.nmt.wait_for_heartbeat() self.local_node.nmt.stop_heartbeat() # The NMT master will change the state INITIALISING (0) # to PRE-OPERATIONAL (127) self.assertEqual(state, 'PRE-OPERATIONAL') - def test_nmt_state_initializing_to_preoper(self): + async def test_nmt_state_initializing_to_preoper(self): # Initialize the heartbeat timer - self.local_node.sdo["Producer heartbeat time"].raw = 100 + if self.async_test: + await self.local_node.sdo["Producer heartbeat time"].aset_raw(100) + else: + self.local_node.sdo["Producer heartbeat time"].raw = 100 self.local_node.nmt.stop_heartbeat() # This transition shall start the heartbeating self.local_node.nmt.state = 'INITIALISING' self.local_node.nmt.state = 'PRE-OPERATIONAL' - state = self.remote_node.nmt.wait_for_heartbeat() + if self.async_test: + state = await self.remote_node.nmt.await_for_heartbeat() + else: + state = self.remote_node.nmt.wait_for_heartbeat() self.local_node.nmt.stop_heartbeat() self.assertEqual(state, 'PRE-OPERATIONAL') - def test_receive_abort_request(self): - self.remote_node.sdo.abort(0x0504_0003) # Invalid sequence number + async def test_receive_abort_request(self): + if self.async_test: + await self.remote_node.sdo.aabort(0x0504_0003) + else: + self.remote_node.sdo.abort(0x0504_0003) # Line below is just so that we are sure the client have received the abort # before we do the check - time.sleep(0.1) + if self.async_test: + await asyncio.sleep(0.1) + else: + time.sleep(0.1) self.assertEqual(self.local_node.sdo.last_received_error, 0x0504_0003) - def test_start_remote_node(self): + async def test_start_remote_node(self): self.remote_node.nmt.state = 'OPERATIONAL' # Line below is just so that we are sure the client have received the command # before we do the check - time.sleep(0.1) + if self.async_test: + await asyncio.sleep(0.1) + else: + time.sleep(0.1) slave_state = self.local_node.nmt.state self.assertEqual(slave_state, 'OPERATIONAL') - def test_two_nodes_on_the_bus(self): - self.local_node.sdo["Manufacturer device name"].raw = "Some cool device" - device_name = self.remote_node.sdo["Manufacturer device name"].data + async def test_two_nodes_on_the_bus(self): + if self.async_test: + await self.local_node.sdo["Manufacturer device name"].aset_raw("Some cool device") + device_name = await self.remote_node.sdo["Manufacturer device name"].aget_data() + else: + self.local_node.sdo["Manufacturer device name"].raw = "Some cool device" + device_name = self.remote_node.sdo["Manufacturer device name"].data self.assertEqual(device_name, b"Some cool device") - self.local_node2.sdo["Manufacturer device name"].raw = "Some cool device2" - device_name = self.remote_node2.sdo["Manufacturer device name"].data + if self.async_test: + await self.local_node2.sdo["Manufacturer device name"].aset_raw("Some cool device2") + device_name = await self.remote_node2.sdo["Manufacturer device name"].aget_data() + else: + self.local_node2.sdo["Manufacturer device name"].raw = "Some cool device2" + device_name = self.remote_node2.sdo["Manufacturer device name"].data self.assertEqual(device_name, b"Some cool device2") - def test_abort(self): - with self.assertRaises(canopen.SdoAbortedError) as cm: - _ = self.remote_node.sdo.upload(0x1234, 0) + async def test_abort(self): + if self.async_test: + with self.assertRaises(canopen.SdoAbortedError) as cm: + _ = await self.remote_node.sdo.aupload(0x1234, 0) + else: + with self.assertRaises(canopen.SdoAbortedError) as cm: + _ = self.remote_node.sdo.upload(0x1234, 0) # Should be Object does not exist self.assertEqual(cm.exception.code, 0x06020000) - with self.assertRaises(canopen.SdoAbortedError) as cm: - _ = self.remote_node.sdo.upload(0x1018, 100) + if self.async_test: + with self.assertRaises(canopen.SdoAbortedError) as cm: + _ = await self.remote_node.sdo.aupload(0x1018, 100) + else: + with self.assertRaises(canopen.SdoAbortedError) as cm: + _ = self.remote_node.sdo.upload(0x1018, 100) # Should be Subindex does not exist self.assertEqual(cm.exception.code, 0x06090011) - with self.assertRaises(canopen.SdoAbortedError) as cm: - _ = self.remote_node.sdo[0x1001].data + if self.async_test: + with self.assertRaises(canopen.SdoAbortedError) as cm: + _ = await self.remote_node.sdo[0x1001].aget_data() + else: + with self.assertRaises(canopen.SdoAbortedError) as cm: + _ = self.remote_node.sdo[0x1001].data # Should be Resource not available self.assertEqual(cm.exception.code, 0x060A0023) @@ -163,54 +248,98 @@ def _some_read_callback(self, **kwargs): def _some_write_callback(self, **kwargs): self._kwargs = kwargs - def test_callbacks(self): + async def test_callbacks(self): self.local_node.add_read_callback(self._some_read_callback) self.local_node.add_write_callback(self._some_write_callback) - data = self.remote_node.sdo.upload(0x1003, 5) + if self.async_test: + data = await self.remote_node.sdo.aupload(0x1003, 5) + else: + data = self.remote_node.sdo.upload(0x1003, 5) self.assertEqual(data, b"\x01\x02\x00\x00") self.assertEqual(self._kwargs["index"], 0x1003) self.assertEqual(self._kwargs["subindex"], 5) - self.remote_node.sdo.download(0x1017, 0, b"\x03\x04") + if self.async_test: + await self.remote_node.sdo.adownload(0x1017, 0, b"\x03\x04") + else: + self.remote_node.sdo.download(0x1017, 0, b"\x03\x04") self.assertEqual(self._kwargs["index"], 0x1017) self.assertEqual(self._kwargs["subindex"], 0) self.assertEqual(self._kwargs["data"], b"\x03\x04") -class TestPDO(unittest.TestCase): +class TestSDOSync(TestSDO): + """ Run the test in non-async mode. """ + __test__ = True + async_test = False + + +class TestSDOAsync(TestSDO): + """ Run the test in async mode. """ + __test__ = True + async_test = True + + +class TestPDO(unittest.IsolatedAsyncioTestCase): """ Test PDO slave. """ - @classmethod - def setUpClass(cls): - cls.network1 = canopen.Network() - cls.network1.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 - cls.network1.connect("test", interface="virtual") - cls.remote_node = cls.network1.add_node(2, SAMPLE_EDS) + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + + def setUp(self): + loop = None + if self.async_test: + loop = asyncio.get_event_loop() - cls.network2 = canopen.Network() - cls.network2.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 - cls.network2.connect("test", interface="virtual") - cls.local_node = cls.network2.create_node(2, SAMPLE_EDS) + self.network1 = canopen.Network(loop=loop) + self.network1.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network1.connect("test", interface="virtual") + with AllowBlocking(): + self.remote_node = self.network1.add_node(2, SAMPLE_EDS) - @classmethod - def tearDownClass(cls): - cls.network1.disconnect() - cls.network2.disconnect() + self.network2 = canopen.Network(loop=loop) + self.network2.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network2.connect("test", interface="virtual") + self.local_node = self.network2.create_node(2, SAMPLE_EDS) - def test_read(self): + def tearDown(self): + self.network1.disconnect() + self.network2.disconnect() + + async def test_read(self): # TODO: Do some more checks here. Currently it only tests that they # can be called without raising an error. - self.remote_node.pdo.read() - self.local_node.pdo.read() - - def test_save(self): + if self.async_test: + await self.remote_node.pdo.aread() + await self.local_node.pdo.aread() + else: + self.remote_node.pdo.read() + self.local_node.pdo.read() + + async def test_save(self): # TODO: Do some more checks here. Currently it only tests that they # can be called without raising an error. - self.remote_node.pdo.save() - self.local_node.pdo.save() + if self.async_test: + await self.remote_node.pdo.asave() + await self.local_node.pdo.asave() + else: + self.remote_node.pdo.save() + self.local_node.pdo.save() + + +class TestPDOSync(TestPDO): + """ Run the test in non-async mode. """ + __test__ = True + async_test = False + + +class TestPDOAsync(TestPDO): + """ Run the test in async mode. """ + __test__ = True + async_test = True if __name__ == "__main__": diff --git a/test/test_network.py b/test/test_network.py index 8017f89c..553784a5 100644 --- a/test/test_network.py +++ b/test/test_network.py @@ -1,6 +1,7 @@ import logging import time import unittest +import asyncio import can @@ -9,22 +10,39 @@ from .util import SAMPLE_EDS -class TestNetwork(unittest.TestCase): +class TestNetwork(unittest.IsolatedAsyncioTestCase): + + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool def setUp(self): - self.network = canopen.Network() + self.loop = None + if self.async_test: + self.loop = asyncio.get_event_loop() + + self.network = canopen.Network(loop=self.loop) self.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 - def test_network_add_node(self): + def tearDown(self): + if self.network.bus is not None: + self.network.disconnect() + + async def test_network_add_node(self): # Add using str. with self.assertLogs(): - node = self.network.add_node(2, SAMPLE_EDS) + if self.async_test: + node = await self.network.aadd_node(2, SAMPLE_EDS) + else: + node = self.network.add_node(2, SAMPLE_EDS) self.assertEqual(self.network[2], node) self.assertEqual(node.id, 2) self.assertIsInstance(node, canopen.RemoteNode) # Add using OD. - node = self.network.add_node(3, self.network[2].object_dictionary) + if self.async_test: + node = await self.network.aadd_node(3, self.network[2].object_dictionary) + else: + node = self.network.add_node(3, self.network[2].object_dictionary) self.assertEqual(self.network[3], node) self.assertEqual(node.id, 3) self.assertIsInstance(node, canopen.RemoteNode) @@ -32,7 +50,10 @@ def test_network_add_node(self): # Add using RemoteNode. with self.assertLogs(): node = canopen.RemoteNode(4, SAMPLE_EDS) - self.network.add_node(node) + if self.async_test: + await self.network.aadd_node(node) + else: + self.network.add_node(node) self.assertEqual(self.network[4], node) self.assertEqual(node.id, 4) self.assertIsInstance(node, canopen.RemoteNode) @@ -40,7 +61,10 @@ def test_network_add_node(self): # Add using LocalNode. with self.assertLogs(): node = canopen.LocalNode(5, SAMPLE_EDS) - self.network.add_node(node) + if self.async_test: + await self.network.aadd_node(node) + else: + self.network.add_node(node) self.assertEqual(self.network[5], node) self.assertEqual(node.id, 5) self.assertIsInstance(node, canopen.LocalNode) @@ -48,12 +72,15 @@ def test_network_add_node(self): # Verify that we've got the correct number of nodes. self.assertEqual(len(self.network), 4) - def test_network_add_node_upload_eds(self): + async def test_network_add_node_upload_eds(self): # Will err because we're not connected to a real network. with self.assertLogs(level=logging.ERROR): - self.network.add_node(2, SAMPLE_EDS, upload_eds=True) + if self.async_test: + await self.network.aadd_node(2, SAMPLE_EDS, upload_eds=True) + else: + self.network.add_node(2, SAMPLE_EDS, upload_eds=True) - def test_network_create_node(self): + async def test_network_create_node(self): with self.assertLogs(): self.network.create_node(2, SAMPLE_EDS) self.network.create_node(3, SAMPLE_EDS) @@ -63,7 +90,7 @@ def test_network_create_node(self): self.assertIsInstance(self.network[3], canopen.LocalNode) self.assertIsInstance(self.network[4], canopen.RemoteNode) - def test_network_check(self): + async def test_network_check(self): self.network.connect(interface="virtual") def cleanup(): @@ -87,18 +114,29 @@ class Custom(Exception): with self.assertLogs(level=logging.ERROR): self.network.disconnect() - def test_network_notify(self): + async def test_network_notify(self): with self.assertLogs(): - self.network.add_node(2, SAMPLE_EDS) + if self.async_test: + await self.network.aadd_node(2, SAMPLE_EDS) + else: + self.network.add_node(2, SAMPLE_EDS) node = self.network[2] - self.network.notify(0x82, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1473418396.0) + async def notify(*args): + """Simulate a notification from the network.""" + if self.async_test: + # If we're using async, we must run the notify in a thread + # to avoid getting blocking call errors. + await asyncio.to_thread(self.network.notify, *args) + else: + self.network.notify(*args) + await notify(0x82, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1473418396.0) self.assertEqual(len(node.emcy.active), 1) - self.network.notify(0x702, b'\x05', 1473418396.0) + await notify(0x702, b'\x05', 1473418396.0) self.assertEqual(node.nmt.state, 'OPERATIONAL') self.assertListEqual(self.network.scanner.nodes, [2]) - def test_network_send_message(self): - bus = can.interface.Bus(interface="virtual") + async def test_network_send_message(self): + bus = can.interface.Bus(interface="virtual", loop=self.loop) self.addCleanup(bus.shutdown) self.network.connect(interface="virtual") @@ -119,7 +157,7 @@ def test_network_send_message(self): self.assertEqual(msg.arbitration_id, 0x12345) self.assertTrue(msg.is_extended_id) - def test_network_subscribe_unsubscribe(self): + async def test_network_subscribe_unsubscribe(self): N_HOOKS = 3 accumulators = [] * N_HOOKS @@ -149,7 +187,7 @@ def hook(*args, i=i): # Verify that no new data was added to the accumulator. self.assertEqual(accumulators[0], [(0, bytes([1, 2, 3]), 1000)]) - def test_network_subscribe_multiple(self): + async def test_network_subscribe_multiple(self): N_HOOKS = 3 self.network.connect(interface="virtual", receive_own_messages=True) self.addCleanup(self.network.disconnect) @@ -202,16 +240,20 @@ def hook(*args, i=i): self.assertEqual(accumulators[1], BATCH1) self.assertEqual(accumulators[2], BATCH1 + [BATCH2] + [BATCH3]) - def test_network_context_manager(self): + async def test_network_context_manager(self): with self.network.connect(interface="virtual"): pass with self.assertRaisesRegex(RuntimeError, "Not connected"): self.network.send_message(0, []) - def test_network_item_access(self): + async def test_network_item_access(self): with self.assertLogs(): - self.network.add_node(2, SAMPLE_EDS) - self.network.add_node(3, SAMPLE_EDS) + if self.async_test: + await self.network.aadd_node(2, SAMPLE_EDS) + await self.network.aadd_node(3, SAMPLE_EDS) + else: + self.network.add_node(2, SAMPLE_EDS) + self.network.add_node(3, SAMPLE_EDS) self.assertEqual([2, 3], [node for node in self.network]) # Check __delitem__. @@ -230,7 +272,7 @@ def test_network_item_access(self): self.assertNotEqual(self.network[3], old) self.assertEqual([3], [node for node in self.network]) - def test_network_send_periodic(self): + async def test_network_send_periodic(self): DATA1 = bytes([1, 2, 3]) DATA2 = bytes([4, 5, 6]) COB_ID = 0x123 @@ -239,7 +281,7 @@ def test_network_send_periodic(self): self.network.connect(interface="virtual") self.addCleanup(self.network.disconnect) - bus = can.Bus(interface="virtual") + bus = can.Bus(interface="virtual", loop=self.loop) self.addCleanup(bus.shutdown) acc = [] @@ -315,13 +357,88 @@ class Custom(Exception): self.assertIsNone(self.network.notifier) -class TestScanner(unittest.TestCase): + result1 = 0 + result2 = 0 + + def callback1(arg): + nonlocal result1 + result1 = arg + 1 + + def callback2(arg): + nonlocal result2 + result2 = arg * 2 + + # Check that the synchronous callbacks are called correctly + self.network.dispatch_callbacks([callback1, callback2], 5) + self.assertEqual([result1, result2], [6, 10]) + + async def async_callback(arg): + return arg + 1 + + # This is a workaround to create an async callback which we have the + # ability to clean up after the test. Logicallt its the same as calling + # async_callback directly. + coro = None + def _create_async_callback(arg): + nonlocal coro + coro = async_callback(arg) + return coro + + # Check that it's not possible to call async callbacks in a non-async context + with self.assertRaises(RuntimeError): + self.network.dispatch_callbacks([_create_async_callback], 5) + + # Cleanup + if coro is not None: + coro.close() # Close the coroutine to prevent warnings. + + async def test_dispatch_callbacks_async(self): + + result1 = 0 + result2 = 0 + + event = asyncio.Event() + + def callback(arg): + nonlocal result1 + result1 = arg + 1 + + async def async_callback(arg): + nonlocal result2 + result2 = arg * 2 + event.set() # Notify the test that the async callback is done + + # Check that both callbacks are called correctly in an async context + self.network.dispatch_callbacks([callback, async_callback], 5) + await event.wait() + self.assertEqual([result1, result2], [6, 10]) + + +class TestNetworkSync(TestNetwork): + """ Run tests in a synchronous context. """ + __test__ = True + async_test = False + + +class TestNetworkAsync(TestNetwork): + """ Run tests in an asynchronous context. """ + __test__ = True + async_test = True + + +class TestScanner(unittest.IsolatedAsyncioTestCase): TIMEOUT = 0.1 + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + def setUp(self): + self.loop = None + if self.async_test: + self.loop = asyncio.get_event_loop() self.scanner = canopen.network.NodeScanner() - def test_scanner_on_message_received(self): + async def test_scanner_on_message_received(self): # Emergency frames should be recognized. self.scanner.on_message_received(0x081) # Heartbeats should be recognized. @@ -341,23 +458,23 @@ def test_scanner_on_message_received(self): self.scanner.on_message_received(0x50e) self.assertListEqual(self.scanner.nodes, [1, 3, 5, 7, 9, 11, 13]) - def test_scanner_reset(self): + async def test_scanner_reset(self): self.scanner.nodes = [1, 2, 3] # Mock scan. self.scanner.reset() self.assertListEqual(self.scanner.nodes, []) - def test_scanner_search_no_network(self): + async def test_scanner_search_no_network(self): with self.assertRaisesRegex(RuntimeError, "No actual Network object was assigned"): self.scanner.search() - def test_scanner_search(self): - rxbus = can.Bus(interface="virtual") + async def test_scanner_search(self): + rxbus = can.Bus(interface="virtual", loop=self.loop) self.addCleanup(rxbus.shutdown) - txbus = can.Bus(interface="virtual") + txbus = can.Bus(interface="virtual", loop=self.loop) self.addCleanup(txbus.shutdown) - net = canopen.Network(txbus) + net = canopen.Network(txbus, loop=self.loop) net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 net.connect() self.addCleanup(net.disconnect) @@ -375,9 +492,9 @@ def test_scanner_search(self): # Check that no spurious packets were sent. self.assertIsNone(rxbus.recv(self.TIMEOUT)) - def test_scanner_search_limit(self): - bus = can.Bus(interface="virtual", receive_own_messages=True) - net = canopen.Network(bus) + async def test_scanner_search_limit(self): + bus = can.Bus(interface="virtual", receive_own_messages=True, loop=self.loop) + net = canopen.Network(bus, loop=self.loop) net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 net.connect() self.addCleanup(net.disconnect) @@ -392,5 +509,17 @@ def test_scanner_search_limit(self): self.assertIsNone(bus.recv(self.TIMEOUT)) +class TestScannerSync(TestScanner): + """ Run the tests in a synchronous context. """ + __test__ = True + async_test = False + + +class TestScannerAsync(TestScanner): + """ Run the tests in an asynchronous context. """ + __test__ = True + async_test = True + + if __name__ == "__main__": unittest.main() diff --git a/test/test_nmt.py b/test/test_nmt.py index 7b1b7e1d..f0a0d068 100644 --- a/test/test_nmt.py +++ b/test/test_nmt.py @@ -1,16 +1,19 @@ import threading import time import unittest +import asyncio import can import canopen +from canopen.async_guard import AllowBlocking from canopen.nmt import COMMAND_TO_STATE, NMT_COMMANDS, NMT_STATES, NmtError from .util import SAMPLE_EDS class TestNmtBase(unittest.TestCase): + def setUp(self): node_id = 2 self.node_id = node_id @@ -42,19 +45,27 @@ def test_state_set_invalid(self): self.nmt.state = "INVALID" -class TestNmtMaster(unittest.TestCase): +class TestNmtMaster(unittest.IsolatedAsyncioTestCase): NODE_ID = 2 PERIOD = 0.01 TIMEOUT = PERIOD * 10 + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + def setUp(self): - net = canopen.Network() + loop = None + if self.async_test: + loop = asyncio.get_event_loop() + + net = canopen.Network(loop=loop) net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 net.connect(interface="virtual") with self.assertLogs(): - node = net.add_node(self.NODE_ID, SAMPLE_EDS) + with AllowBlocking(): + node = net.add_node(self.NODE_ID, SAMPLE_EDS) - self.bus = can.Bus(interface="virtual") + self.bus = can.Bus(interface="virtual", loop=loop) self.net = net self.node = node @@ -67,47 +78,65 @@ def dispatch_heartbeat(self, code): hb = can.Message(arbitration_id=cob_id, data=[code]) self.bus.send(hb) - def test_nmt_master_no_heartbeat(self): + async def test_nmt_master_no_heartbeat(self): with self.assertRaisesRegex(NmtError, "heartbeat"): - self.node.nmt.wait_for_heartbeat(self.TIMEOUT) + if self.async_test: + await self.node.nmt.await_for_heartbeat(self.TIMEOUT) + else: + self.node.nmt.wait_for_heartbeat(self.TIMEOUT) with self.assertRaisesRegex(NmtError, "boot-up"): - self.node.nmt.wait_for_bootup(self.TIMEOUT) + if self.async_test: + await self.node.nmt.await_for_bootup(self.TIMEOUT) + else: + self.node.nmt.wait_for_bootup(self.TIMEOUT) - def test_nmt_master_on_heartbeat(self): + async def test_nmt_master_on_heartbeat(self): # Skip the special INITIALISING case. for code in [st for st in NMT_STATES if st != 0]: with self.subTest(code=code): t = threading.Timer(0.01, self.dispatch_heartbeat, args=(code,)) t.start() self.addCleanup(t.join) - actual = self.node.nmt.wait_for_heartbeat(0.1) + if self.async_test: + actual = await self.node.nmt.await_for_heartbeat(0.1) + else: + actual = self.node.nmt.wait_for_heartbeat(0.1) expected = NMT_STATES[code] self.assertEqual(actual, expected) - def test_nmt_master_wait_for_bootup(self): + async def test_nmt_master_wait_for_bootup(self): t = threading.Timer(0.01, self.dispatch_heartbeat, args=(0x00,)) t.start() self.addCleanup(t.join) - self.node.nmt.wait_for_bootup(self.TIMEOUT) + if self.async_test: + await self.node.nmt.await_for_bootup(self.TIMEOUT) + else: + self.node.nmt.wait_for_bootup(self.TIMEOUT) self.assertEqual(self.node.nmt.state, "PRE-OPERATIONAL") - def test_nmt_master_on_heartbeat_initialising(self): + async def test_nmt_master_on_heartbeat_initialising(self): t = threading.Timer(0.01, self.dispatch_heartbeat, args=(0x00,)) t.start() self.addCleanup(t.join) - state = self.node.nmt.wait_for_heartbeat(self.TIMEOUT) + if self.async_test: + state = await self.node.nmt.await_for_heartbeat(self.TIMEOUT) + else: + state = self.node.nmt.wait_for_heartbeat(self.TIMEOUT) self.assertEqual(state, "PRE-OPERATIONAL") - def test_nmt_master_on_heartbeat_unknown_state(self): + async def test_nmt_master_on_heartbeat_unknown_state(self): t = threading.Timer(0.01, self.dispatch_heartbeat, args=(0xcb,)) t.start() self.addCleanup(t.join) - state = self.node.nmt.wait_for_heartbeat(self.TIMEOUT) + if self.async_test: + state = await self.node.nmt.await_for_heartbeat(self.TIMEOUT) + else: + state = self.node.nmt.wait_for_heartbeat(self.TIMEOUT) # Expect the high bit to be masked out, and a formatted string to # be returned. self.assertEqual(state, "UNKNOWN STATE '75'") - def test_nmt_master_add_heartbeat_callback(self): + async def test_nmt_master_add_heartbeat_callback(self): event = threading.Event() state = None def hook(st): @@ -117,10 +146,13 @@ def hook(st): self.node.nmt.add_heartbeat_callback(hook) self.dispatch_heartbeat(0x7f) - self.assertTrue(event.wait(self.TIMEOUT)) + if self.async_test: + await asyncio.to_thread(event.wait, self.TIMEOUT) + else: + self.assertTrue(event.wait(self.TIMEOUT)) self.assertEqual(state, 127) - def test_nmt_master_node_guarding(self): + async def test_nmt_master_node_guarding(self): self.node.nmt.start_node_guarding(self.PERIOD) msg = self.bus.recv(self.TIMEOUT) self.assertIsNotNone(msg) @@ -135,60 +167,95 @@ def test_nmt_master_node_guarding(self): self.assertIsNone(self.bus.recv(self.TIMEOUT)) -class TestNmtSlave(unittest.TestCase): +class TestNmtMasterSync(TestNmtMaster): + """ Run tests in non-asynchronous mode. """ + __test__ = True + async_test = False + + +class TestNmtMasterAsync(TestNmtMaster): + """ Run tests in asynchronous mode. """ + __test__ = True + async_test = True + + +class TestNmtSlave(unittest.IsolatedAsyncioTestCase): + + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + def setUp(self): - self.network1 = canopen.Network() + loop = None + if self.async_test: + loop = asyncio.get_event_loop() + + self.network1 = canopen.Network(loop=loop) self.network1.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 self.network1.connect("test", interface="virtual") with self.assertLogs(): - self.remote_node = self.network1.add_node(2, SAMPLE_EDS) + with AllowBlocking(): + self.remote_node = self.network1.add_node(2, SAMPLE_EDS) - self.network2 = canopen.Network() + self.network2 = canopen.Network(loop=loop) self.network2.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 self.network2.connect("test", interface="virtual") with self.assertLogs(): self.local_node = self.network2.create_node(2, SAMPLE_EDS) - self.remote_node2 = self.network1.add_node(3, SAMPLE_EDS) + with AllowBlocking(): + self.remote_node2 = self.network1.add_node(3, SAMPLE_EDS) self.local_node2 = self.network2.create_node(3, SAMPLE_EDS) def tearDown(self): self.network1.disconnect() self.network2.disconnect() - def test_start_two_remote_nodes(self): + async def test_start_two_remote_nodes(self): self.remote_node.nmt.state = "OPERATIONAL" # Line below is just so that we are sure the client have received the command # before we do the check - time.sleep(0.1) + if self.async_test: + await asyncio.sleep(0.1) + else: + time.sleep(0.1) slave_state = self.local_node.nmt.state self.assertEqual(slave_state, "OPERATIONAL") self.remote_node2.nmt.state = "OPERATIONAL" # Line below is just so that we are sure the client have received the command # before we do the check - time.sleep(0.1) + if self.async_test: + await asyncio.sleep(0.1) + else: + time.sleep(0.1) slave_state = self.local_node2.nmt.state self.assertEqual(slave_state, "OPERATIONAL") - def test_stop_two_remote_nodes_using_broadcast(self): + async def test_stop_two_remote_nodes_using_broadcast(self): # This is a NMT broadcast "Stop remote node" # ie. set the node in STOPPED state self.network1.send_message(0, [2, 0]) # Line below is just so that we are sure the slaves have received the command # before we do the check - time.sleep(0.1) + if self.async_test: + await asyncio.sleep(0.1) + else: + time.sleep(0.1) slave_state = self.local_node.nmt.state self.assertEqual(slave_state, "STOPPED") slave_state = self.local_node2.nmt.state self.assertEqual(slave_state, "STOPPED") - def test_heartbeat(self): + async def test_heartbeat(self): self.assertEqual(self.remote_node.nmt.state, "INITIALISING") self.assertEqual(self.local_node.nmt.state, "INITIALISING") self.local_node.nmt.state = "OPERATIONAL" - self.local_node.sdo[0x1017].raw = 100 - time.sleep(0.2) + if self.async_test: + await self.local_node.sdo[0x1017].aset_raw(100) + await asyncio.sleep(0.2) + else: + self.local_node.sdo[0x1017].raw = 100 + time.sleep(0.2) self.assertEqual(self.remote_node.nmt.state, "OPERATIONAL") self.local_node.nmt.stop_heartbeat() @@ -202,5 +269,17 @@ def test_heartbeat_no_producer_time(self): node.nmt.state = "PRE-OPERATIONAL" +class TestNmtSlaveSync(TestNmtSlave): + """ Run tests in non-asynchronous mode. """ + __test__ = True + async_test = False + + +class TestNmtSlaveAsync(TestNmtSlave): + """ Run tests in asynchronous mode. """ + __test__ = True + async_test = True + + if __name__ == "__main__": unittest.main() diff --git a/test/test_node.py b/test/test_node.py index 973cd664..10bb4bff 100644 --- a/test/test_node.py +++ b/test/test_node.py @@ -1,4 +1,5 @@ import unittest +import asyncio import canopen @@ -27,21 +28,26 @@ def test_invalid_node_id(self): _ = canopen.node.base.BaseNode(128, canopen.ObjectDictionary()) -class TestLocalNode(unittest.TestCase): +class TestLocalNode(unittest.IsolatedAsyncioTestCase): - @classmethod - def setUpClass(cls): - cls.network = canopen.Network() - cls.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 - cls.network.connect(interface="virtual") + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool - cls.node = canopen.LocalNode(2, canopen.objectdictionary.ObjectDictionary()) + def setUp(self): + loop = None + if self.async_test: + loop = asyncio.get_event_loop() - @classmethod - def tearDownClass(cls): - cls.network.disconnect() + self.network = canopen.Network(loop=loop) + self.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network.connect(interface="virtual") - def test_associate_network(self): + self.node = canopen.LocalNode(2, canopen.objectdictionary.ObjectDictionary()) + + def tearDown(self): + self.network.disconnect() + + async def test_associate_network(self): # Need to store the number of subscribers before associating because the # network implementation automatically adds subscribers to the list n_subscribers = count_subscribers(self.network) @@ -76,21 +82,38 @@ def test_associate_network(self): self.node.remove_network() -class TestRemoteNode(unittest.TestCase): +class TestLocalNodeSync(TestLocalNode): + """ Run the tests in non-asynchronous mode. """ + __test__ = True + async_test = False + + +class TestLocalNodeAsync(TestLocalNode): + """ Run the tests in asynchronous mode. """ + __test__ = True + async_test = True + + +class TestRemoteNode(unittest.IsolatedAsyncioTestCase): + + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool - @classmethod - def setUpClass(cls): - cls.network = canopen.Network() - cls.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 - cls.network.connect(interface="virtual") + def setUp(self): + loop = None + if self.async_test: + loop = asyncio.get_event_loop() - cls.node = canopen.RemoteNode(2, canopen.objectdictionary.ObjectDictionary()) + self.network = canopen.Network(loop=loop) + self.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network.connect(interface="virtual") - @classmethod - def tearDownClass(cls): - cls.network.disconnect() + self.node = canopen.RemoteNode(2, canopen.objectdictionary.ObjectDictionary()) - def test_associate_network(self): + def tearDown(self): + self.network.disconnect() + + async def test_associate_network(self): # Need to store the number of subscribers before associating because the # network implementation automatically adds subscribers to the list n_subscribers = count_subscribers(self.network) @@ -102,6 +125,7 @@ def test_associate_network(self): self.assertIs(self.node.tpdo.network, self.network) self.assertIs(self.node.rpdo.network, self.network) self.assertIs(self.node.nmt.network, self.network) + self.assertIs(self.node.emcy.network, self.network) # Test that its not possible to associate the network multiple times with self.assertRaises(RuntimeError) as cm: @@ -117,7 +141,20 @@ def test_associate_network(self): self.assertIs(self.node.tpdo.network, uninitalized) self.assertIs(self.node.rpdo.network, uninitalized) self.assertIs(self.node.nmt.network, uninitalized) + self.assertIs(self.node.emcy.network, uninitalized) self.assertEqual(count_subscribers(self.network), n_subscribers) # Test that its possible to deassociate the network multiple times self.node.remove_network() + + +class TestRemoteNodeSync(TestRemoteNode): + """ Run the tests in non-asynchronous mode. """ + __test__ = True + async_test = False + + +class TestRemoteNodeAsync(TestRemoteNode): + """ Run the tests in asynchronous mode. """ + __test__ = True + async_test = True diff --git a/test/test_pdo.py b/test/test_pdo.py index 1b641147..336fc66e 100644 --- a/test/test_pdo.py +++ b/test/test_pdo.py @@ -5,7 +5,11 @@ from .util import SAMPLE_EDS, tmp_file -class TestPDO(unittest.TestCase): +class TestPDO(unittest.IsolatedAsyncioTestCase): + + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + def setUp(self): node = canopen.LocalNode(1, SAMPLE_EDS) pdo = node.pdo.tx[1] @@ -17,37 +21,42 @@ def setUp(self): pdo.add_variable('BOOLEAN value 2', length=1) # 0x2006 # Write some values - pdo['INTEGER16 value'].raw = -3 - pdo['UNSIGNED8 value'].raw = 0xf - pdo['INTEGER8 value'].raw = -2 - pdo['INTEGER32 value'].raw = 0x01020304 - pdo['BOOLEAN value'].raw = False - pdo['BOOLEAN value 2'].raw = True + # Async: moved to set_values() method self.pdo = pdo self.node = node - def test_pdo_map_bit_mapping(self): - self.assertEqual(self.pdo.data, b'\xfd\xff\xef\x04\x03\x02\x01\x02') + async def test_pdo_map_bit_mapping(self): + await self.set_values() + if not self.async_test: + self.assertEqual(self.pdo.data, b'\xfd\xff\xef\x04\x03\x02\x01\x02') + else: + self.assertEqual(self.pdo.data, b'\x0c\x00\xce\xbc\x9a\x78\x56\x01') - def test_pdo_map_getitem(self): - pdo = self.pdo - self.assertEqual(pdo['INTEGER16 value'].raw, -3) - self.assertEqual(pdo['UNSIGNED8 value'].raw, 0xf) - self.assertEqual(pdo['INTEGER8 value'].raw, -2) - self.assertEqual(pdo['INTEGER32 value'].raw, 0x01020304) - self.assertEqual(pdo['BOOLEAN value'].raw, False) - self.assertEqual(pdo['BOOLEAN value 2'].raw, True) - - def test_pdo_getitem(self): + async def set_values(self): + """Initialize the PDO with some valuues. + + Do this in a separate method in order to be abel to use the + async and sync versions of the tests. + """ node = self.node - self.assertEqual(node.tpdo[1]['INTEGER16 value'].raw, -3) - self.assertEqual(node.tpdo[1]['UNSIGNED8 value'].raw, 0xf) - self.assertEqual(node.tpdo[1]['INTEGER8 value'].raw, -2) - self.assertEqual(node.tpdo[1]['INTEGER32 value'].raw, 0x01020304) - self.assertEqual(node.tpdo['INTEGER32 value'].raw, 0x01020304) - self.assertEqual(node.tpdo[1]['BOOLEAN value'].raw, False) - self.assertEqual(node.tpdo[1]['BOOLEAN value 2'].raw, True) + pdo = node.pdo.tx[1] + if not self.async_test: + # Write some values + pdo['INTEGER16 value'].raw = -3 + pdo['UNSIGNED8 value'].raw = 0xf + pdo['INTEGER8 value'].raw = -2 + pdo['INTEGER32 value'].raw = 0x01020304 + pdo['BOOLEAN value'].raw = False + pdo['BOOLEAN value 2'].raw = True + else: + # Write some values (different from the synchronous values) + await pdo['INTEGER16 value'].aset_raw(12) + await pdo['UNSIGNED8 value'].aset_raw(0xe) + await pdo['INTEGER8 value'].aset_raw(-4) + await pdo['INTEGER32 value'].aset_raw(0x56789abc) + await pdo['BOOLEAN value'].aset_raw(True) + await pdo['BOOLEAN value 2'].aset_raw(False) # Test different types of access by_mapping_record = node.pdo[0x1A00] @@ -106,24 +115,102 @@ def test_pdo_maps_iterate(self): pdo = node.tpdo[1] self.assertEqual(len(pdo), sum(1 for _ in pdo)) - def test_pdo_save(self): - self.node.tpdo.save() - self.node.rpdo.save() - - def test_pdo_save_skip_readonly(self): + async def test_pdo_map_getitem(self): + await self.set_values() + pdo = self.pdo + if not self.async_test: + self.assertEqual(pdo['INTEGER16 value'].raw, -3) + self.assertEqual(pdo['UNSIGNED8 value'].raw, 0xf) + self.assertEqual(pdo['INTEGER8 value'].raw, -2) + self.assertEqual(pdo['INTEGER32 value'].raw, 0x01020304) + self.assertEqual(pdo['BOOLEAN value'].raw, False) + self.assertEqual(pdo['BOOLEAN value 2'].raw, True) + else: + self.assertEqual(await pdo['INTEGER16 value'].aget_raw(), 12) + self.assertEqual(await pdo['UNSIGNED8 value'].aget_raw(), 0xe) + self.assertEqual(await pdo['INTEGER8 value'].aget_raw(), -4) + self.assertEqual(await pdo['INTEGER32 value'].aget_raw(), 0x56789abc) + self.assertEqual(await pdo['BOOLEAN value'].aget_raw(), True) + self.assertEqual(await pdo['BOOLEAN value 2'].aget_raw(), False) + + async def test_pdo_getitem(self): + await self.set_values() + node = self.node + if not self.async_test: + self.assertEqual(node.tpdo[1]['INTEGER16 value'].raw, -3) + self.assertEqual(node.tpdo[1]['UNSIGNED8 value'].raw, 0xf) + self.assertEqual(node.tpdo[1]['INTEGER8 value'].raw, -2) + self.assertEqual(node.tpdo[1]['INTEGER32 value'].raw, 0x01020304) + self.assertEqual(node.tpdo['INTEGER32 value'].raw, 0x01020304) + self.assertEqual(node.tpdo[1]['BOOLEAN value'].raw, False) + self.assertEqual(node.tpdo[1]['BOOLEAN value 2'].raw, True) + + # Test different types of access + self.assertEqual(node.pdo[0x1600]['INTEGER16 value'].raw, -3) + self.assertEqual(node.pdo['INTEGER16 value'].raw, -3) + self.assertEqual(node.pdo.tx[1]['INTEGER16 value'].raw, -3) + self.assertEqual(node.pdo[0x2001].raw, -3) + self.assertEqual(node.tpdo[0x2001].raw, -3) + self.assertEqual(node.pdo[0x2002].raw, 0xf) + self.assertEqual(node.pdo['0x2002'].raw, 0xf) + self.assertEqual(node.tpdo[0x2002].raw, 0xf) + self.assertEqual(node.pdo[0x1600][0x2002].raw, 0xf) + else: + self.assertEqual(await node.tpdo[1]['INTEGER16 value'].aget_raw(), 12) + self.assertEqual(await node.tpdo[1]['UNSIGNED8 value'].aget_raw(), 0xe) + self.assertEqual(await node.tpdo[1]['INTEGER8 value'].aget_raw(), -4) + self.assertEqual(await node.tpdo[1]['INTEGER32 value'].aget_raw(), 0x56789abc) + self.assertEqual(await node.tpdo['INTEGER32 value'].aget_raw(), 0x56789abc) + self.assertEqual(await node.tpdo[1]['BOOLEAN value'].aget_raw(), True) + self.assertEqual(await node.tpdo[1]['BOOLEAN value 2'].aget_raw(), False) + + # Test different types of access + self.assertEqual(await node.pdo[0x1600]['INTEGER16 value'].aget_raw(), 12) + self.assertEqual(await node.pdo['INTEGER16 value'].aget_raw(), 12) + self.assertEqual(await node.pdo.tx[1]['INTEGER16 value'].aget_raw(), 12) + self.assertEqual(await node.pdo[0x2001].aget_raw(), 12) + self.assertEqual(await node.tpdo[0x2001].aget_raw(), 12) + self.assertEqual(await node.pdo[0x2002].aget_raw(), 0xe) + self.assertEqual(await node.pdo['0x2002'].aget_raw(), 0xe) + self.assertEqual(await node.tpdo[0x2002].aget_raw(), 0xe) + self.assertEqual(await node.pdo[0x1600][0x2002].aget_raw(), 0xe) + + async def test_pdo_save(self): + await self.set_values() + if not self.async_test: + self.node.tpdo.save() + self.node.rpdo.save() + else: + await self.node.tpdo.asave() + await self.node.rpdo.asave() + + async def test_pdo_save_skip_readonly(self): """Expect no exception when a record entry is not writable.""" - # Saving only happens with a defined COB ID and for specified parameters - self.node.tpdo[1].cob_id = self.node.tpdo[1].predefined_cob_id - self.node.tpdo[1].trans_type = 1 - self.node.tpdo[1].map_array[1].od.access_type = "r" - self.node.tpdo[1].save() - - self.node.tpdo[2].cob_id = self.node.tpdo[2].predefined_cob_id - self.node.tpdo[2].trans_type = 1 - self.node.tpdo[2].com_record[2].od.access_type = "r" - self.node.tpdo[2].save() - - def test_pdo_export(self): + await self.set_values() + if not self.async_test: + # Saving only happens with a defined COB ID and for specified parameters + self.node.tpdo[1].cob_id = self.node.tpdo[1].predefined_cob_id + self.node.tpdo[1].trans_type = 1 + self.node.tpdo[1].map_array[1].od.access_type = "r" + self.node.tpdo[1].save() + + self.node.tpdo[2].cob_id = self.node.tpdo[2].predefined_cob_id + self.node.tpdo[2].trans_type = 1 + self.node.tpdo[2].com_record[2].od.access_type = "r" + self.node.tpdo[2].save() + else: + # Saving only happens with a defined COB ID and for specified parameters + self.node.tpdo[1].cob_id = self.node.tpdo[1].predefined_cob_id + self.node.tpdo[1].trans_type = 1 + self.node.tpdo[1].map_array[1].od.access_type = "r" + await self.node.tpdo[1].asave() + + self.node.tpdo[2].cob_id = self.node.tpdo[2].predefined_cob_id + self.node.tpdo[2].trans_type = 1 + self.node.tpdo[2].com_record[2].od.access_type = "r" + await self.node.tpdo[2].asave() + + async def test_pdo_export(self): try: import canmatrix except ImportError: @@ -140,5 +227,17 @@ def test_pdo_export(self): self.assertIn("Frame Name", header) +class TestPDOSync(TestPDO): + """ Test the functions in synchronous mode. """ + __test__ = True + async_test = False + + +class TestPDOAsync(TestPDO): + """ Test the functions in asynchronous mode. """ + __test__ = True + async_test = True + + if __name__ == "__main__": unittest.main() diff --git a/test/test_sdo.py b/test/test_sdo.py index ebf7a8f3..ac55d198 100644 --- a/test/test_sdo.py +++ b/test/test_sdo.py @@ -1,6 +1,8 @@ import unittest +import asyncio import canopen +from canopen.async_guard import AllowBlocking import canopen.objectdictionary.datatypes as dt from canopen.objectdictionary import ODVariable @@ -11,17 +13,20 @@ RX = 2 -class TestSDOVariables(unittest.TestCase): +class TestSDOVariables(unittest.IsolatedAsyncioTestCase): """Some basic assumptions on the behavior of SDO variable objects. Mostly what is stated in the API docs. """ + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + def setUp(self): node = canopen.LocalNode(1, SAMPLE_EDS) self.sdo_node = node.sdo - def test_record_iter_length(self): + async def test_record_iter_length(self): """Assume the "highest subindex supported" entry is not counted. Sub-objects without an OD entry should be skipped as well. @@ -31,7 +36,7 @@ def test_record_iter_length(self): self.assertEqual(len(record), 3) self.assertEqual(subs, 3) - def test_array_iter_length(self): + async def test_array_iter_length(self): """Assume the "highest subindex supported" entry is not counted.""" array = self.sdo_node[0x1003] subs = sum(1 for _ in iter(array)) @@ -42,11 +47,15 @@ def test_array_iter_length(self): subs = sum(1 for _ in iter(array)) self.assertEqual(subs, 8) - def test_array_members_dynamic(self): + async def test_array_members_dynamic(self): """Check if sub-objects missing from OD entry are generated dynamically.""" array = self.sdo_node[0x1003] - for var in array.values(): - self.assertIsInstance(var, canopen.sdo.SdoVariable) + if self.async_test: + async for i in array: + self.assertIsInstance(array[i], canopen.sdo.SdoVariable) + else: + for var in array.values(): + self.assertIsInstance(var, canopen.sdo.SdoVariable) def test_array_contains_non_int(self): """SdoArray.__contains__ should handle non-int types gracefully.""" @@ -58,12 +67,27 @@ def test_get_variable_not_found(self): self.assertIsNone(self.sdo_node.get_variable(0x9999)) -class TestSDO(unittest.TestCase): +class TestSDOVariablesSync(TestSDOVariables): + """ Run tests in non-asynchronous mode. """ + __test__ = True + async_test = False + + +class TestSDOVariablesAsync(TestSDOVariables): + """ Run tests in asynchronous mode. """ + __test__ = True + async_test = True + + +class TestSDO(unittest.IsolatedAsyncioTestCase): """ Test SDO traffic by example. Most are taken from http://www.canopensolutions.com/english/about_canopen/device_configuration_canopen.shtml """ + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + def _send_message(self, can_id, data, remote=False): """Will be used instead of the usual Network.send_message method. @@ -80,21 +104,30 @@ def _send_message(self, can_id, data, remote=False): self.message_sent = True def setUp(self): - network = canopen.Network() + loop = None + if self.async_test: + loop = asyncio.get_event_loop() + + network = canopen.Network(loop=loop) network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 network.send_message = self._send_message - node = network.add_node(2, SAMPLE_EDS) + with AllowBlocking(): + node = network.add_node(2, SAMPLE_EDS) node.sdo.RESPONSE_TIMEOUT = 0.01 self.network = network - self.message_sent = False + def tearDown(self): + self.network.disconnect() - def test_expedited_upload(self): + async def test_expedited_upload(self): self.data = [ (TX, b'\x40\x18\x10\x01\x00\x00\x00\x00'), (RX, b'\x43\x18\x10\x01\x04\x00\x00\x00') ] - vendor_id = self.network[2].sdo[0x1018][1].raw + if self.async_test: + vendor_id = await self.network[2].sdo[0x1018][1].aget_raw() + else: + vendor_id = self.network[2].sdo[0x1018][1].raw self.assertEqual(vendor_id, 4) # UNSIGNED8 without padded data part (see issue #5) @@ -110,29 +143,38 @@ def test_expedited_upload(self): (TX, b'\x40\x00\x14\x02\x00\x00\x00\x00'), # upload initiate 0x1400:02 (RX, b'\x42\x00\x14\x02\xfe\x00\x00\x00'), # expedited, no size indicated ] - trans_type = self.network[2].sdo[0x1400]['Transmission type RPDO 1'].raw + if self.async_test: + trans_type = await self.network[2].sdo[0x1400]['Transmission type RPDO 1'].aget_raw() + else: + trans_type = self.network[2].sdo[0x1400]['Transmission type RPDO 1'].raw self.assertEqual(trans_type, 254) self.assertTrue(self.message_sent) - def test_size_not_specified(self): + async def test_size_not_specified(self): self.data = [ (TX, b'\x40\x00\x14\x02\x00\x00\x00\x00'), (RX, b'\x42\x00\x14\x02\xfe\x00\x00\x00') ] # This method used to truncate to 1 byte, but returns raw content now - data = self.network[2].sdo.upload(0x1400, 2) + if self.async_test: + data = await self.network[2].sdo.aupload(0x1400, 2) + else: + data = self.network[2].sdo.upload(0x1400, 2) self.assertEqual(data, b'\xfe\x00\x00\x00') self.assertTrue(self.message_sent) - def test_expedited_download(self): + async def test_expedited_download(self): self.data = [ (TX, b'\x2b\x17\x10\x00\xa0\x0f\x00\x00'), (RX, b'\x60\x17\x10\x00\x00\x00\x00\x00') ] - self.network[2].sdo[0x1017].raw = 4000 + if self.async_test: + await self.network[2].sdo[0x1017].aset_raw(4000) + else: + self.network[2].sdo[0x1017].raw = 4000 self.assertTrue(self.message_sent) - def test_segmented_upload(self): + async def test_segmented_upload(self): self.data = [ (TX, b'\x40\x08\x10\x00\x00\x00\x00\x00'), (RX, b'\x41\x08\x10\x00\x1A\x00\x00\x00'), @@ -145,7 +187,10 @@ def test_segmented_upload(self): (TX, b'\x70\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x15\x69\x6E\x73\x20\x21\x00\x00') ] - device_name = self.network[2].sdo[0x1008].raw + if self.async_test: + device_name = await self.network[2].sdo[0x1008].aget_raw() + else: + device_name = self.network[2].sdo[0x1008].raw self.assertEqual(device_name, "Tiny Node - Mega Domains !") def test_segmented_upload_too_much_data(self): @@ -159,7 +204,7 @@ def test_segmented_upload_too_much_data(self): device_name = self.network[2].sdo[0x1008].raw self.assertEqual(device_name, "Tiny") - def test_segmented_download(self): + async def test_segmented_download(self): self.data = [ (TX, b'\x21\x00\x20\x00\x0d\x00\x00\x00'), (RX, b'\x60\x00\x20\x00\x00\x00\x00\x00'), @@ -168,9 +213,12 @@ def test_segmented_download(self): (TX, b'\x13\x73\x74\x72\x69\x6e\x67\x00'), (RX, b'\x30\x00\x20\x00\x00\x00\x00\x00') ] - self.network[2].sdo['Writable string'].raw = 'A long string' + if self.async_test: + await self.network[2].sdo['Writable string'].aset_raw('A long string') + else: + self.network[2].sdo['Writable string'].raw = 'A long string' - def test_block_download(self): + async def test_block_download(self): self.data = [ (TX, b'\xc6\x00\x20\x00\x1e\x00\x00\x00'), (RX, b'\xa4\x00\x20\x00\x7f\x00\x00\x00'), @@ -184,21 +232,27 @@ def test_block_download(self): (RX, b'\xa1\x00\x00\x00\x00\x00\x00\x00') ] data = b'A really really long string...' - with self.network[2].sdo['Writable string'].open( - 'wb', size=len(data), block_transfer=True) as fp: - fp.write(data) - - def test_segmented_download_zero_length(self): + if self.async_test: + self.skipTest("Async SDO block download not implemented yet") + else: + with self.network[2].sdo['Writable string'].open( + 'wb', size=len(data), block_transfer=True) as fp: + fp.write(data) + + async def test_segmented_download_zero_length(self): self.data = [ (TX, b'\x21\x00\x20\x00\x00\x00\x00\x00'), (RX, b'\x60\x00\x20\x00\x00\x00\x00\x00'), (TX, b'\x0F\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x20\x00\x00\x00\x00\x00\x00\x00'), ] - self.network[2].sdo[0x2000].raw = "" + if self.async_test: + await self.network[2].sdo[0x2000].aset_raw("") + else: + self.network[2].sdo[0x2000].raw = "" self.assertTrue(self.message_sent) - def test_block_upload(self): + async def test_block_upload(self): self.data = [ (TX, b'\xa4\x08\x10\x00\x7f\x00\x00\x00'), (RX, b'\xc6\x08\x10\x00\x1a\x00\x00\x00'), @@ -211,11 +265,14 @@ def test_block_upload(self): (RX, b'\xc9\x40\xe1\x00\x00\x00\x00\x00'), (TX, b'\xa1\x00\x00\x00\x00\x00\x00\x00') ] - with self.network[2].sdo[0x1008].open('r', block_transfer=True) as fp: - data = fp.read() + if self.async_test: + self.skipTest("Async SDO block upload not implemented yet") + else: + with self.network[2].sdo[0x1008].open('r', block_transfer=True) as fp: + data = fp.read() self.assertEqual(data, 'Tiny Node - Mega Domains !') - def test_sdo_block_upload_retransmit(self): + async def test_sdo_block_upload_retransmit(self): """Trigger a retransmit by only validating a block partially.""" self.data = [ (TX, b'\xa4\x08\x10\x00\x7f\x00\x00\x00'), @@ -516,11 +573,14 @@ def test_sdo_block_upload_retransmit(self): (RX, b'\xc9\x3b\x49\x00\x00\x00\x00\x00'), (TX, b'\xa1\x00\x00\x00\x00\x00\x00\x00'), # --> Transfer ends without issues ] - with self.network[2].sdo[0x1008].open('r', block_transfer=True) as fp: - data = fp.read() + if self.async_test: + self.skipTest("Async SDO block upload not implemented yet") + else: + with self.network[2].sdo[0x1008].open('r', block_transfer=True) as fp: + data = fp.read() self.assertEqual(data, 39 * 'the crazy fox jumps over the lazy dog\n') - def test_writable_file(self): + async def test_writable_file(self): self.data = [ (TX, b'\x20\x00\x20\x00\x00\x00\x00\x00'), (RX, b'\x60\x00\x20\x00\x00\x00\x00\x00'), @@ -531,31 +591,65 @@ def test_writable_file(self): (TX, b'\x0f\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x20\x00\x20\x00\x00\x00\x00\x00') ] - with self.network[2].sdo['Writable string'].open('wb') as fp: - fp.write(b'1234') - fp.write(b'56789') - self.assertTrue(fp.closed) - # Write on closed file - with self.assertRaises(ValueError): - fp.write(b'123') - - def test_abort(self): + if self.async_test: + self.skipTest("Async SDO writable file not implemented yet") + else: + with self.network[2].sdo['Writable string'].open('wb') as fp: + fp.write(b'1234') + fp.write(b'56789') + self.assertTrue(fp.closed) + # Write on closed file + with self.assertRaises(ValueError): + fp.write(b'123') + + async def test_abort(self): self.data = [ (TX, b'\x40\x18\x10\x01\x00\x00\x00\x00'), (RX, b'\x80\x18\x10\x01\x11\x00\x09\x06') ] - with self.assertRaises(canopen.SdoAbortedError) as cm: - _ = self.network[2].sdo[0x1018][1].raw + if self.async_test: + with self.assertRaises(canopen.SdoAbortedError) as cm: + _ = await self.network[2].sdo[0x1018][1].aget_raw() + else: + with self.assertRaises(canopen.SdoAbortedError) as cm: + _ = self.network[2].sdo[0x1018][1].raw self.assertEqual(cm.exception.code, 0x06090011) - def test_add_sdo_channel(self): + async def test_add_sdo_channel(self): client = self.network[2].add_sdo(0x123456, 0x234567) self.assertIn(client, self.network[2].sdo_channels) + async def test_async_protection(self): + self.data = [ + (TX, b'\x40\x18\x10\x01\x00\x00\x00\x00'), + (RX, b'\x43\x18\x10\x01\x04\x00\x00\x00') + ] + if self.async_test: + # Test that regular commands are not allowed in async mode + with self.assertRaises(RuntimeError): + _ = self.network[2].sdo[0x1018][1].raw + else: + self.skipTest("No async protection test needed in sync mode") + + +class TestSDOSync(TestSDO): + """ Run tests in synchronous mode. """ + __test__ = True + async_test = False + -class TestSDOClientDatatypes(unittest.TestCase): +class TestSDOAsync(TestSDO): + """ Run tests in asynchronous mode. """ + __test__ = True + async_test = True + + +class TestSDOClientDatatypes(unittest.IsolatedAsyncioTestCase): """Test the SDO client uploads with the different data types in CANopen.""" + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + def _send_message(self, can_id, data, remote=False): """Will be used instead of the usual Network.send_message method. @@ -570,85 +664,117 @@ def _send_message(self, can_id, data, remote=False): self.network.notify(0x582, self.data.pop(0)[1], 0.0) def setUp(self): - network = canopen.Network() + loop = None + if self.async_test: + loop = asyncio.get_event_loop() + + network = canopen.Network(loop=loop) network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 network.send_message = self._send_message - node = network.add_node(2, DATATYPES_EDS) + with AllowBlocking(): + node = network.add_node(2, DATATYPES_EDS) node.sdo.RESPONSE_TIMEOUT = 0.01 self.node = node self.network = network - def test_boolean(self): + def tearDown(self): + self.network.disconnect() + + async def test_boolean(self): self.data = [ (TX, b'\x40\x01\x20\x00\x00\x00\x00\x00'), (RX, b'\x4f\x01\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2000 + dt.BOOLEAN, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.BOOLEAN, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.BOOLEAN, 0) self.assertEqual(data, b'\xfe') - def test_unsigned8(self): + async def test_unsigned8(self): self.data = [ (TX, b'\x40\x05\x20\x00\x00\x00\x00\x00'), (RX, b'\x4f\x05\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED8, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.UNSIGNED8, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED8, 0) self.assertEqual(data, b'\xfe') - def test_unsigned16(self): + async def test_unsigned16(self): self.data = [ (TX, b'\x40\x06\x20\x00\x00\x00\x00\x00'), (RX, b'\x4b\x06\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED16, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.UNSIGNED16, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED16, 0) self.assertEqual(data, b'\xfe\xfd') - def test_unsigned24(self): + async def test_unsigned24(self): self.data = [ (TX, b'\x40\x16\x20\x00\x00\x00\x00\x00'), (RX, b'\x47\x16\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED24, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.UNSIGNED24, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED24, 0) self.assertEqual(data, b'\xfe\xfd\xfc') - def test_unsigned32(self): + async def test_unsigned32(self): self.data = [ (TX, b'\x40\x07\x20\x00\x00\x00\x00\x00'), (RX, b'\x43\x07\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED32, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.UNSIGNED32, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED32, 0) self.assertEqual(data, b'\xfe\xfd\xfc\xfb') - def test_unsigned40(self): + async def test_unsigned40(self): self.data = [ (TX, b'\x40\x18\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x18\x20\x00\xfe\xfd\xfc\xfb'), (TX, b'\x60\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x05\xb2\x01\x20\x02\x91\x12\x03'), ] - data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED40, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.UNSIGNED40, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED40, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91') - def test_unsigned48(self): + async def test_unsigned48(self): self.data = [ (TX, b'\x40\x19\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x19\x20\x00\xfe\xfd\xfc\xfb'), (TX, b'\x60\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x03\xb2\x01\x20\x02\x91\x12\x03'), ] - data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED48, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.UNSIGNED48, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED48, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91\x12') - def test_unsigned56(self): + async def test_unsigned56(self): self.data = [ (TX, b'\x40\x1a\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x1a\x20\x00\xfe\xfd\xfc\xfb'), (TX, b'\x60\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x01\xb2\x01\x20\x02\x91\x12\x03'), ] - data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED56, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.UNSIGNED56, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED56, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91\x12\x03') - def test_unsigned64(self): + async def test_unsigned64(self): self.data = [ (TX, b'\x40\x1b\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x1b\x20\x00\xfe\xfd\xfc\xfb'), @@ -657,72 +783,96 @@ def test_unsigned64(self): (TX, b'\x70\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x1d\x19\x21\x70\xfe\xfd\xfc\xfb'), ] - data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED64, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.UNSIGNED64, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.UNSIGNED64, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91\x12\x03\x19') - def test_integer8(self): + async def test_integer8(self): self.data = [ (TX, b'\x40\x02\x20\x00\x00\x00\x00\x00'), (RX, b'\x4f\x02\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2000 + dt.INTEGER8, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.INTEGER8, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.INTEGER8, 0) self.assertEqual(data, b'\xfe') - def test_integer16(self): + async def test_integer16(self): self.data = [ (TX, b'\x40\x03\x20\x00\x00\x00\x00\x00'), (RX, b'\x4b\x03\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2000 + dt.INTEGER16, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.INTEGER16, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.INTEGER16, 0) self.assertEqual(data, b'\xfe\xfd') - def test_integer24(self): + async def test_integer24(self): self.data = [ (TX, b'\x40\x10\x20\x00\x00\x00\x00\x00'), (RX, b'\x47\x10\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2000 + dt.INTEGER24, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.INTEGER24, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.INTEGER24, 0) self.assertEqual(data, b'\xfe\xfd\xfc') - def test_integer32(self): + async def test_integer32(self): self.data = [ (TX, b'\x40\x04\x20\x00\x00\x00\x00\x00'), (RX, b'\x43\x04\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2000 + dt.INTEGER32, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.INTEGER32, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.INTEGER32, 0) self.assertEqual(data, b'\xfe\xfd\xfc\xfb') - def test_integer40(self): + async def test_integer40(self): self.data = [ (TX, b'\x40\x12\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x12\x20\x00\xfe\xfd\xfc\xfb'), (TX, b'\x60\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x05\xb2\x01\x20\x02\x91\x12\x03'), ] - data = self.network[2].sdo.upload(0x2000 + dt.INTEGER40, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.INTEGER40, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.INTEGER40, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91') - def test_integer48(self): + async def test_integer48(self): self.data = [ (TX, b'\x40\x13\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x13\x20\x00\xfe\xfd\xfc\xfb'), (TX, b'\x60\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x03\xb2\x01\x20\x02\x91\x12\x03'), ] - data = self.network[2].sdo.upload(0x2000 + dt.INTEGER48, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.INTEGER48, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.INTEGER48, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91\x12') - def test_integer56(self): + async def test_integer56(self): self.data = [ (TX, b'\x40\x14\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x14\x20\x00\xfe\xfd\xfc\xfb'), (TX, b'\x60\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x01\xb2\x01\x20\x02\x91\x12\x03'), ] - data = self.network[2].sdo.upload(0x2000 + dt.INTEGER56, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.INTEGER56, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.INTEGER56, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91\x12\x03') - def test_integer64(self): + async def test_integer64(self): self.data = [ (TX, b'\x40\x15\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x15\x20\x00\xfe\xfd\xfc\xfb'), @@ -731,18 +881,24 @@ def test_integer64(self): (TX, b'\x70\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x1d\x19\x21\x70\xfe\xfd\xfc\xfb'), ] - data = self.network[2].sdo.upload(0x2000 + dt.INTEGER64, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.INTEGER64, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.INTEGER64, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91\x12\x03\x19') - def test_real32(self): + async def test_real32(self): self.data = [ (TX, b'\x40\x08\x20\x00\x00\x00\x00\x00'), (RX, b'\x43\x08\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2000 + dt.REAL32, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.REAL32, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.REAL32, 0) self.assertEqual(data, b'\xfe\xfd\xfc\xfb') - def test_real64(self): + async def test_real64(self): self.data = [ (TX, b'\x40\x11\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x11\x20\x00\xfe\xfd\xfc\xfb'), @@ -751,10 +907,13 @@ def test_real64(self): (TX, b'\x70\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x1d\x19\x21\x70\xfe\xfd\xfc\xfb'), ] - data = self.network[2].sdo.upload(0x2000 + dt.REAL64, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.REAL64, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.REAL64, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91\x12\x03\x19') - def test_visible_string(self): + async def test_visible_string(self): self.data = [ (TX, b'\x40\x09\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x09\x20\x00\x1A\x00\x00\x00'), @@ -767,10 +926,13 @@ def test_visible_string(self): (TX, b'\x70\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x15\x69\x6E\x73\x20\x21\x00\x00') ] - data = self.network[2].sdo.upload(0x2000 + dt.VISIBLE_STRING, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.VISIBLE_STRING, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.VISIBLE_STRING, 0) self.assertEqual(data, b'Tiny Node - Mega Domains !') - def test_unicode_string(self): + async def test_unicode_string(self): self.data = [ (TX, b'\x40\x0b\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x0b\x20\x00\x1A\x00\x00\x00'), @@ -783,10 +945,13 @@ def test_unicode_string(self): (TX, b'\x70\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x15\x69\x6E\x73\x20\x21\x00\x00') ] - data = self.network[2].sdo.upload(0x2000 + dt.UNICODE_STRING, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.UNICODE_STRING, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.UNICODE_STRING, 0) self.assertEqual(data, b'Tiny Node - Mega Domains !') - def test_octet_string(self): + async def test_octet_string(self): self.data = [ (TX, b'\x40\x0a\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x0a\x20\x00\x1A\x00\x00\x00'), @@ -799,10 +964,13 @@ def test_octet_string(self): (TX, b'\x70\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x15\x69\x6E\x73\x20\x21\x00\x00') ] - data = self.network[2].sdo.upload(0x2000 + dt.OCTET_STRING, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.OCTET_STRING, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.OCTET_STRING, 0) self.assertEqual(data, b'Tiny Node - Mega Domains !') - def test_domain(self): + async def test_domain(self): self.data = [ (TX, b'\x40\x0f\x20\x00\x00\x00\x00\x00'), (RX, b'\x41\x0f\x20\x00\x1A\x00\x00\x00'), @@ -815,19 +983,25 @@ def test_domain(self): (TX, b'\x70\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x15\x69\x6E\x73\x20\x21\x00\x00') ] - data = self.network[2].sdo.upload(0x2000 + dt.DOMAIN, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2000 + dt.DOMAIN, 0) + else: + data = self.network[2].sdo.upload(0x2000 + dt.DOMAIN, 0) self.assertEqual(data, b'Tiny Node - Mega Domains !') - def test_unknown_od_32(self): + async def test_unknown_od_32(self): """Test an unknown OD entry of 32 bits (4 bytes).""" self.data = [ (TX, b'\x40\xFF\x20\x00\x00\x00\x00\x00'), (RX, b'\x43\xFF\x20\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x20FF, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x20FF, 0) + else: + data = self.network[2].sdo.upload(0x20FF, 0) self.assertEqual(data, b'\xfe\xfd\xfc\xfb') - def test_unknown_od_112(self): + async def test_unknown_od_112(self): """Test an unknown OD entry of 112 bits (14 bytes).""" self.data = [ (TX, b'\x40\xFF\x20\x00\x00\x00\x00\x00'), @@ -837,10 +1011,13 @@ def test_unknown_od_112(self): (TX, b'\x70\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x11\x19\x21\x70\xfe\xfd\xfc\xfb'), ] - data = self.network[2].sdo.upload(0x20FF, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x20FF, 0) + else: + data = self.network[2].sdo.upload(0x20FF, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91\x12\x03\x19\x21\x70\xfe\xfd\xfc\xfb') - def test_unknown_datatype32(self): + async def test_unknown_datatype32(self): """Test an unknown datatype, but known OD, of 32 bits (4 bytes).""" # Add fake entry 0x2100 to OD, using fake datatype 0xFF if 0x2100 not in self.node.object_dictionary: @@ -851,10 +1028,13 @@ def test_unknown_datatype32(self): (TX, b'\x40\x00\x21\x00\x00\x00\x00\x00'), (RX, b'\x43\x00\x21\x00\xfe\xfd\xfc\xfb') ] - data = self.network[2].sdo.upload(0x2100, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2100, 0) + else: + data = self.network[2].sdo.upload(0x2100, 0) self.assertEqual(data, b'\xfe\xfd\xfc\xfb') - def test_unknown_datatype112(self): + async def test_unknown_datatype112(self): """Test an unknown datatype, but known OD, of 112 bits (14 bytes).""" # Add fake entry 0x2100 to OD, using fake datatype 0xFF if 0x2100 not in self.node.object_dictionary: @@ -869,8 +1049,24 @@ def test_unknown_datatype112(self): (TX, b'\x70\x00\x00\x00\x00\x00\x00\x00'), (RX, b'\x11\x19\x21\x70\xfe\xfd\xfc\xfb'), ] - data = self.network[2].sdo.upload(0x2100, 0) + if self.async_test: + data = await self.network[2].sdo.aupload(0x2100, 0) + else: + data = self.network[2].sdo.upload(0x2100, 0) self.assertEqual(data, b'\xb2\x01\x20\x02\x91\x12\x03\x19\x21\x70\xfe\xfd\xfc\xfb') + +class TestSDOClientDatatypesSync(TestSDOClientDatatypes): + """ Run tests in synchronous mode. """ + __test__ = True + async_test = False + + +class TestSDOClientDatatypesAsync(TestSDOClientDatatypes): + """ Run tests in asynchronous mode. """ + __test__ = True + async_test = True + + if __name__ == "__main__": unittest.main() diff --git a/test/test_sync.py b/test/test_sync.py index 66c4867d..aefea06f 100644 --- a/test/test_sync.py +++ b/test/test_sync.py @@ -1,5 +1,6 @@ import threading import unittest +import asyncio import can @@ -10,26 +11,34 @@ TIMEOUT = PERIOD * 10 -class TestSync(unittest.TestCase): +class TestSync(unittest.IsolatedAsyncioTestCase): + + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + def setUp(self): - self.net = canopen.Network() + loop = None + if self.async_test: + loop = asyncio.get_event_loop() + + self.net = canopen.Network(loop=loop) self.net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 self.net.connect(interface="virtual") self.sync = canopen.sync.SyncProducer(self.net) - self.rxbus = can.Bus(interface="virtual") + self.rxbus = can.Bus(interface="virtual", loop=loop) def tearDown(self): self.net.disconnect() self.rxbus.shutdown() - def test_sync_producer_transmit(self): + async def test_sync_producer_transmit(self): self.sync.transmit() msg = self.rxbus.recv(TIMEOUT) self.assertIsNotNone(msg) self.assertEqual(msg.arbitration_id, 0x80) self.assertEqual(msg.dlc, 0) - def test_sync_producer_transmit_count(self): + async def test_sync_producer_transmit_count(self): self.sync.transmit(2) msg = self.rxbus.recv(TIMEOUT) self.assertIsNotNone(msg) @@ -37,11 +46,11 @@ def test_sync_producer_transmit_count(self): self.assertEqual(msg.dlc, 1) self.assertEqual(msg.data, b"\x02") - def test_sync_producer_start_invalid_period(self): + async def test_sync_producer_start_invalid_period(self): with self.assertRaises(ValueError): self.sync.start(0) - def test_sync_producer_start(self): + async def test_sync_producer_start(self): self.sync.start(PERIOD) self.addCleanup(self.sync.stop) @@ -85,5 +94,17 @@ def test_sync_producer_restart(self): self.sync.start(PERIOD) +class TestSyncSync(TestSync): + """ Test the functions in synchronous mode. """ + __test__ = True + async_test = False + + +class TestSyncAsync(TestSync): + """ Test the functions in asynchronous mode. """ + __test__ = True + async_test = True + + if __name__ == "__main__": unittest.main() diff --git a/test/test_time.py b/test/test_time.py index fa45a444..90c8f629 100644 --- a/test/test_time.py +++ b/test/test_time.py @@ -1,3 +1,4 @@ +import asyncio import struct import time import unittest @@ -8,17 +9,26 @@ import canopen.timestamp -class TestTime(unittest.TestCase): +class TestTime(unittest.IsolatedAsyncioTestCase): - def test_epoch(self): + __test__ = False # This is a base class, tests should not be run directly. + async_test: bool + + def setUp(self): + self.loop = None + if self.async_test: + self.loop = asyncio.get_event_loop() + + async def test_epoch(self): """Verify that the epoch matches the standard definition.""" epoch = datetime.strptime( "1984-01-01 00:00:00 +0000", "%Y-%m-%d %H:%M:%S %z" ).timestamp() self.assertEqual(int(epoch), canopen.timestamp.OFFSET) - def test_time_producer(self): - network = canopen.Network() + async def test_time_producer(self): + network = canopen.Network(loop=self.loop) + self.addCleanup(network.disconnect) network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 network.connect(interface="virtual", receive_own_messages=True) producer = canopen.timestamp.TimeProducer(network) @@ -42,7 +52,17 @@ def test_time_producer(self): self.assertEqual(days, int(current_from_epoch) // 86400) self.assertEqual(ms, int(current_from_epoch % 86400 * 1000)) - network.disconnect() + +class TestTimeSync(TestTime): + """ Test time functions in synchronous mode. """ + __test__ = True + async_test = False + + +class TestTimeAsync(TestTime): + """ Test time functions in asynchronous mode. """ + __test__ = True + async_test = True if __name__ == "__main__": diff --git a/test/test_utils.py b/test/test_utils.py index a17cce92..18c60d2f 100644 --- a/test/test_utils.py +++ b/test/test_utils.py @@ -3,7 +3,7 @@ from canopen.utils import pretty_index -class TestUtils(unittest.TestCase): +class TestUtils(unittest.IsolatedAsyncioTestCase): def test_pretty_index(self): self.assertEqual(pretty_index(0x12ab), "0x12AB")