diff --git a/LICENSE.txt b/LICENSE.txt index bfce6f06..c96ca25f 100644 --- a/LICENSE.txt +++ b/LICENSE.txt @@ -1,6 +1,7 @@ MIT License Copyright (c) 2016 Christian Sandberg +Copyright (c) 2026 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..784766cc 100644 --- a/README.rst +++ b/README.rst @@ -8,6 +8,80 @@ 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 async use of :code:`Network` must be used in an async context. This is + required to setup the async tasks and handles proper cleanup and exception + handling. + + async with canopen.Network().connect() as network: + # do async stuff with network + +* 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'] # async + var = await sdo['Variable'].aread() # async (equivalent) + await sdo['Variable'].awrite(12) # async + +* Opt-in :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, the code is not using the async variants of the library when + it shouldn't. + + To enable it call :code:`canopen.async_guard.enable_async_guard(True)` in + your main thread/main loop. + +* 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:`BaseNode402` does not work with async + +* :code:`Bits` is not working differently in async mode. In non-async mode, + the raw value is read from the node when the :code:`Bits` object is created. + In async mode, the raw value must be manually read by calling + :code:`await bits.aread()` before accessing the bits. + Features -------- @@ -156,6 +230,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'].awrite(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'].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'].awrite(42, "phys") + 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..cfa12de7 --- /dev/null +++ b/canopen/async_guard.py @@ -0,0 +1,68 @@ +""" Utils for async """ +import asyncio +import functools +import logging +import threading +import traceback + + +_ASYNC_GUARDS: dict[int, bool] = {} +"""Per-thread boolean indicating allowance of running blocking functions. + +:code:`True` indicates that blocking functions are not allowed to be called +from the current thread. +""" + +logger = logging.getLogger(__name__) + + +def enable_async_guard(enable: bool): + """Enable or disable the async guard for the current thread. + + :param enable: True to enable the async guard, False to disable it. + """ + _ASYNC_GUARDS[threading.get_ident()] = enable + + +def is_async_guarded() -> bool: + """Check if async guard is enabled for this thread. + + :return: True if async guard is enabled, False otherwise. + """ + return _ASYNC_GUARDS.get(threading.get_ident(), False) + + +def ensure_not_async(fn=None, error_message=None): + """Guard a function from being called from the async main thread. + + This function is used to guard functions that are blocking and should not + be called from async code. If the function is called while async is + running, a RuntimeError will be raised. + + Can be used either as a plain decorator, :code:`@ensure_not_async`, or + called with an extra error message, :code:`@ensure_not_async("message")`. + + :param error_message: Optional message appended to the RuntimeError raised + when the guard trips. + """ + if isinstance(fn, str): + fn, error_message = None, fn + + def decorator(fn): + @functools.wraps(fn) + def async_guard_wrap(*args, **kwargs): + if is_async_guarded(): + st = "".join(traceback.format_stack()) + logger.debug("Traceback:\n%s", st.rstrip()) + msg = ("Calling a blocking function while running async. " + f"Function {fn.__qualname__}() " + f"in {fn.__code__.co_filename}:{fn.__code__.co_firstlineno}") + if error_message: + msg += f". {error_message}" + raise RuntimeError(msg) + return fn(*args, **kwargs) + return async_guard_wrap + + if fn is not None: + return decorator(fn) + return decorator diff --git a/canopen/emcy.py b/canopen/emcy.py index 8b7d3bff..5230c746 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,6 +26,7 @@ def __init__(self): self.active: list[EmcyError] = [] self.callbacks = [] self.emcy_received = threading.Condition() + self.network: canopen.network.Network = canopen.network._UNINITIALIZED_NETWORK def on_emcy(self, can_id, data, timestamp): code, register, data = EMCY_STRUCT.unpack(data) @@ -38,11 +41,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, ignore_errors=True) def add_callback(self, callback: Callable[[EmcyError], None]): """Get notified on EMCY messages from this node. @@ -58,6 +58,7 @@ def reset(self): self.log = [] self.active = [] + @ensure_not_async("Use async_wait() instead") def wait( self, emcy_code: Optional[int] = None, timeout: float = 10 ) -> Optional[EmcyError]: @@ -86,6 +87,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 + ) -> Optional[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..9d1a8937 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("Use afast_scan() instead") 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(' Network: self.bus = can.Bus(*args, **kwargs) logger.info("Connected to '%s'", self.bus.channel_info) if self.notifier is None: + # The notifier is started without setting the loop paramter, even + # when running in async mode. The notifier changes in sublte ways + # when the loop parameter is set. All callbacks via the Listener + # interface will be called from the separate rx thread, which is + # what canopen is designed for. The async mode of the notifier will + # send all callbacks to the event loop thread, which is not + # compatible with the blocking locks and queues used in canopen. self.notifier = can.Notifier(self.bus, self.listeners, self.NOTIFIER_CYCLE) return self @@ -136,6 +163,65 @@ def __enter__(self): def __exit__(self, type, value, traceback): self.disconnect() + async def __aenter__(self): + if self.loop is None: + self.loop = asyncio.get_running_loop() + else: + if self.loop != asyncio.get_running_loop(): + raise RuntimeError("Network is running in a different event loop") + self.thread_id = threading.get_ident() + try: + # Enter the async context for the taskgroup and leave it last + await self.exit_stack.enter_async_context(self.taskgroup) + # Cleanup the network + self.exit_stack.callback(self.disconnect) + except Exception: + await self.exit_stack.aclose() + raise + return self + + async def __aexit__(self, type, value, traceback): + # Cleanup by running the context managers in reverse order of entry. + # Please see __aenter__ for list of contexts. + return await self.exit_stack.__aexit__(type, value, traceback) + + @property + def is_running_async(self) -> bool: + """Check if canopen has been connected with async""" + return self.loop is not None + + def create_task(self, coro: Coroutine, *args, **kwarge) -> asyncio.Task: + """Create an async task. + + This function is thread-safe and can be called from any thread. If + called from the same thread as the event loop, it will use + :code:`asyncio.create_task()` directly. If called from a different + thread, it will use :code:`asyncio.run_coroutine_threadsafe()` to + schedule the task in the event loop. + + All tasks created with this function is managed by the task group + in :attr:`canopen.Network.taskgroup`, which takes care of cleaning up + the tasks when the network is closed and handles exceptions in the tasks. + + :param coro: + The coroutine to run in the event loop. + """ + if threading.get_ident() == self.thread_id: + # If we are running in the same thread as the event loop + # asyncio.create_task() can be used directly. + return self.taskgroup.create_task(coro, *args, **kwarge) + + else: + # Running in a different thread. + async def _create_task(): + return self.taskgroup.create_task(coro, *args, **kwarge) + # Since this is another thread asyncio.get_running_loop() will + # not work. We need to use the stored event loop. + future = asyncio.run_coroutine_threadsafe(_create_task(), self.loop) + # The result() will block until the _create_task() coroutine has + # been executed and it returns the actual task object. + return future.result() + def add_node( self, node: Union[int, RemoteNode, LocalNode], @@ -154,6 +240,19 @@ def add_node( :param upload_eds: Set ``True`` if EDS file should be uploaded from 0x1021. + .. note:: + Using this option will fail in async mode, since uploading the + EDS requires blocking SDO transfers during node setup. Use a + pre-fetched ``object_dictionary`` instead when running under + asyncio. + + Example of pre-fetching the object dictionary with async: + + .. code-block:: python + + od = await aimport_from_node(node_id, network) + node = network.add_node(node_id, od) + :return: The Node object that was added. """ @@ -247,11 +346,58 @@ 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, ignore_errors: Optional[bool] = None) -> None: + """Handle any exception in the callbacks. + + With self.FILTER_ERRORS set to True, exceptions in callbacks will be logged + only, and the program will continue running. This is useful for + production systems where you want to log errors but not crash the + entire application due to a single callback failure. + + With self.FILTER_ERRORS set to False, exceptions in callbacks will be raised, + which will stop the program. This is useful for development and debugging, + where you want to catch errors early and fix them. This is also + important for unit tests, as only logging errors may hide problems. + + :param exc: + The exception that was raised. + :param ignore_errors: + If True, exceptions in callbacks will be logged only, and the program + will continue running. If False, exceptions in callbacks will be raised, + which will stop the program. If None, the value of self.FILTER_ERRORS + """ + logger.exception("Exception in callback: %s", exc_info=exc) + + if ignore_errors is None: + ignore_errors = self.FILTER_ERRORS + if not ignore_errors: + raise exc + + def dispatch_callbacks(self, callbacks: list[Callable], *args, **kwargs) -> 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 + :param kwargs: + Keyword arguments to pass to the callbacks. The "ignore_errors" + keyword argument can be used to override the default error handling + behavior for this specific call. See :meth:`canopen.Network.on_error` + for details. + """ + ignore_errors = kwargs.pop("ignore_errors", None) + for callback in callbacks: + try: + result = callback(*args, **kwargs) + if result is not None and asyncio.iscoroutine(result): + self.create_task(result) + except Exception as e: + self.on_error(e, ignore_errors) + def check(self) -> None: """Check that no fatal error has occurred in the receiving thread. @@ -365,8 +511,18 @@ class MessageListener(can.Listener): def __init__(self, network: Network): self.network = network + self._warning_logged = False def on_message_received(self, msg): + + if not self._warning_logged: + self._warning_logged = True + if is_async_guarded(): + logger.warning( + "MessageListener.on_message_received() called from async mainloop. " + "This may affect the async performance." + ) + if msg.is_error_frame or msg.is_remote_frame: return @@ -374,7 +530,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..4e6d7ad3 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: @@ -135,8 +137,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 +151,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("Use await_for_heartbeat() instead") def wait_for_heartbeat(self, timeout: float = 10): """Wait until a heartbeat message is received.""" with self.state_update: @@ -158,6 +161,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("Use await_for_bootup() instead") def wait_for_bootup(self, timeout: float = 10) -> None: """Wait until a boot-up message is received.""" end_time = time.time() + timeout @@ -171,6 +179,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 +242,23 @@ 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: - heartbeat_time_ms = self._local_node.sdo[0x1017].raw - self.start_heartbeat(heartbeat_time_ms) - except KeyError: - pass + if self.network.is_running_async: + # In async mode we cannot read the heartbeat directly, so we + # create a task to read it asynchronously and start the heartbeat + # service when the read is complete. + async def start_heartbeat_async(): + try: + heartbeat_time_ms = await self._local_node.sdo[0x1017].aread() + self.start_heartbeat(heartbeat_time_ms) + except KeyError: + pass + self.network.create_task(start_heartbeat_async()) + else: + try: + heartbeat_time_ms = self._local_node.sdo[0x1017].raw + self.start_heartbeat(heartbeat_time_ms) + except KeyError: + pass 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/__init__.py b/canopen/objectdictionary/__init__.py index 038b4c5d..a4742a5e 100644 --- a/canopen/objectdictionary/__init__.py +++ b/canopen/objectdictionary/__init__.py @@ -9,6 +9,7 @@ from collections.abc import Collection, Iterator, Mapping, MutableMapping from typing import Optional, TextIO, Union +from canopen.async_guard import is_async_guarded from canopen.objectdictionary.datatypes import * from canopen.objectdictionary.datatypes import IntegerN, UnsignedN from canopen.utils import pretty_index @@ -57,6 +58,12 @@ def export_od( break else: doc_type = "eds" + if is_async_guarded(): + logger.warning( + "Opening EDS file %s in async is not recommended, " + "use a thread or pass a file-like object instead", + dest + ) dest = open(dest, 'w') opened_here = True diff --git a/canopen/objectdictionary/eds.py b/canopen/objectdictionary/eds.py index 608024f3..c05f3c38 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, is_async_guarded from canopen.objectdictionary import ( ODArray, ODRecord, @@ -31,6 +33,12 @@ def import_eds(source, node_id): if hasattr(source, "read"): fp = source else: + if is_async_guarded(): + logger.warning( + "Opening EDS file %s in async is not recommended, " + "use a thread or pass a file-like object instead", + source + ) fp = open(source) opened_here = True eds.read_file(fp) @@ -184,6 +192,7 @@ def import_eds(source, node_id): return od +@ensure_not_async("Use aimport_from_node() instead") 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 @@ -196,6 +205,7 @@ def import_from_node(node_id: int, network: canopen.network.Network): network.subscribe(0x580 + node_id, sdo_client.on_response) # Create file like object for Store EDS variable try: + # Opening an SDO channel with file-like object is not supported in async with sdo_client.open(0x1021, 0, "rt") as eds_fp: od = import_eds(eds_fp, node_id) except Exception as e: @@ -207,6 +217,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..73d9be92 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: @@ -65,11 +67,21 @@ def read(self, from_od=False): 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) + 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. @@ -338,8 +350,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 +363,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 +419,69 @@ def _raw_from(param): self.subscribe() - def save(self) -> None: - """Save PDO configuration for this map using SDO.""" + 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.aread() + 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 +494,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 +531,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 +549,25 @@ 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() + 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.aread()) + else: + await sdo.awrite(value) + def subscribe(self) -> None: """Register the PDO for reception on the network. @@ -592,6 +675,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("Use await_for_reception() instead") def wait_for_reception(self, timeout: float = 10) -> float: """Wait for the next transmit PDO. @@ -603,6 +687,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.""" @@ -675,6 +767,12 @@ def set_data(self, data: bytes): self.pdo_parent.update() + async def aget_data(self) -> bytes: + raise RuntimeError("Read of PDO data asynchronously is not supported, use regular access") + + async def aset_data(self, data: bytes): + raise RuntimeError("Writing PDO data asynchronously is not supported, use regular access") + # For compatibility Variable = PdoVariable diff --git a/canopen/sdo/base.py b/canopen/sdo/base.py index d0088ec4..627bd82b 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,9 +157,19 @@ 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].aread() # type: ignore[return-value] + def __contains__(self, subindex: object) -> bool: if not isinstance(subindex, int): return False @@ -150,8 +183,8 @@ def __init__(self, sdo_node: SdoBase, od: objectdictionary.ODVariable): self.sdo_node = sdo_node variable.Variable.__init__(self, od) - def get_data(self) -> bytes: - data = self.sdo_node.upload(self.od.index, self.od.subindex) + def _truncate_data(self, data: bytes) -> bytes: + """Truncate data to the size specified in the object dictionary.""" response_size = len(data) # If size is available through variable in OD, then use the smaller of the two sizes. @@ -164,10 +197,22 @@ def get_data(self) -> bytes: data = data[:var_size] return data + def get_data(self) -> bytes: + data = self.sdo_node.upload(self.od.index, self.od.subindex) + return self._truncate_data(data) + + async def aget_data(self) -> bytes: + data = await self.sdo_node.aupload(self.od.index, self.od.subindex) + return self._truncate_data(data) + 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 @@ -210,6 +255,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..8fe36943 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("Use aupload() instead") 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("Use adownload() instead") 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("This function is not async compatible. Use aupload() or adownload() instead") 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..dc713a90 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 * @@ -205,6 +206,21 @@ 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) + def download( self, index: int, @@ -225,3 +241,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/utils.py b/canopen/utils.py index 7ddffda3..349826a9 100644 --- a/canopen/utils.py +++ b/canopen/utils.py @@ -1,5 +1,6 @@ """Additional utility functions for canopen.""" +import asyncio from typing import Optional, Union @@ -21,3 +22,16 @@ def pretty_index(index: Optional[Union[int, str]], sub_str = f"{sub!r}" return ":".join(s for s in (index_str, sub_str) if s) + + +def is_running_async() -> bool: + """Check if the current thread is running in an async context. + + :return: :code:`True` if the current thread is running in an async context, + :code:`False` otherwise. + """ + try: + asyncio.get_running_loop() + return True + except RuntimeError: + return False diff --git a/canopen/variable.py b/canopen/variable.py index 8441ac32..f222aaae 100644 --- a/canopen/variable.py +++ b/canopen/variable.py @@ -5,7 +5,7 @@ from typing import Union from canopen import objectdictionary -from canopen.utils import pretty_index +from canopen.utils import is_running_async, pretty_index logger = logging.getLogger(__name__) @@ -35,9 +35,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 +83,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()) + + @raw.setter + def raw(self, value: Union[int, bool, float, str, bytes]): + self.set_data(self._set_raw(value)) + + 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) @@ -87,12 +100,23 @@ def raw(self) -> Union[int, bool, float, str, bytes]: logger.debug(text) return value - @raw.setter - def raw(self, value: Union[int, bool, float, str, bytes]): + 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) + + 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()) + + 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 __await__(self): + """Awaiting the variable to get its raw value.""" + return self._aget_raw().__await__() @property def phys(self) -> Union[int, bool, float, str, bytes]: @@ -102,32 +126,37 @@ 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) - if self.od.unit: - logger.debug("Physical value is %s %s", value, self.od.unit) - return value + return self._get_phys(self.raw) @phys.setter def phys(self, value: Union[int, bool, float, str, bytes]): self.raw = self.od.encode_phys(value) + 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 + @property def desc(self) -> str: """Convert to and from a description of the value as a string. :raises TypeError: If the received raw data was anything but an integer value. """ - raw_int = self.raw - if not isinstance(raw_int, int): - raise TypeError("Description of values only supported for integer objects") - value = self.od.decode_desc(raw_int) - logger.debug("Description is '%s'", value) - return value + return self._get_desc(self.raw) @desc.setter def desc(self, desc: str): self.raw = self.od.encode_desc(desc) + def _get_desc(self, raw: Union[int, bool, float, str, bytes]): + if not isinstance(raw, int): + raise TypeError("Description of values only supported for integer objects") + value = self.od.decode_desc(raw) + logger.debug("Description is '%s'", value) + return value + @property def bits(self) -> Bits: """Access bits using integers, slices, or bit descriptions.""" @@ -156,6 +185,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 self._get_phys(await self._aget_raw()) + elif fmt == "desc": + return self._get_desc(await self._aget_raw()) + raise ValueError(f"Invalid format '{fmt}'") + def write( self, value: Union[int, bool, float, str, bytes], @@ -181,13 +220,48 @@ 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_raw(self.od.encode_phys(value)) + elif fmt == "desc": + if not isinstance(value, str): + raise TypeError("fmt=desc requires a string value") + await self._aset_raw(self.od.encode_desc(value)) # type: ignore[arg-type] + class Bits(Mapping): + """Access bits using integers, slices, or bit descriptions. + + In a synchronous context, the underlying raw value is read from the + variable automatically on initialization, so the bits are immediately + accessible. In an async context, the underlying value cannot be fetched + during ``__init__`` (which cannot await), so :meth:`aread` must be called + explicitly before accessing bits. + + Similarly, in a synchronous context, changes made via ``__setitem__`` are + immediately written to the variable, but in an async context, :meth:`awrite` + must be called explicitly to write the changes. + """ def __init__(self, variable: Variable): assert variable.od.data_type in objectdictionary.datatypes.INTEGER_TYPES self.variable = variable - self.read() + + # There is a slight caveat here: is_running_async() indicates that there + # is a running event loop in the current thread, but it does not tell us + # if the canopen.Network instance is running in async mode. + self._is_not_running_async = not is_running_async() + + # To remain backwards compatible, read immediately if not running in + # an async context. + if self._is_not_running_async: + self.read() + self.raw: int @staticmethod @@ -207,7 +281,11 @@ def __getitem__(self, key: Union[slice, int, str, Collection[int]]) -> int: def __setitem__(self, key: Union[slice, int, str, Collection[int]], value: int): self.raw = self.variable.od.encode_bits( self.raw, self._get_bits(key), value) - self.write() + + # To remain backwards compatible, write immediately if not running in + # an async context. + if self._is_not_running_async: + self.write() def __iter__(self): return iter(self.variable.od.bit_definitions) @@ -221,3 +299,11 @@ def read(self): def write(self): self.variable.raw = self.raw + + async def aread(self): + raw_int = await self.variable.aread() + assert isinstance(raw_int, int) + self.raw = raw_int + + async def awrite(self): + await self.variable.awrite(self.raw) diff --git a/examples/canopen_async.py b/examples/canopen_async.py new file mode 100644 index 00000000..cb3b0599 --- /dev/null +++ b/examples/canopen_async.py @@ -0,0 +1,64 @@ +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.add_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'].awrite(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 + async with canopen.Network().connect( + interface='virtual', bitrate=1000000, receive_own_messages=True + ) as network: + + # Start two instances and run them concurrently + network.create_task(do_loop(network, 20)) + network.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..77e3f7f7 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" @@ -23,6 +24,7 @@ classifiers = [ ] dependencies = [ "python-can >= 3.0.0", + "taskgroup; python_version <= '3.13'", ] dynamic = ["version"] @@ -67,3 +69,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/README.md b/test/README.md new file mode 100644 index 00000000..78dd1f1c --- /dev/null +++ b/test/README.md @@ -0,0 +1,133 @@ +# Canopen unit tests + +This directory contains the unittests for the canopoen library. canopen use +`unittest` as the framework for testing. + + +## Testing without async + +If writing a test that doesn't require or depend on async features, tests +can be written as. + +```python +class TestVariable(unittest.TestCase): + ... +``` + +See [`test_variable.py`](`test_variable.py`) as an example + + +## Testing with async + +Since this library supports usage with async support and with regular blocking +calls, the unit tests must test both cases. This does requre a little bit more +setup in the testing. + +First create a base class that is intended to be run twice, once without async +and once with async enabled. + +```python +from .async_tests import DualSyncAsyncTestCase + +class TestEmcy(DualSyncAsyncTestCase): + __test__ = False # This is a base class that shall not run directly + + # The following attrobutes are available: + # async_test: bool # Flag if async testing is running + # loop: Optional[asyncio.AbstractEventLoop] # The loop in async mode, + # # `None` in regular mode. + + def setUp(self): + super().setUp() # Make sure this is called when overriding `setUp` + # ... do your setup + + # Any tests that doesn't depend on async, can be written as regular + # test methods + def test_emcy_error(self): + self.assertEqual(...) + + # Any tests that requre async, use `async def` + async def test_method(self): + if self.async_test: + # This is when async is enabled. + await some_async_command() + else: + # This is when async is not running + some_regular_command() +``` + +To run this class, two instances of the test class must be created. One with +async and one without: + +```python +class TestEmcySync(TestEmcy): + """Run the tests in non-asynchronous mode.""" + __test__ = True # This is test to run + async_test = False # Not async mode + +class TestEmcyAsync(TestEmcy): + """Run the tests in asynchronous mode.""" + __test__ = True # This is tests to run + async_test = True # In async mode +``` + +This results in two sets of the same tests, `TestEmcySync`, where async is not +enabled and `TestEmcyAsync` where async is enabled. + +There is nothing special about these two runs, except the value of +`self.async_test` and `self.loop`. + +```python + async def test_method(self): + if self.async_test: + # This is when async is enabled. + await some_async_command() + else: + # This is when async is not running + some_regular_command() +``` + +What the sync and async does, is run this test function twice, once with +`self.async_test` False and then a second time with `self.async_test` True. +It is the resposibility of the unittests to decide if there is a need to +differentiate the test flow between the two run. + + +### Setting up a Network instance in async + +`Network()` is the main component that have difference between usage in sync +and async mode. To use network proper in async, it's async context must be +entered in the test. + +Say that `setUp()` contains `self.network = Network()` then the following can +be added to enter and exits its async context: + +```python + async def asyncSetUp(self): + if self.async_test: + await self.network.__aenter__() + + async def asyncTearDown(self): + if self.async_test: + await self.network.__aexit__(None, None, None) +``` + + +### Async or regular test function? + +When writing tests, should I use `async def` or just `def`? + +If not making any async operations with `async` or `await` there is no need +to mark the function as `async def`. Note that the function will be run in both +sync and async mode even if its not a coroutine. + + +### Excluding async from a test + +The easiest way is to do: + +```python +def test_something(self): + if self.async_test: + self.skipTest("Async is not supported because ...") +``` diff --git a/test/async_tests.py b/test/async_tests.py new file mode 100644 index 00000000..eb408bd0 --- /dev/null +++ b/test/async_tests.py @@ -0,0 +1,30 @@ +import asyncio +import unittest +from typing import Optional + +from canopen.async_guard import enable_async_guard + + +class DualSyncAsyncTestCase(unittest.IsolatedAsyncioTestCase): + """Base class for async test cases.""" + + __test__ = False # This is a base class, tests should not be run directly. + + async_test: bool + """Flag to indicate the test mode. If True, the test will run in async + mode, otherwise it will run in sync mode.""" + + loop: Optional[asyncio.AbstractEventLoop] + """The event loop to use for async tests. This will be set in the setUp + method if async_test is True, otherwise it will be None.""" + + def setUp(self): + """Set up an object for async testing.""" + enable_async_guard(self.async_test) + loop = None + if self.async_test: + loop = asyncio.get_event_loop() + self.loop = loop + + # Add a cleanup to disable the async guard after the test + self.addCleanup(enable_async_guard, False) diff --git a/test/conftest.py b/test/conftest.py new file mode 100644 index 00000000..19722c71 --- /dev/null +++ b/test/conftest.py @@ -0,0 +1,15 @@ + +import pytest + +import canopen + + +@pytest.fixture(scope="session", autouse=True) +def enable_network_exceptions(): + """Fixture to enable exceptions in the reception threads. + + This makes sure exceptions are not swallowed in the reception threads, + which is useful for debugging and testing. + """ + canopen.Network.FILTER_ERRORS = False + yield diff --git a/test/test_emcy.py b/test/test_emcy.py index b4b54a19..75079bac 100644 --- a/test/test_emcy.py +++ b/test/test_emcy.py @@ -1,3 +1,4 @@ +import asyncio import logging import threading import unittest @@ -7,6 +8,7 @@ import canopen +from .async_tests import DualSyncAsyncTestCase TIMEOUT = 0.1 @@ -22,7 +24,29 @@ def mock_rx_thread(consumer: canopen.emcy.EmcyConsumer, func): t.join(TIMEOUT) -class TestEmcy(unittest.TestCase): +class TestEmcy(DualSyncAsyncTestCase): + + __test__ = False # This is a base class, tests should not be run directly. + + def setUp(self): + super().setUp() + + self.net = canopen.Network() + self.net.connect(interface="virtual") + self.net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.emcy = canopen.emcy.EmcyConsumer() + self.emcy.network = self.net + + async def asyncSetUp(self): + if self.async_test: + await self.net.__aenter__() + + async def asyncTearDown(self): + if self.async_test: + await self.net.__aexit__(None, None, None) + + def tearDown(self): + self.net.disconnect() def check_error(self, err, code, reg, data, ts): self.assertIsInstance(err, canopen.emcy.EmcyError) @@ -32,15 +56,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 on_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 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.on_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 +86,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.on_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 +98,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.on_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 + await self.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000) + await self.on_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,8 +113,11 @@ def test_emcy_consumer_reset(self): self.assertEqual(len(emcy.log), 0) self.assertEqual(len(emcy.active), 0) - def test_emcy_consumer_wait(self): - emcy = canopen.emcy.EmcyConsumer() + async def test_emcy_consumer_wait(self): + if self.async_test: + self.skipTest("Not implemented for async") + + emcy = self.emcy def push_err(): emcy.on_emcy(0x81, b'\x01\x20\x01\x01\x02\x03\x04\x05', 100) @@ -94,19 +130,22 @@ 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 ( self.assertLogs(level=logging.INFO), - mock_rx_thread(emcy, push_err), + mock_rx_thread(emcy, push_err), # FIXME for async ): check_err(emcy.wait(timeout=TIMEOUT)) # Check filtered wait, on success. with ( self.assertLogs(level=logging.INFO), - mock_rx_thread(emcy, push_err), + mock_rx_thread(emcy, push_err), # FIXME for async ): check_err(emcy.wait(0x2001, TIMEOUT)) @@ -120,44 +159,50 @@ def push_reset(): with mock_rx_thread(emcy, push_reset): self.assertIsNone(emcy.wait(0x9000, TIMEOUT)) - def test_emcy_consumer_multiple_callbacks(self): + async def test_emcy_consumer_multiple_callbacks(self): """Test adding multiple callbacks and their execution order.""" - emcy = canopen.emcy.EmcyConsumer() + emcy = self.emcy call_order = [] emcy.add_callback(lambda err: call_order.append('callback1')) emcy.add_callback(lambda err: call_order.append('callback2')) emcy.add_callback(lambda err: call_order.append('callback3')) - emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000) + await self.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000) self.assertEqual(call_order, ['callback1', 'callback2', 'callback3']) - def test_emcy_consumer_callback_exception_handling(self): + async def test_emcy_consumer_callback_exception_handling(self): """Test that callback exceptions don't break other callbacks or the system.""" - emcy = canopen.emcy.EmcyConsumer() + emcy = self.emcy successful_callbacks = [] emcy.add_callback(lambda err: successful_callbacks.append('success1')) emcy.add_callback( lambda err: exec('raise ValueError("Test exception in callback")') ) emcy.add_callback(lambda err: successful_callbacks.append('success2')) - emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000) + await self.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000) self.assertEqual(successful_callbacks, ['success1', 'success2']) - def test_emcy_consumer_error_reset_variants(self): + async def test_emcy_consumer_error_reset_variants(self): """Test different error reset code patterns.""" - 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) + if self.async_test: + self.skipTest("Not implemented for async") + + emcy = self.emcy + await self.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000) + await self.on_emcy(0x81, b'\x10\x90\x01\x04\x03\x02\x01\x00', 2000) self.assertEqual(len(emcy.active), 2) - emcy.on_emcy(0x81, b'\x00\x00\x00\x00\x00\x00\x00\x00', 3000) + await self.on_emcy(0x81, b'\x00\x00\x00\x00\x00\x00\x00\x00', 3000) self.assertEqual(len(emcy.active), 0) - emcy.on_emcy(0x81, b'\x01\x30\x02\x00\x01\x02\x03\x04', 4000) + await self.on_emcy(0x81, b'\x01\x30\x02\x00\x01\x02\x03\x04', 4000) self.assertEqual(len(emcy.active), 1) - emcy.on_emcy(0x81, b'\x99\x00\x01\x00\x00\x00\x00\x00', 5000) + await self.on_emcy(0x81, b'\x99\x00\x01\x00\x00\x00\x00\x00', 5000) self.assertEqual(len(emcy.active), 0) def test_emcy_consumer_wait_timeout_edge_cases(self): """Test wait method with various timeout scenarios.""" - emcy = canopen.emcy.EmcyConsumer() + if self.async_test: + self.skipTest("Not implemented for async") + + emcy = self.emcy result = emcy.wait(timeout=0) self.assertIsNone(result) result = emcy.wait(timeout=0.001) @@ -165,7 +210,10 @@ def test_emcy_consumer_wait_timeout_edge_cases(self): def test_emcy_consumer_wait_concurrent_errors(self): """Test wait method when multiple errors arrive concurrently.""" - emcy = canopen.emcy.EmcyConsumer() + if self.async_test: + self.skipTest("Not implemented for async") + + emcy = self.emcy def push_multiple_errors(): emcy.on_emcy(0x81, b'\x01\x20\x01\x01\x02\x03\x04\x05', 100) @@ -181,6 +229,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,9 +299,12 @@ def check(code, expected): check(0xffff, "Device Specific") -class TestEmcyProducer(unittest.TestCase): +class TestEmcyProducer(DualSyncAsyncTestCase): + __test__ = False # This is a base class, tests should not be run directly. def setUp(self): + super().setUp() + self.txbus = can.Bus(interface="virtual") self.rxbus = can.Bus(interface="virtual") self.net = canopen.Network(self.txbus) @@ -250,6 +313,14 @@ def setUp(self): self.emcy = canopen.emcy.EmcyProducer(0x80 + 1) self.emcy.network = self.net + async def asyncSetUp(self): + if self.async_test: + await self.net.__aenter__() + + async def asyncTearDown(self): + if self.async_test: + await self.net.__aexit__(None, None, None) + def tearDown(self): self.net.disconnect() self.txbus.shutdown() @@ -298,6 +369,18 @@ def test_emcy_producer_reset_edge_cases(self): self.check_response(b'\x00\x00\x12\xAB\xCD\x00\x00\x00') +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 + + class TestEmcyIntegration(unittest.TestCase): """Integration tests for EMCY producer and consumer.""" @@ -313,6 +396,7 @@ def setUp(self): self.producer = canopen.emcy.EmcyProducer(0x081) self.producer.network = self.net self.consumer = canopen.emcy.EmcyConsumer() + self.consumer.network = self.rx_net self.rx_net.subscribe(0x081, self.consumer.on_emcy) def tearDown(self): diff --git a/test/test_local.py b/test/test_local.py index 6ab94645..f1ab573b 100644 --- a/test/test_local.py +++ b/test/test_local.py @@ -1,43 +1,61 @@ import time import unittest +import asyncio import canopen from .util import SAMPLE_EDS +from .async_tests import DualSyncAsyncTestCase -class TestSDO(unittest.TestCase): +class TestSDO(DualSyncAsyncTestCase): """ 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. + + def setUp(self): + super().setUp() + + self.network1 = canopen.Network() + self.network1.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network1.connect("test", interface="virtual") + self.remote_node = self.network1.add_node(2, SAMPLE_EDS) + + self.network2 = canopen.Network() + self.network2.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network2.connect("test", interface="virtual") + self.local_node = self.network2.create_node(2, SAMPLE_EDS) + self.remote_node2 = self.network1.add_node(3, SAMPLE_EDS) + self.local_node2 = self.network2.create_node(3, SAMPLE_EDS) + + async def asyncSetUp(self): + if self.async_test: + await self.network1.__aenter__() + await self.network2.__aenter__() + + async def asyncTearDown(self): + if self.async_test: + await self.network1.__aexit__(None, None, None) + await self.network2.__aexit__(None, None, None) + + 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].awrite(0x99) + vendor_id = await self.remote_node.sdo[0x1400][1] + 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): + 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 @@ -46,6 +64,8 @@ def test_block_upload_switch_to_expedite_upload(self): self.assertEqual("Unexpected response 0x41", str(context.exception)) 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 +74,172 @@ 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"] + 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)"] + 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"].awrite(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"].awrite("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].awrite(0xfeff) + value = await self.local_node.sdo[0x2004] + 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].awrite("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"].awrite(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"].awrite(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 - # Line below is just so that we are sure the client have received the abort - # before we do the check - time.sleep(0.1) + async def test_receive_abort_request(self): + if self.async_test: + await self.remote_node.sdo.aabort(0x0504_0003) # Invalid sequence number + await asyncio.sleep(0.1) + else: + self.remote_node.sdo.abort(0x0504_0003) # Invalid sequence number + time.sleep(0.1) + # The delay was to ensure the abort request was received by the client + # before we check the last received error. 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"].awrite("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"].awrite("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 +251,104 @@ 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(DualSyncAsyncTestCase): """ 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. - 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) + def setUp(self): + super().setUp() - @classmethod - def tearDownClass(cls): - cls.network1.disconnect() - cls.network2.disconnect() + self.network1 = canopen.Network() + self.network1.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network1.connect("test", interface="virtual") + self.remote_node = self.network1.add_node(2, SAMPLE_EDS) - def test_read(self): + self.network2 = canopen.Network() + self.network2.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network2.connect("test", interface="virtual") + self.local_node = self.network2.create_node(2, SAMPLE_EDS) + + async def asyncSetUp(self): + if self.async_test: + await self.network1.__aenter__() + await self.network2.__aenter__() + + async def asyncTearDown(self): + if self.async_test: + await self.network1.__aexit__(None, None, None) + await self.network2.__aexit__(None, None, None) + + 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..c2bf640f 100644 --- a/test/test_network.py +++ b/test/test_network.py @@ -1,19 +1,33 @@ import logging import time import unittest +import asyncio import can import canopen from .util import SAMPLE_EDS +from .async_tests import DualSyncAsyncTestCase -class TestNetwork(unittest.TestCase): +class TestNetwork(DualSyncAsyncTestCase): + __test__ = False # This is a base class, tests should not be run directly. def setUp(self): + super().setUp() + self.network = canopen.Network() self.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.addCleanup(self.network.disconnect) + + async def asyncSetUp(self): + if self.async_test: + await self.network.__aenter__() + + async def asyncTearDown(self): + if self.async_test: + await self.network.__aexit__(None, None, None) def test_network_add_node(self): # Add using str. @@ -48,10 +62,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: + # Async doesn't support uploads_eds directly + od = await canopen.objectdictionary.eds.aimport_from_node(2, self.network) + self.network.add_node(2, od, upload_eds=False) + else: + self.network.add_node(2, SAMPLE_EDS, upload_eds=True) def test_network_create_node(self): with self.assertLogs(): @@ -87,13 +106,21 @@ 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) 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]) @@ -314,11 +341,70 @@ class Custom(Exception): # Notifier must be released even when check() raises self.assertIsNone(self.network.notifier) + async def test_dispatch_callbacks(self): + + 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 + + # Check that it's not possible to call async callbacks in a non-async context + if self.async_test: + self.network.dispatch_callbacks([async_callback], 5) + # await coro # Wait for the coroutine to finish + # assert result1 == 6 # Ensure the synchronous callback was called + else: + + # This is a workaround to create an async callback and capture + # the coroutine object so we can clean it up after the test. + # It is equivalent as calling `async_callback` directly. + coro = None + def _create_async_callback(arg): + nonlocal coro + coro = async_callback(arg) + return coro + + # In a non-async context, calling with an async callback should raise a RuntimeError + 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. + + +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(DualSyncAsyncTestCase): + __test__ = False # This is a base class, tests should not be run directly. -class TestScanner(unittest.TestCase): TIMEOUT = 0.1 def setUp(self): + super().setUp() self.scanner = canopen.network.NodeScanner() def test_scanner_on_message_received(self): @@ -350,7 +436,7 @@ 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): + async def test_scanner_search(self): rxbus = can.Bus(interface="virtual") self.addCleanup(rxbus.shutdown) @@ -360,36 +446,62 @@ def test_scanner_search(self): net = canopen.Network(txbus) net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 net.connect() - self.addCleanup(net.disconnect) - - self.scanner.network = net - self.scanner.search() - - payload = bytes([64, 0, 16, 0, 0, 0, 0, 0]) - acc = [rxbus.recv(self.TIMEOUT) for _ in range(127)] - for node_id, msg in enumerate(acc, start=1): - with self.subTest(node_id=node_id): - self.assertIsNotNone(msg) - self.assertEqual(msg.arbitration_id, 0x600 + node_id) - self.assertEqual(msg.data, payload) - # Check that no spurious packets were sent. - self.assertIsNone(rxbus.recv(self.TIMEOUT)) - - def test_scanner_search_limit(self): + + def _test(): + self.scanner.network = net + self.scanner.search() + + payload = bytes([64, 0, 16, 0, 0, 0, 0, 0]) + acc = [rxbus.recv(self.TIMEOUT) for _ in range(127)] + for node_id, msg in enumerate(acc, start=1): + with self.subTest(node_id=node_id): + self.assertIsNotNone(msg) + self.assertEqual(msg.arbitration_id, 0x600 + node_id) + self.assertEqual(msg.data, payload) + # Check that no spurious packets were sent. + self.assertIsNone(rxbus.recv(self.TIMEOUT)) + + if self.async_test: + async with net: # Run tests with async + _test() + else: + with net: # Run tests with sync + _test() + + async def test_scanner_search_limit(self): bus = can.Bus(interface="virtual", receive_own_messages=True) net = canopen.Network(bus) net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 net.connect() - self.addCleanup(net.disconnect) - self.scanner.network = net - self.scanner.search(limit=1) + def _test(): + self.scanner.network = net + self.scanner.search(limit=1) - msg = bus.recv(self.TIMEOUT) - self.assertIsNotNone(msg) - self.assertEqual(msg.arbitration_id, 0x601) - # Check that no spurious packets were sent. - self.assertIsNone(bus.recv(self.TIMEOUT)) + msg = bus.recv(self.TIMEOUT) + self.assertIsNotNone(msg) + self.assertEqual(msg.arbitration_id, 0x601) + # Check that no spurious packets were sent. + self.assertIsNone(bus.recv(self.TIMEOUT)) + + if self.async_test: + async with net: # Run tests with async + _test() + else: + with net: # Run tests with sync + _test() + + +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__": diff --git a/test/test_nmt.py b/test/test_nmt.py index 7b1b7e1d..1dd5255d 100644 --- a/test/test_nmt.py +++ b/test/test_nmt.py @@ -1,6 +1,7 @@ import threading import time import unittest +import asyncio import can @@ -8,6 +9,7 @@ from canopen.nmt import COMMAND_TO_STATE, NMT_COMMANDS, NMT_STATES, NmtError from .util import SAMPLE_EDS +from .async_tests import DualSyncAsyncTestCase class TestNmtBase(unittest.TestCase): @@ -42,12 +44,16 @@ def test_state_set_invalid(self): self.nmt.state = "INVALID" -class TestNmtMaster(unittest.TestCase): +class TestNmtMaster(DualSyncAsyncTestCase): + __test__ = False # This is a base class, tests should not be run directly. + NODE_ID = 2 PERIOD = 0.01 TIMEOUT = PERIOD * 10 def setUp(self): + super().setUp() + net = canopen.Network() net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 net.connect(interface="virtual") @@ -58,6 +64,14 @@ def setUp(self): self.net = net self.node = node + async def asyncSetUp(self): + if self.async_test: + await self.net.__aenter__() + + async def asyncTearDown(self): + if self.async_test: + await self.net.__aexit__(None, None, None) + def tearDown(self): self.net.disconnect() self.bus.shutdown() @@ -67,47 +81,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,7 +149,10 @@ 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): @@ -135,8 +170,24 @@ 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(DualSyncAsyncTestCase): + __test__ = False # This is a base class, tests should not be run directly. + def setUp(self): + super().setUp() + self.network1 = canopen.Network() self.network1.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 self.network1.connect("test", interface="virtual") @@ -151,56 +202,96 @@ def setUp(self): self.remote_node2 = self.network1.add_node(3, SAMPLE_EDS) self.local_node2 = self.network2.create_node(3, SAMPLE_EDS) + async def asyncSetUp(self): + if self.async_test: + await self.network1.__aenter__() + await self.network2.__aenter__() + + async def asyncTearDown(self): + if self.async_test: + await self.network1.__aexit__(None, None, None) + await self.network2.__aexit__(None, None, None) + 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].awrite(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() - def test_heartbeat_no_producer_time(self): + async def test_heartbeat_no_producer_time(self): # Create a node without the producer heartbeat time parameter node = canopen.LocalNode(1, canopen.ObjectDictionary()) - with self.assertRaises(KeyError): - node.sdo[0x1017].raw = 100 + self.network1.add_node(node) + if self.async_test: + with self.assertRaises(KeyError): + await node.sdo[0x1017].awrite(100) + else: + with self.assertRaises(KeyError): + node.sdo[0x1017].raw = 100 # Should not fail because of missing 0x1017 object entry 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..b8bbc9f7 100644 --- a/test/test_node.py +++ b/test/test_node.py @@ -29,17 +29,15 @@ def test_invalid_node_id(self): class TestLocalNode(unittest.TestCase): - @classmethod - def setUpClass(cls): - cls.network = canopen.Network() - cls.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 - cls.network.connect(interface="virtual") + def setUp(self): + self.network = canopen.Network() + self.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network.connect(interface="virtual") - cls.node = canopen.LocalNode(2, canopen.objectdictionary.ObjectDictionary()) + self.node = canopen.LocalNode(2, canopen.objectdictionary.ObjectDictionary()) - @classmethod - def tearDownClass(cls): - cls.network.disconnect() + def tearDown(self): + self.network.disconnect() def test_associate_network(self): # Need to store the number of subscribers before associating because the @@ -78,17 +76,15 @@ def test_associate_network(self): class TestRemoteNode(unittest.TestCase): - @classmethod - def setUpClass(cls): - cls.network = canopen.Network() - cls.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 - cls.network.connect(interface="virtual") + def setUp(self): + self.network = canopen.Network() + self.network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 + self.network.connect(interface="virtual") - cls.node = canopen.RemoteNode(2, canopen.objectdictionary.ObjectDictionary()) + self.node = canopen.RemoteNode(2, canopen.objectdictionary.ObjectDictionary()) - @classmethod - def tearDownClass(cls): - cls.network.disconnect() + def tearDown(self): + self.network.disconnect() def test_associate_network(self): # Need to store the number of subscribers before associating because the @@ -102,6 +98,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,6 +114,7 @@ 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 diff --git a/test/test_pdo.py b/test/test_pdo.py index 1b641147..021b8108 100644 --- a/test/test_pdo.py +++ b/test/test_pdo.py @@ -3,10 +3,15 @@ import canopen from .util import SAMPLE_EDS, tmp_file +from .async_tests import DualSyncAsyncTestCase -class TestPDO(unittest.TestCase): +class TestPDO(DualSyncAsyncTestCase): + __test__ = False # This is a base class, tests should not be run directly. + def setUp(self): + super().setUp() + node = canopen.LocalNode(1, SAMPLE_EDS) pdo = node.pdo.tx[1] pdo.add_variable('INTEGER16 value') # 0x2001 @@ -106,22 +111,34 @@ 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() + async def test_pdo_save(self): + if not self.async_test: + self.node.tpdo.save() + self.node.rpdo.save() + else: + await self.node.tpdo.asave() + await self.node.rpdo.asave() - def test_pdo_save_skip_readonly(self): + 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() + + if not self.async_test: + self.node.tpdo[1].save() + else: + 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" - self.node.tpdo[2].save() + + if not self.async_test: + self.node.tpdo[2].save() + else: + await self.node.tpdo[2].asave() def test_pdo_export(self): try: @@ -140,5 +157,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..9cf3072d 100644 --- a/test/test_sdo.py +++ b/test/test_sdo.py @@ -5,19 +5,22 @@ from canopen.objectdictionary import ODVariable from .util import DATATYPES_EDS, SAMPLE_EDS +from .async_tests import DualSyncAsyncTestCase TX = 1 RX = 2 -class TestSDOVariables(unittest.TestCase): +class TestSDOVariables(DualSyncAsyncTestCase): """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. def setUp(self): + super().setUp() node = canopen.LocalNode(1, SAMPLE_EDS) self.sdo_node = node.sdo @@ -31,22 +34,34 @@ 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)) - self.assertEqual(len(array), 3) - self.assertEqual(subs, 3) - # Simulate more entries getting added dynamically - array[0].set_data(b'\x08') - subs = sum(1 for _ in iter(array)) - self.assertEqual(subs, 8) - - def test_array_members_dynamic(self): + if self.async_test: + subs = len([_ async for _ in array.aiter()]) + self.assertEqual(await array.alen(), 3) + # Simulate more entries getting added dynamically + await array[0].aset_data(b'\x08') + subs = len([_ async for _ in array.aiter()]) + self.assertEqual(subs, 8) + else: + subs = sum(1 for _ in iter(array)) + self.assertEqual(len(array), 3) + self.assertEqual(subs, 3) + # Simulate more entries getting added dynamically + array[0].set_data(b'\x08') + subs = sum(1 for _ in iter(array)) + self.assertEqual(subs, 8) + + 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,11 +73,24 @@ 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(DualSyncAsyncTestCase): """ 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. def _send_message(self, can_id, data, remote=False): """Will be used instead of the usual Network.send_message method. @@ -80,6 +108,8 @@ def _send_message(self, can_id, data, remote=False): self.message_sent = True def setUp(self): + super().setUp() + network = canopen.Network() network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 network.send_message = self._send_message @@ -87,14 +117,26 @@ def setUp(self): node.sdo.RESPONSE_TIMEOUT = 0.01 self.network = network - self.message_sent = False + async def asyncSetUp(self): + if self.async_test: + await self.network.__aenter__() + + async def asyncTearDown(self): + if self.async_test: + await self.network.__aexit__(None, None, None) - def test_expedited_upload(self): + def tearDown(self): + self.network.disconnect() + + 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] + else: + vendor_id = self.network[2].sdo[0x1018][1].raw self.assertEqual(vendor_id, 4) # UNSIGNED8 without padded data part (see issue #5) @@ -102,7 +144,10 @@ def test_expedited_upload(self): (TX, b'\x40\x00\x14\x02\x00\x00\x00\x00'), # upload initiate 0x1400:02 (RX, b'\x4f\x00\x14\x02\xfe'), # expedited, size=1 ] - 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'] + else: + trans_type = self.network[2].sdo[0x1400]['Transmission type RPDO 1'].raw self.assertEqual(trans_type, 254) # Same with padding to a full SDO frame @@ -110,29 +155,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'] + 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].awrite(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,10 +199,13 @@ 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] + 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): + async def test_segmented_upload_too_much_data(self): # Server sends 5 bytes, but indicated size 4 self.data = [ (TX, b'\x40\x08\x10\x00\x00\x00\x00\x00'), # upload initiate, 0x1008:00 @@ -156,10 +213,13 @@ def test_segmented_upload_too_much_data(self): (TX, b'\x60\x00\x00\x00\x00\x00\x00\x00'), # upload segment (RX, b'\x05\x54\x69\x6E\x79\x20\x00\x00'), # segment complete, 5 bytes ] - device_name = self.network[2].sdo[0x1008].raw + if self.async_test: + device_name = await self.network[2].sdo[0x1008] + else: + 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,7 +228,10 @@ 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'].awrite('A long string') + else: + self.network[2].sdo['Writable string'].raw = 'A long string' def test_block_download(self): self.data = [ @@ -184,18 +247,23 @@ def test_block_download(self): (RX, b'\xa1\x00\x00\x00\x00\x00\x00\x00') ] data = b'A really really long string...' + if self.async_test: + self.skipTest("Async SDO block download not implemented yet") 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): + 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].awrite("") + else: + self.network[2].sdo[0x2000].raw = "" self.assertTrue(self.message_sent) def test_block_upload(self): @@ -211,6 +279,8 @@ 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') ] + if self.async_test: + self.skipTest("Async SDO block upload not implemented yet") with self.network[2].sdo[0x1008].open('r', block_transfer=True) as fp: data = fp.read() self.assertEqual(data, 'Tiny Node - Mega Domains !') @@ -516,6 +586,8 @@ 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 ] + if self.async_test: + self.skipTest("Async SDO block upload not implemented yet") 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') @@ -531,6 +603,8 @@ 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') ] + if self.async_test: + self.skipTest("Async SDO writable file not implemented yet") with self.network[2].sdo['Writable string'].open('wb') as fp: fp.write(b'1234') fp.write(b'56789') @@ -539,22 +613,52 @@ def test_writable_file(self): with self.assertRaises(ValueError): fp.write(b'123') - def test_abort(self): + 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] + 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): client = self.network[2].add_sdo(0x123456, 0x234567) self.assertIn(client, self.network[2].sdo_channels) + 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: + # Working fine in sync mode + _ = self.network[2].sdo[0x1018][1].raw -class TestSDOClientDatatypes(unittest.TestCase): + +class TestSDOSync(TestSDO): + """ Run tests in synchronous mode. """ + __test__ = True + async_test = False + + +class TestSDOAsync(TestSDO): + """ Run tests in asynchronous mode. """ + __test__ = True + async_test = True + + +class TestSDOClientDatatypes(DualSyncAsyncTestCase): """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. def _send_message(self, can_id, data, remote=False): """Will be used instead of the usual Network.send_message method. @@ -570,6 +674,8 @@ def _send_message(self, can_id, data, remote=False): self.network.notify(0x582, self.data.pop(0)[1], 0.0) def setUp(self): + super().setUp() + network = canopen.Network() network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 network.send_message = self._send_message @@ -578,77 +684,112 @@ def setUp(self): self.node = node self.network = network - def test_boolean(self): + async def asyncSetUp(self): + if self.async_test: + await self.network.__aenter__() + + async def asyncTearDown(self): + if self.async_test: + await self.network.__aexit__(None, None, None) + + 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 +798,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 +896,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 +922,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 +941,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 +960,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 +979,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 +998,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 +1026,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 +1043,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 +1064,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_time.py b/test/test_time.py index fa45a444..314cbba5 100644 --- a/test/test_time.py +++ b/test/test_time.py @@ -19,6 +19,7 @@ def test_epoch(self): def test_time_producer(self): network = canopen.Network() + self.addCleanup(network.disconnect) network.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0 network.connect(interface="virtual", receive_own_messages=True) producer = canopen.timestamp.TimeProducer(network) @@ -42,8 +43,6 @@ def test_time_producer(self): self.assertEqual(days, int(current_from_epoch) // 86400) self.assertEqual(ms, int(current_from_epoch % 86400 * 1000)) - network.disconnect() - if __name__ == "__main__": unittest.main()