From c0bb8a881cf9ce72420467e621e0122b65805ed2 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 25 Oct 2026 21:32:59 +0000 Subject: [PATCH] Cookbook lot 4: vendor .NET assembly with pythonnet Actuator and 0D detector on a simulated .NET assembly (out parameter, event, exception type, IDisposable), the helpers to start the runtime and load an assembly, and real calls to the .NET libraries of the runtime. pythonnet is a dependency; the tests of the real libraries are skipped when no runtime is available. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01Tdiso1u1ZhikSutj2Qs5Cg --- README.rst | 17 +- pyproject.toml | 1 + .../daq_move_CookbookDotNet.py | 116 ++++++++++ .../plugins_0D/daq_0Dviewer_CookbookDotNet.py | 58 +++++ .../hardware/dotnet_controller.py | 112 +++++++++ .../hardware/dotnet_runtime.py | 99 ++++++++ .../hardware/mock_dotnet_controller.py | 156 +++++++++++++ .../hardware/system_dotnet_demo.py | 98 ++++++++ tests/test_cookbook_dotnet.py | 213 ++++++++++++++++++ 9 files changed, 869 insertions(+), 1 deletion(-) create mode 100644 src/pymodaq_plugins_mockexamples/daq_move_plugins/daq_move_CookbookDotNet.py create mode 100644 src/pymodaq_plugins_mockexamples/daq_viewer_plugins/plugins_0D/daq_0Dviewer_CookbookDotNet.py create mode 100644 src/pymodaq_plugins_mockexamples/hardware/dotnet_controller.py create mode 100644 src/pymodaq_plugins_mockexamples/hardware/dotnet_runtime.py create mode 100644 src/pymodaq_plugins_mockexamples/hardware/mock_dotnet_controller.py create mode 100644 src/pymodaq_plugins_mockexamples/hardware/system_dotnet_demo.py create mode 100644 tests/test_cookbook_dotnet.py diff --git a/README.rst b/README.rst index 3772f6f..fe4bb5a 100644 --- a/README.rst +++ b/README.rst @@ -181,7 +181,7 @@ Short, runnable and tested plugins to copy from when writing a plugin for a real to a simulated device: a stage with a voltmeter (``hardware/mock_serial_visa_instrument.py``) or a 3-axis controller behind a library (``hardware/mock_dll_controller.py``). Each file starts with a header telling what to copy, what to adapt and what to keep. The tests (``tests/test_cookbook_drivers.py``, ``tests/test_cookbook_plugins.py``, -``tests/test_cookbook_dll.py``, ``tests/test_cookbook_spectrometer.py``, ``tests/test_bs_camera.py``) show how to test a +``tests/test_cookbook_dll.py``, ``tests/test_cookbook_dotnet.py``, ``tests/test_cookbook_spectrometer.py``, ``tests/test_bs_camera.py``) show how to test a plugin against a simulated device with the harness of ``pymodaq.utils.plugin_testing``. Giving the same *controller ID* to an actuator and a detector of a recipe makes them share one driver: the first @@ -220,6 +220,17 @@ changes, actuator slave of a detector ``daq_move_CookbookSpectrometer``, ``hardware/spectrometer.py`` and the detector share one instrument; a homing that waits for its end and calls ``move_done``. It also has the interface of the spectrometer extension (laser line, exposure time) +Vendor .NET assembly, multi-axis ``daq_move_CookbookDotNet`` (3 axes), A ``pythonnet`` driver: ``out`` parameter returned in a tuple, +controller, ``pythonnet`` ``daq_0Dviewer_CookbookDotNet``, an event given a Python function with ``+=`` and ``-=``, the + ``hardware/dotnet_controller.py`` exception type of the assembly caught, ``IDisposable``. The + assembly is simulated by an object used like its namespace. The + end of a move is an event raised by a thread of the assembly, + turned into a Qt signal +Starting .NET and loading an assembly ``hardware/dotnet_runtime.py``, ``load_dotnet`` (runtime per platform, clear error) and + ``hardware/system_dotnet_demo.py`` ``load_assembly`` (names, path from a setting or an environment + variable, an error that lists what was tried). Real calls to the + .NET libraries (``Int32.TryParse``, ``List``, + ``MemoryStream``, ``Timer.Elapsed`` from another thread) ====================================== ============================================== ================================================================ Patterns already shown by the mock instruments of this package @@ -242,6 +253,10 @@ Simulated device, no hardware every plugin of this package Patterns not yet covered: an instrument driven by a Python binding of the manufacturer (it is packaged like the drivers of ``hardware/``: a class that wraps the binding, created by ``ini_stage``). +The .NET recipe needs ``pythonnet`` (installed with this package) and a .NET runtime only when it talks to a real assembly: +the simulated one runs anywhere. The tests that call the .NET libraries of the runtime are skipped if there is none +(Windows has the .NET Framework; elsewhere install .NET 6 or later). + PID Models ========== diff --git a/pyproject.toml b/pyproject.toml index aca3b8d..916b5d9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,7 @@ dependencies = [ "pymodaq_data>=5.3.0", "pillow", "pymeasure>=0.14", # also brings pyvisa and pyserial: needed by the cookbook recipes of the hardware folder + "pythonnet>=3.0.3", # the cookbook recipe on a .NET assembly (it needs a .NET runtime only when it talks to one) ] diff --git a/src/pymodaq_plugins_mockexamples/daq_move_plugins/daq_move_CookbookDotNet.py b/src/pymodaq_plugins_mockexamples/daq_move_plugins/daq_move_CookbookDotNet.py new file mode 100644 index 0000000..c96be75 --- /dev/null +++ b/src/pymodaq_plugins_mockexamples/daq_move_plugins/daq_move_CookbookDotNet.py @@ -0,0 +1,116 @@ +"""Cookbook recipe 4a: multi-axis actuator on a vendor .NET assembly (``pythonnet``), with a move that ends by an event. + +COPY this file for an actuator whose manufacturer supplies a .NET assembly (``Vendor.Device.dll``) for a controller +with several axes. It is the recipe of ``daq_move_CookbookDll`` with the driver of ``hardware/dotnet_controller.py``: +read that file for what is specific to ``pythonnet`` (``out`` parameters, events, exceptions, ``IDisposable``). +ADAPT: ``params`` (what the assembly needs to connect), ``_axis_names``, ``_controller_units``, the calls of the driver +in ``move_abs``, ``move_rel``, ``get_actuator_value``, ``move_home`` and ``stop_motion``. +KEEP: +- the master/slave handling in ``ini_stage`` and ``close``: the first axis (master) opens the controller, the others + (slaves, same controller ID in the Dashboard) receive it; +- the event handling. The assembly raises the event from one of its own threads: ``_forward_event`` only emits a Qt + signal, which is delivered to ``_on_motion_event`` in the thread of the plugin. Never call ``move_done`` or touch a + widget from the thread of the assembly; +- ``user_condition_to_reach_target``: PyMoDAQ polls the position until it is within epsilon of the target, and also + waits for the end-of-move event. The homing is not polled, so ``move_done`` is called when its event arrives. + +Try it without the assembly: leave the port to ``MOCK``, the assembly is then the simulated one of +``hardware/mock_dotnet_controller.py``. The runtime of .NET and the loading of the assembly are in +``hardware/dotnet_runtime.py``. +""" +from qtpy.QtCore import Signal, Slot + +from pymodaq.control_modules.move_utility_classes import (DAQ_Move_base, comon_parameters_fun, DataActuatorType, + DataActuator, main) + +from pymodaq_plugins_mockexamples.hardware.dotnet_controller import MotionController, open_controller +from pymodaq_plugins_mockexamples.hardware.mock_dll_controller import (EVENT_MOVE_DONE, EVENT_STOPPED, + EVENT_HOME_DONE) + + +class DAQ_Move_CookbookDotNet(DAQ_Move_base): + """One axis of the simulated 3-axis controller, driven through ``MotionController`` on a .NET assembly""" + _controller_units = 'mm' + is_multiaxes = True + _axis_names = {'X': 0, 'Y': 1, 'Z': 2} # the assembly numbers its axes + _epsilons = 0.01 + data_actuator_type = DataActuatorType.DataActuator + params = comon_parameters_fun(is_multiaxes, axis_names=_axis_names, epsilon=_epsilons) + [ + {'title': 'Port:', 'name': 'port', 'type': 'str', 'value': 'MOCK', + 'tip': 'Port of the controller, MOCK for the simulated assembly'}, + {'title': 'Assembly path:', 'name': 'assembly_path', 'type': 'browsepath', 'value': '', 'filetype': True, + 'tip': 'Full path to the .dll, empty to search it (environment variable MM_ASSEMBLY, name of the assembly)'}, + ] + + motion_event = Signal(int, int) # axis, event: emitted from the thread of the assembly, received in this thread + + def ini_attributes(self): + self.controller: MotionController = None + self._move_finished = True + self._homing = False + self.motion_event.connect(self._on_motion_event) + + def ini_stage(self, controller=None): + if self.is_master: + self.controller = open_controller(self.settings['port'], self.settings['assembly_path']) + else: # a master axis shares its controller + self.controller = controller + self.controller.subscribe(self._forward_event) + info = self.controller.info() + return f'{info.name}, firmware {info.firmware[0]}.{info.firmware[1]}', True + + def close(self): + if self.controller is not None: + self.controller.unsubscribe(self._forward_event) + if self.is_master: + self.controller.close() + + def _forward_event(self, axis: int, event: int): + """Called from the thread of the assembly: only emit the signal""" + self.motion_event.emit(axis, event) + + @Slot(int, int) + def _on_motion_event(self, axis: int, event: int): + """Runs in the thread of the plugin""" + if axis != self.axis_value: # the controller is shared: events of the other axes come too + return + if event == EVENT_HOME_DONE and self._homing: + self._homing = False + self.move_done() + elif event in (EVENT_MOVE_DONE, EVENT_STOPPED): + self._move_finished = True + + def user_condition_to_reach_target(self) -> bool: + return self._move_finished + + def get_actuator_value(self) -> DataActuator: + position = DataActuator(data=self.controller.position(self.axis_value), units=self.axis_unit) + return self.get_position_with_scaling(position) + + def move_abs(self, value: DataActuator): + value = self.check_bound(value) + self.target_value = value + value = self.set_position_with_scaling(value) + self._move_finished = False + self.controller.move_abs(self.axis_value, value.value(self.axis_unit)) + + def move_rel(self, value: DataActuator): + value = self.check_bound(self.current_value + value) - self.current_value + self.target_value = value + self.current_value + value = self.set_position_with_scaling(self.target_value) + self._move_finished = False + self.controller.move_abs(self.axis_value, value.value(self.axis_unit)) + + def move_home(self): + self._homing = True + self.controller.home(self.axis_value) + + def stop_motion(self): + self.controller.stop(self.axis_value) + + def commit_settings(self, param): + pass + + +if __name__ == '__main__': + main(__file__) diff --git a/src/pymodaq_plugins_mockexamples/daq_viewer_plugins/plugins_0D/daq_0Dviewer_CookbookDotNet.py b/src/pymodaq_plugins_mockexamples/daq_viewer_plugins/plugins_0D/daq_0Dviewer_CookbookDotNet.py new file mode 100644 index 0000000..035623d --- /dev/null +++ b/src/pymodaq_plugins_mockexamples/daq_viewer_plugins/plugins_0D/daq_0Dviewer_CookbookDotNet.py @@ -0,0 +1,58 @@ +"""Cookbook recipe 4b: 0D detector reading the positions of the axes of a controller driven by a vendor .NET assembly. + +COPY this file for a detector of a controller that has several functions (here the encoders of three axes) behind one +object of the assembly. Alone it opens its own controller (master); with the same controller ID as the actuators +``CookbookDotNet`` it is a slave and shares their object: a controller must be opened once, not once per plugin. +ADAPT: ``params``, ``ini_detector``, and the acquisition in ``grab_data`` (name, labels and units of the data). +See ``daq_move_CookbookDotNet`` and ``hardware/dotnet_controller.py`` for what is specific to ``pythonnet``. +KEEP: the master/slave handling in ``ini_detector`` and ``close``, and the units given to the data. +""" +import numpy as np + +from pymodaq.control_modules.viewer_utility_classes import DAQ_Viewer_base, comon_parameters, main +from pymodaq.utils.data import DataFromPlugins +from pymodaq_data import DataToExport + +from pymodaq_plugins_mockexamples.hardware.dotnet_controller import MotionController, open_controller + + +class DAQ_0DViewer_CookbookDotNet(DAQ_Viewer_base): + """Positions of the three axes of the simulated controller, read through ``MotionController``""" + params = comon_parameters + [ + {'title': 'Port:', 'name': 'port', 'type': 'str', 'value': 'MOCK', + 'tip': 'Port of the controller, MOCK for the simulated assembly'}, + {'title': 'Assembly path:', 'name': 'assembly_path', 'type': 'browsepath', 'value': '', 'filetype': True, + 'tip': 'Full path to the .dll, empty to search it (environment variable MM_ASSEMBLY, name of the assembly)'}, + ] + + def ini_attributes(self): + self.controller: MotionController = None + + def ini_detector(self, controller=None): + if self.is_master: + self.controller = open_controller(self.settings['port'], self.settings['assembly_path']) + else: # a master actuator or detector shares its controller + self.controller = controller + info = self.controller.info() + return f'{info.name}, firmware {info.firmware[0]}.{info.firmware[1]}', True + + def close(self): + if self.is_master and self.controller is not None: + self.controller.close() + + def grab_data(self, Naverage=1, **kwargs): + positions = [np.array([self.controller.position(axis)]) for axis in range(self.controller.n_axes)] + self.dte_signal.emit(DataToExport( + name='CookbookDotNet', + data=[DataFromPlugins(name='Positions', data=positions, dim='Data0D', labels=['X', 'Y', 'Z'], + units='mm')])) + + def stop(self): + pass + + def commit_settings(self, param): + pass + + +if __name__ == '__main__': + main(__file__) diff --git a/src/pymodaq_plugins_mockexamples/hardware/dotnet_controller.py b/src/pymodaq_plugins_mockexamples/hardware/dotnet_controller.py new file mode 100644 index 0000000..5813017 --- /dev/null +++ b/src/pymodaq_plugins_mockexamples/hardware/dotnet_controller.py @@ -0,0 +1,112 @@ +"""Driver of the 3-axis motion controller of ``mock_dotnet_controller.py``, written like a driver for a vendor .NET +assembly with ``pythonnet``. + +What differs from a ``ctypes`` driver (``dll_controller.py``): there is no ``argtypes`` or ``restype`` to declare, +``pythonnet`` converts the arguments and the results, but some conventions must be known: + +- an ``out`` parameter of a .NET method is returned by ``pythonnet`` in a tuple, after the result of the method: + ``ok, position = controller.TryGetPosition(axis)``; +- an event receives a Python function with ``+=`` (it is converted to a delegate) and gives it up with ``-=``. The + function is stored in an attribute, to remove the very same object. The assembly raises the event from one of + its threads: the function must only forward it to the listeners, which must not touch Qt widgets; +- a .NET exception is a Python exception of the .NET type: catch the exception type of the assembly (here + ``MmException``) and read its ``Message`` (``str(exception)`` also holds the .NET stack trace); +- the objects that hold hardware are ``IDisposable``: call ``Dispose()`` when done (in ``close``); +- an enumeration value gives its number with ``int(...)``. + +Load the real assembly with ``MotionController(load_assembly('Mm.Net', namespace='Mm'), port)`` (see +``dotnet_runtime.py``) or the simulated one with ``MotionController.mock()``. +""" +from typing import Callable + +from pymodaq_plugins_mockexamples.hardware.dll_controller import ControllerInfo, MotionControllerError +from pymodaq_plugins_mockexamples.hardware.mock_dotnet_controller import MockMmNamespace +from pymodaq_plugins_mockexamples.hardware.mock_dll_controller import PORT, N_AXES + + +class MotionController: + """Handle of an opened controller and its axes (numbered from 0), on a .NET assembly""" + n_axes = N_AXES + + def __init__(self, api, port: str = PORT): + """``api`` is the namespace of the assembly: ``import Mm`` with ``pythonnet``, or its simulation""" + self._api = api + self._controller = self._call(api.Controller.Open, port) + self._listeners: list[Callable[[int, int], None]] = [] + self._handler = self._on_event # kept in an attribute: ``-=`` must be given the object given to ``+=`` + self._controller.MoveFinished += self._handler + + @classmethod + def mock(cls) -> 'MotionController': + """The controller on the simulated assembly (no assembly or .NET runtime needed)""" + return cls(MockMmNamespace()) + + @property + def _open(self): + """The .NET controller, refused once closed""" + if self._controller is None: + raise MotionControllerError('The controller is closed') + return self._controller + + def _call(self, method, *args): + try: + return method(*args) + except self._api.MmException as exc: + raise MotionControllerError(exc.Message) from None + + def _on_event(self, sender, args): + """Called by the assembly, from one of its threads""" + for listener in list(self._listeners): + listener(args.Axis, int(args.Event)) + + def subscribe(self, listener: Callable[[int, int], None]): + """``listener(axis, event)`` is called, from a thread of the assembly, when a move ends (see the EVENT_*)""" + self._listeners.append(listener) + + def unsubscribe(self, listener: Callable[[int, int], None]): + if listener in self._listeners: + self._listeners.remove(listener) + + def info(self) -> ControllerInfo: + info = self._call(lambda: self._open.Info) + return ControllerInfo(info.Name, info.AxisCount, (info.FirmwareMajor, info.FirmwareMinor)) + + def position(self, axis: int) -> float: + ok, position = self._open.TryGetPosition(axis) # the ``out`` parameter comes back in a tuple + if not ok: + raise MotionControllerError(f'Could not read the position of axis {axis}') + return position + + def is_moving(self, axis: int) -> bool: + return self._call(self._open.IsMoving, axis) + + def move_abs(self, axis: int, position: float): + """Start the move; the assembly raises ``MoveFinished`` when it is over""" + self._call(self._open.MoveAbs, axis, position) + + def stop(self, axis: int): + self._call(self._open.Stop, axis) + + def home(self, axis: int): + """Start the homing; the assembly raises ``MoveFinished`` when it is over""" + self._call(self._open.Home, axis) + + def close(self): + if self._controller is not None: + self._controller.MoveFinished -= self._handler + self._controller.Dispose() + self._controller = None + self._listeners.clear() + + +ASSEMBLY_NAMES = ['Mm.Net'] # name of the (imaginary) assembly, without .dll + + +def open_controller(port: str, assembly_path: str = '') -> MotionController: + """The controller of the simulated assembly for the port ``MOCK``, else of the vendor assembly found by + ``load_assembly`` (setting of the plugin, environment variable ``MM_ASSEMBLY``, name of the assembly)""" + if port == PORT: + return MotionController.mock() + from pymodaq_plugins_mockexamples.hardware.dotnet_runtime import load_assembly + return MotionController(load_assembly(ASSEMBLY_NAMES, path=assembly_path or None, env_var='MM_ASSEMBLY', + namespace='Mm'), port) diff --git a/src/pymodaq_plugins_mockexamples/hardware/dotnet_runtime.py b/src/pymodaq_plugins_mockexamples/hardware/dotnet_runtime.py new file mode 100644 index 0000000..c3a3192 --- /dev/null +++ b/src/pymodaq_plugins_mockexamples/hardware/dotnet_runtime.py @@ -0,0 +1,99 @@ +"""Start the .NET runtime and load a .NET assembly with ``pythonnet``, the way a plugin for a vendor assembly should. + +Many manufacturers give a .NET assembly (``Vendor.Device.dll``) instead of a C library. ``pythonnet`` makes its +classes importable from Python once the runtime is running and the assembly is referenced. ``load_dotnet`` and +``load_assembly`` are meant to be copied in a plugin package. Like ``load_library`` for a C library, they say what +was tried when something is missing instead of the bare exception of ``pythonnet``. + +``pythonnet`` is imported here and nowhere else at module level of the package, so that a plugin only needs it (and a +.NET runtime: the one of Windows, or ``dotnet`` / ``mono`` elsewhere) when it really talks to an assembly. +""" +import importlib +import os +import platform +from pathlib import Path +from typing import Optional, Sequence, Union + + +class DotNetNotAvailableError(RuntimeError): + """``pythonnet`` is not installed or no .NET runtime could be started""" + + +class AssemblyNotFoundError(OSError): + """No candidate could be loaded; the message lists every candidate and why it failed""" + + +def load_dotnet(runtime: Optional[str] = None): + """Start the .NET runtime (once per process) and return the ``clr`` module of ``pythonnet`` + + Parameters + ---------- + runtime: 'netfx', 'coreclr' or 'mono', optional + The runtime to start. By default ``netfx`` (the .NET Framework installed with Windows) on Windows and + ``coreclr`` (.NET 6 and later, the ``dotnet`` command) elsewhere. An assembly built for .NET Framework 4.x + usually loads on both; one built for .NET 6+ only on ``coreclr``. The runtime can be chosen only once. + + Raises + ------ + DotNetNotAvailableError + ``pythonnet`` is missing, or the runtime could not start (the message tells how to install one). + """ + try: + import pythonnet + except ImportError as e: + raise DotNetNotAvailableError('pythonnet is not installed: pip install pythonnet') from e + runtime = runtime or os.environ.get('PYTHONNET_RUNTIME') or ('netfx' if platform.system() == 'Windows' + else 'coreclr') + try: + pythonnet.load(runtime) + import clr + except Exception as e: # clr_loader raises various exceptions when the runtime is not found + raise DotNetNotAvailableError( + f'Could not start the {runtime} runtime of .NET ({e}). Install it (on Windows the .NET Framework is ' + f'already there; elsewhere install .NET 6+ or mono) or choose another one.') from e + return clr + + +def load_assembly(names: Union[str, Sequence[str]], path: Optional[Union[str, Path]] = None, + env_var: Optional[str] = None, namespace: Optional[str] = None): + """Reference a .NET assembly and return the module of its namespace, when asked + + Parameters + ---------- + names: str or list of str + Names of the assembly (``Vendor.Device``, without ``.dll``) tried in order: the first that can be loaded is + used. Several names are useful when the same classes live in different assemblies on different runtimes. + path: str or Path, optional + Full path to the ``.dll`` or to the folder that holds it (typically a setting of the plugin); tried first. + env_var: str, optional + Name of an environment variable holding such a path; tried second. + namespace: str, optional + If given, the module of this namespace (``import Vendor.Device``) is returned: its classes can then be + used, the .NET exceptions of the assembly can be caught... + + Raises + ------ + AssemblyNotFoundError + Listing everything that was tried. + """ + clr = load_dotnet() + wanted = [names] if isinstance(names, str) else list(names) + candidates = [] + for location in (path, os.environ.get(env_var) if env_var else None): + if location: + location = Path(location) + candidates.extend([location / f'{name}.dll' for name in wanted] if location.is_dir() else [location]) + candidates.extend(wanted) + + failures = [] + for candidate in candidates: + if isinstance(candidate, Path) and not candidate.is_file(): + failures.append(f'{candidate}: no such file') + continue + try: + clr.AddReference(str(candidate)) + except Exception as e: + failures.append(f'{candidate}: {e}') + continue + return importlib.import_module(namespace) if namespace else clr + raise AssemblyNotFoundError('Could not load the .NET assembly. Tried:\n ' + '\n '.join(failures)) diff --git a/src/pymodaq_plugins_mockexamples/hardware/mock_dotnet_controller.py b/src/pymodaq_plugins_mockexamples/hardware/mock_dotnet_controller.py new file mode 100644 index 0000000..876e14a --- /dev/null +++ b/src/pymodaq_plugins_mockexamples/hardware/mock_dotnet_controller.py @@ -0,0 +1,156 @@ +"""A simulated vendor .NET assembly for a 3-axis motion controller: ``MockMmNamespace`` is used like the namespace ``Mm`` +that ``pythonnet`` gives once the assembly is referenced. + +The driver (``dotnet_controller.py``) only uses what ``pythonnet`` gives: the classes of the namespace, their static +methods, properties, an ``out`` parameter (returned in a tuple by ``pythonnet``), an event that receives a Python +function with ``+=``, an exception type to catch, ``IDisposable``. This module offers the same, so the code of the +driver runs and is tested without the assembly. It is the same device as ``mock_dll_controller.py``. + +The C# API that is simulated:: + + namespace Mm { + public class MmException : Exception { } + public enum MoveEvent { MoveDone = 1, Stopped = 2, HomeDone = 3 } + public class MoveEventArgs : EventArgs { public int Axis; public MoveEvent Event; } + public class ControllerInfo { public string Name; public int AxisCount; public int FirmwareMajor, FirmwareMinor; } + public class Controller : IDisposable { + public static Controller Open(string port); + public ControllerInfo Info { get; } + public void MoveAbs(int axis, double position); + public bool TryGetPosition(int axis, out double position); + public bool IsMoving(int axis); + public void Stop(int axis); + public void Home(int axis); + public event EventHandler MoveFinished; // raised from a thread of the assembly + public void Dispose(); + } + } +""" +import enum +from typing import Callable, Optional + +from pymodaq_plugins_mockexamples.hardware.mock_dll_controller import (MockMotionController, N_AXES, RANGE, PORT, + EVENT_MOVE_DONE, EVENT_STOPPED, + EVENT_HOME_DONE) + + +class MmException(Exception): + """A .NET exception: the text is in ``Message``""" + + @property + def Message(self) -> str: + return str(self.args[0]) + + +class MoveEvent(enum.IntEnum): + MoveDone = EVENT_MOVE_DONE + Stopped = EVENT_STOPPED + HomeDone = EVENT_HOME_DONE + + +class MoveEventArgs: + def __init__(self, axis: int, event: MoveEvent): + self.Axis = axis + self.Event = event + + +class ControllerInfo: + Name = 'MockMotion .NET' + AxisCount = N_AXES + FirmwareMajor = 3 + FirmwareMinor = 1 + + +class _Event: + """A .NET event: ``controller.MoveFinished += handler`` adds a listener, ``-=`` removes it""" + + def __init__(self): + self._handlers: list[Callable] = [] + + def __iadd__(self, handler): + self._handlers.append(handler) + return self + + def __isub__(self, handler): + if handler in self._handlers: + self._handlers.remove(handler) + return self + + def invoke(self, sender, args): + for handler in list(self._handlers): + handler(sender, args) + + +class Controller: + """The simulated ``Mm.Controller``""" + speed = 25. # mm/s + + def __init__(self, port: str): + self._engine: Optional[MockMotionController] = MockMotionController(self.speed) + self._engine.callback = self._raise_event + self._move_finished = _Event() + + @staticmethod + def Open(port: str) -> 'Controller': + if port != PORT: + raise MmException(f'No controller on port {port}') + return Controller(port) + + @property + def MoveFinished(self) -> _Event: + return self._move_finished + + @MoveFinished.setter + def MoveFinished(self, event: _Event): + pass # what ``controller.MoveFinished += handler`` assigns back: the same event + + @property + def Info(self) -> ControllerInfo: + self._check_axis(0) + return ControllerInfo() + + def _check_axis(self, axis: int): + if self._engine is None: + raise MmException('The controller is disposed') + if not 0 <= axis < N_AXES: + raise MmException(f'Invalid axis {axis}') + + def _raise_event(self, axis: int, event: int): + """Called by the simulation from its own thread, like the thread of a vendor assembly""" + self._move_finished.invoke(self, MoveEventArgs(axis, MoveEvent(event))) + + def MoveAbs(self, axis: int, position: float): + self._check_axis(axis) + if not RANGE[0] <= position <= RANGE[1]: + raise MmException('Position out of range') + self._engine.move_to(axis, float(position)) + + def TryGetPosition(self, axis: int): + """``out double position``: ``pythonnet`` returns the result and the ``out`` values in a tuple""" + if self._engine is None or not 0 <= axis < N_AXES: + return False, 0. + return True, self._engine.position(axis) + + def IsMoving(self, axis: int) -> bool: + self._check_axis(axis) + return self._engine.is_moving(axis) + + def Stop(self, axis: int): + self._check_axis(axis) + self._engine.stop(axis) + + def Home(self, axis: int): + self._check_axis(axis) + self._engine.move_to(axis, 0., EVENT_HOME_DONE) + + def Dispose(self): + if self._engine is not None: + self._engine.close() + self._engine = None + + +class MockMmNamespace: + """Used instead of ``import Mm``: ``api = MockMmNamespace()``, then ``api.Controller.Open(...)``""" + Controller = Controller + MmException = MmException + MoveEvent = MoveEvent diff --git a/src/pymodaq_plugins_mockexamples/hardware/system_dotnet_demo.py b/src/pymodaq_plugins_mockexamples/hardware/system_dotnet_demo.py new file mode 100644 index 0000000..56ae22d --- /dev/null +++ b/src/pymodaq_plugins_mockexamples/hardware/system_dotnet_demo.py @@ -0,0 +1,98 @@ +"""Calls to the real .NET libraries of the runtime with ``pythonnet``: no vendor assembly needed to see the techniques. + +Each function shows one thing a driver for a vendor assembly does: receive an ``out`` parameter, catch a .NET +exception, pass a typed array or a generic collection, give a Python function to an event raised by another thread, +dispose an object. They run on the .NET Framework (Windows) and on .NET 6+ (``coreclr``). See ``dotnet_runtime.py`` +for the start of the runtime and ``mock_dotnet_controller.py`` for the same techniques on a simulated assembly. +""" +import threading +from typing import Callable, Optional, Sequence + +from pymodaq_plugins_mockexamples.hardware.dotnet_runtime import load_dotnet, load_assembly + + +def parse_integer(text: str) -> Optional[int]: + """``static bool Int32.TryParse(string s, out int result)``: ``pythonnet`` returns ``(bool, result)``""" + load_dotnet() + from System import Int32 + ok, value = Int32.TryParse(text) + return value if ok else None + + +def parse_integer_or_raise(text: str) -> int: + """``static int Int32.Parse(string s)`` throws a ``FormatException``: catch the .NET type, read its ``Message``""" + load_dotnet() + from System import Int32, FormatException + try: + return Int32.Parse(text) + except FormatException as exc: + raise ValueError(exc.Message) from None # str(exc) also holds the .NET stack trace + + +def sum_of_list(values: Sequence[float]) -> float: + """A generic collection: ``List[float]`` is ``List`` (the type is given between brackets)""" + load_assembly(['System.Collections']) + from System.Collections.Generic import List + numbers = List[float]() + for value in values: + numbers.Add(value) + return sum(numbers) # a .NET collection is iterable + + +def bytes_through_a_stream(data: bytes) -> bytes: + """A typed array and an ``IDisposable``: ``Array[Byte]`` is ``byte[]``, ``Dispose`` is called in ``finally``""" + load_dotnet() + from System import Array, Byte + from System.IO import MemoryStream + stream = MemoryStream() + try: + buffer = Array[Byte](data) + stream.Write(buffer, 0, buffer.Length) + return bytes(stream.ToArray()) + finally: + stream.Dispose() + + +def call_back_from_another_thread(on_elapsed: Callable[[], None], delay_ms: int = 10) -> Callable[[], None]: + """An event raised by a thread of the runtime: ``timer.Elapsed += function`` + + ``System.Timers.Timer`` raises ``Elapsed`` from a thread of the .NET thread pool, as a vendor assembly raises its + own events. ``on_elapsed`` is converted to a delegate and called in that thread: it must be thread safe, and + only emit a Qt signal. The function returned stops and releases the timer: call it, or the event and the delegate + stay alive. + """ + load_assembly(['System.ComponentModel.TypeConverter', 'System']) # the assembly that holds Timer depends on the runtime + from System.Timers import Timer + timer = Timer(delay_ms) + timer.AutoReset = False + + def handler(sender, args): + on_elapsed() + + timer.Elapsed += handler + + def release(): + timer.Stop() + timer.Elapsed -= handler # the very same object given to += + timer.Dispose() + + timer.Start() + return release + + +def thread_of_the_event(delay_ms: int = 10, timeout: float = 5.) -> threading.Thread: + """The Python thread in which the event handler was called (not the main one)""" + done = threading.Event() + threads = [] + + def on_elapsed(): + threads.append(threading.current_thread()) + done.set() + + release = call_back_from_another_thread(on_elapsed, delay_ms) + try: + if not done.wait(timeout): + raise TimeoutError('The event was not raised') + finally: + release() + return threads[0] diff --git a/tests/test_cookbook_dotnet.py b/tests/test_cookbook_dotnet.py new file mode 100644 index 0000000..4e38836 --- /dev/null +++ b/tests/test_cookbook_dotnet.py @@ -0,0 +1,213 @@ +"""The .NET assembly recipe: starting the runtime, the ``pythonnet`` driver on the simulated assembly, the plugins on it + +The first tests call the real .NET libraries of the runtime (.NET Framework on Windows, .NET 6+ elsewhere) and are +skipped if ``pythonnet`` or a runtime is missing. The others need neither. To test your own driver copy the last part +and replace the plugin classes. +""" +import threading +import time + +import pytest + +from pymodaq.utils.data import DataActuator +from pymodaq.utils.plugin_testing import (make_actuator, make_detector, move_abs_and_wait, move_rel_and_wait, + move_home_and_wait, grab_and_wait) + +from pymodaq_plugins_mockexamples.daq_move_plugins.daq_move_CookbookDotNet import DAQ_Move_CookbookDotNet +from pymodaq_plugins_mockexamples.daq_viewer_plugins.plugins_0D.daq_0Dviewer_CookbookDotNet import \ + DAQ_0DViewer_CookbookDotNet +from pymodaq_plugins_mockexamples.hardware import mock_dll_controller as mock +from pymodaq_plugins_mockexamples.hardware import system_dotnet_demo +from pymodaq_plugins_mockexamples.hardware.dll_controller import MotionControllerError +from pymodaq_plugins_mockexamples.hardware.dotnet_controller import MotionController, open_controller +from pymodaq_plugins_mockexamples.hardware.dotnet_runtime import (load_dotnet, load_assembly, DotNetNotAvailableError, + AssemblyNotFoundError) +from pymodaq_plugins_mockexamples.hardware.mock_dotnet_controller import MockMmNamespace + + +@pytest.fixture(scope='module') +def runtime(): + """The .NET runtime, or skip the tests of the real .NET libraries if there is none on this machine""" + try: + return load_dotnet() + except DotNetNotAvailableError as e: + pytest.skip(str(e)) + + +class TestSystemLibraries: + """Real calls to the libraries of the runtime""" + + def test_out_parameter(self, runtime): + assert system_dotnet_demo.parse_integer('42') == 42 + assert system_dotnet_demo.parse_integer('forty-two') is None + + def test_dotnet_exception(self, runtime): + assert system_dotnet_demo.parse_integer_or_raise('7') == 7 + with pytest.raises(ValueError, match='not in a correct format'): + system_dotnet_demo.parse_integer_or_raise('x') + + def test_generic_collection(self, runtime): + assert system_dotnet_demo.sum_of_list([1., 2.5, 3.]) == pytest.approx(6.5) + + def test_typed_array_and_dispose(self, runtime): + assert system_dotnet_demo.bytes_through_a_stream(b'\x00\x01\xffabc') == b'\x00\x01\xffabc' + + def test_event_raised_by_another_thread(self, runtime): + thread = system_dotnet_demo.thread_of_the_event() + assert thread is not threading.main_thread() + + def test_missing_assembly_says_what_was_tried(self, runtime, tmp_path): + missing = tmp_path / 'Nowhere.dll' + with pytest.raises(AssemblyNotFoundError) as exc: + load_assembly(['Not.An.Assembly'], path=missing) + assert str(missing) in str(exc.value) + assert 'Not.An.Assembly' in str(exc.value) + + def test_second_candidate_is_used(self, runtime): + assert load_assembly(['Not.An.Assembly', 'System.Collections']) is not None + + +def test_no_runtime_is_a_clear_error(monkeypatch): + monkeypatch.setitem(__import__('sys').modules, 'pythonnet', None) # import pythonnet fails + with pytest.raises(DotNetNotAvailableError, match='pip install pythonnet'): + load_dotnet() + + +class TestMockAssembly: + def test_behaves_like_a_namespace_of_pythonnet(self): + api = MockMmNamespace() + controller = api.Controller.Open('MOCK') + try: + ok, position = controller.TryGetPosition(0) # an out parameter comes back in a tuple + assert ok and position == 0. + with pytest.raises(api.MmException) as exc: + api.Controller.Open('COM1') + assert 'COM1' in exc.value.Message + finally: + controller.Dispose() + + def test_unknown_member(self): + with pytest.raises(AttributeError): + MockMmNamespace().Controller.Open('MOCK').DoesNotExist() + + +@pytest.fixture +def controller(): + ctl = MotionController.mock() + yield ctl + ctl.close() + + +class TestDriver: + def test_info(self, controller): + info = controller.info() + assert info.name == 'MockMotion .NET' + assert info.n_axes == 3 + assert info.firmware == (3, 1) + + def test_move_and_event_from_another_thread(self, controller): + events = [] + controller.subscribe(lambda axis, event: events.append((axis, event, threading.current_thread()))) + controller.move_abs(1, 5.) + deadline = time.monotonic() + 5 + while not events and time.monotonic() < deadline: + time.sleep(0.01) + assert events[0][:2] == (1, mock.EVENT_MOVE_DONE) + assert events[0][2] is not threading.main_thread() + assert controller.position(1) == pytest.approx(5.) + + def test_dotnet_exceptions_become_driver_errors(self, controller): + with pytest.raises(MotionControllerError, match='out of range'): + controller.move_abs(0, 500.) + with pytest.raises(MotionControllerError, match='Invalid axis'): + controller.move_abs(7, 1.) + + def test_stop_reports_a_stopped_event(self): + ctl = MotionController(MockMmNamespace()) + events = [] + ctl.subscribe(lambda axis, event: events.append(event)) + try: + ctl.move_abs(0, 40.) + ctl.stop(0) + assert events == [mock.EVENT_STOPPED] + assert not ctl.is_moving(0) + finally: + ctl.close() + + def test_closed_controller_is_refused(self): + ctl = MotionController.mock() + ctl.close() + ctl.close() # closing twice is harmless + with pytest.raises(MotionControllerError, match='closed'): + ctl.position(0) + + def test_close_removes_the_handler(self): + ctl = MotionController.mock() + event = ctl._controller.MoveFinished + assert len(event._handlers) == 1 + ctl.close() + assert event._handlers == [] + + def test_open_controller_without_the_assembly(self, runtime, tmp_path): + with pytest.raises(AssemblyNotFoundError): + open_controller('COM1', str(tmp_path / 'Mm.Net.dll')) + + +@pytest.fixture +def actuator(qapp): + plugin = make_actuator(DAQ_Move_CookbookDotNet) + assert plugin.initialized + yield plugin + plugin.close() + + +class TestPlugins: + def test_ini_returns_the_identity(self, actuator): + assert 'MockMotion .NET' in actuator.init_info + + def test_move_abs_waits_for_the_event(self, actuator): + position = move_abs_and_wait(actuator, 5.) + assert position.value('mm') == pytest.approx(5., abs=0.01) + assert actuator._move_finished + + def test_move_rel(self, actuator): + move_abs_and_wait(actuator, 1.) + assert move_rel_and_wait(actuator, 0.5).value('mm') == pytest.approx(1.5, abs=0.01) + + def test_move_home_ends_with_the_event(self, actuator): + move_abs_and_wait(actuator, 3.) + assert move_home_and_wait(actuator).value('mm') == pytest.approx(0., abs=0.01) + + def test_out_of_range_is_an_error_of_the_driver(self, actuator): + with pytest.raises(MotionControllerError): + actuator.move_abs(DataActuator(data=500., units='mm')) + + def test_axes_share_the_controller(self, actuator): + slave = make_actuator(DAQ_Move_CookbookDotNet, controller=actuator.controller) + try: + slave.settings.child('controller', 'axis').setValue(1) + assert slave.controller is actuator.controller + move_abs_and_wait(actuator, 4.) + move_abs_and_wait(slave, -2.) + assert actuator.get_actuator_value().value('mm') == pytest.approx(4., abs=0.01) + assert slave.get_actuator_value().value('mm') == pytest.approx(-2., abs=0.01) + finally: + slave.close() # a slave never closes the shared controller + assert actuator.get_actuator_value().value('mm') == pytest.approx(4., abs=0.01) + + def test_detector_shares_the_controller(self, actuator): + detector = make_detector(DAQ_0DViewer_CookbookDotNet, controller=actuator.controller) + try: + move_abs_and_wait(actuator, 2.) + dte = grab_and_wait(detector) + assert [float(array[0]) for array in dte[0].data] == pytest.approx([2., 0., 0.], abs=0.01) + assert dte[0].labels == ['X', 'Y', 'Z'] + finally: + detector.close() + + def test_detector_alone_is_a_master(self, qapp): + detector = make_detector(DAQ_0DViewer_CookbookDotNet) + try: + assert grab_and_wait(detector)[0].shape == (1,) + finally: + detector.close()