#!/usr/bin/env python3

## Copyright (C) 2020 David Miguel Susano Pinto <[email protected]>

## Copyright (C) 2020 Mick Phillips <[email protected]>

##

## This file is part of Microscope.

##

## Microscope is free software: you can redistribute it and/or modify

## it under the terms of the GNU General Public License as published by

## the Free Software Foundation, either version 3 of the License, or

## (at your option) any later version.

##

## Microscope is distributed in the hope that it will be useful,

## but WITHOUT ANY WARRANTY; without even the implied warranty of

## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

## GNU General Public License for more details.

##

## You should have received a copy of the GNU General Public License

## along with Microscope. If not, see <http://www.gnu.org/licenses/>.

"""Abstract Base Classes for the different device types."""

import abc

import functools

import itertools

import logging

import queue

import threading

import time

from enum import EnumMeta

from threading import Thread

from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple

import numpy as np

import Pyro4

import microscope

_logger = logging.getLogger(__name__)

# Mapping of setting data types descriptors to allowed-value types.

#

# We use a descriptor for the type instead of the actual type because

# there may not be a unique proper type as for example in enum.

DTYPES = {

"int": (tuple,),

"float": (tuple,),

"bool": (type(None),),

"enum": (list, EnumMeta, dict, tuple),

"str": (int,),

"tuple": (type(None),),

}

def _call_if_callable(f):

"""Call callables, or return value of non-callables."""

return f() if callable(f) else f

class _Setting:

"""Create a setting.

Args:

name: the setting's name.

dtype: a data type from `"int"`, `"float"`, `"bool"`,

`"enum"`, or `"str"` (see `DTYPES`).

get_func: a function to get the current value.

set_func: a function to set the value.

values: a description of allowed values dependent on dtype, or

function that returns a description.

readonly: an optional function to indicate if the setting is

readonly. A setting may be readonly temporarily, so this

function will return `True` or `False` to indicate its

current state. If set to no `None` (default), then its

value will be dependent on the value of `set_func`.

A client needs some way of knowing a setting name and data type,

retrieving the current value and, if settable, a way to retrieve

allowable values, and set the value.

Setters and getters accept or return:

* the setting value for int, float, bool and str;

* the setting index into a list, dict or Enum type for enum.

.. todo::

refactor into subclasses to avoid if isinstance .. elif

.. else. Settings classes should be private: devices should

use a factory method rather than instantiate settings

directly; most already use add_setting for this.

"""

def __init__(

self,

name: str,

dtype: str,

get_func: Optional[Callable[[], Any]],

set_func: Optional[Callable[[Any], None]] = None,

values: Any = None,

readonly: Optional[Callable[[], bool]] = None,

) -> None:

self.name = name

if dtype not in DTYPES:

raise ValueError("Unsupported dtype.")

elif not (isinstance(values, DTYPES[dtype]) or callable(values)):

raise TypeError(

"Invalid values type for %s '%s': expected function or %s"

% (dtype, name, DTYPES[dtype])

)

self.dtype = dtype

self._get = get_func

self._values = values

self._last_written = None

if self._get is not None:

self._set = set_func

else:

# Cache last written value for write-only settings.

def w(value):

self._last_written = value

set_func(value)

self._set = w

if readonly is None:

if self._set is None:

self._readonly = lambda: True

else:

self._readonly = lambda: False

else:

if self._set is None:

raise ValueError(

"`readonly` is not `None` but `set_func` is `None`"

)

else:

self._readonly = readonly

def describe(self):

return {

"type": self.dtype,

"values": self.values(),

"readonly": self.readonly(),

"cached": self._last_written is not None,

}

def get(self):

if self._get is not None:

value = self._get()

else:

value = self._last_written

if isinstance(self._values, EnumMeta):

return self._values(value).value

else:

return value

def readonly(self) -> bool:

return self._readonly()

def set(self, value) -> None:

"""Set a setting."""

if self._set is None:

raise NotImplementedError()

# TODO further validation.

if isinstance(self._values, EnumMeta):

value = self._values(value)

self._set(value)

def values(self):

if isinstance(self._values, EnumMeta):

return [(v.value, v.name) for v in self._values]

values = _call_if_callable(self._values)

if values is not None:

if self.dtype == "enum":

if isinstance(values, dict):

return list(values.items())

else:

# self._values is a list or tuple

return list(enumerate(values))

elif self._values is not None:

return values

class FloatingDeviceMixin(metaclass=abc.ABCMeta):

"""A mixin for devices that 'float'.

Some SDKs handling multiple devices do not allow for explicit

selection of a specific device. Instead, when the SDK is

initialised it assigns an index to each device. However, this

index is only unique until the program ends and next time the

program runs the device might be assigned a different index. This

means that it is not possible to request a specific device to the

SDK. Instead, one connects to one of the available devices, then

initialises it, and only then can one check which one we got.

Floating devices are a problem in systems where there are multiple

devices of the same type but we only want to initialise a subset

of them. Make sure that a device really is a floating device

before making use of this class. Avoid it if possible.

This class is a mixin which enforces the implementation of a

`get_id` method, which typically returns the device serial number.

Args:

index: the index of the device on a shared library. This

argument is added by the device_server program.

"""

def __init__(self, index: int, **kwargs) -> None:

super().__init__(**kwargs)

self._index = index

@abc.abstractmethod

def get_id(self) -> str:

"""Return a unique hardware identifier such as a serial number."""

pass

class TriggerTargetMixin(metaclass=abc.ABCMeta):

"""Mixin for a device that may be the target of a hardware trigger.

.. todo::

Need some way to retrieve the supported trigger types and

modes. This is not just two lists, one for types and another

for modes, because some modes can only be used with certain

types and vice-versa.

"""

@property

@abc.abstractmethod

def trigger_mode(self) -> microscope.TriggerMode:

raise NotImplementedError()

@property

@abc.abstractmethod

def trigger_type(self) -> microscope.TriggerType:

raise NotImplementedError()

@abc.abstractmethod

def set_trigger(

self, ttype: microscope.TriggerType, tmode: microscope.TriggerMode

) -> None:

"""Set device for a specific trigger."""

raise NotImplementedError()

@abc.abstractmethod

def _do_trigger(self) -> None:

"""Actual trigger of the device.

Classes implementing this interface should implement this

method instead of :meth:`trigger`.

"""

raise NotImplementedError()

def trigger(self) -> None:

"""Trigger device.

The actual effect is device type dependent. For example, on a

``Camera`` it triggers image acquisition while on a

`DeformableMirror` it applies a queued pattern. See

documentation for the devices implementing this interface for

details.

Raises:

microscope.IncompatibleStateError: if trigger type is not

set to ``TriggerType.SOFTWARE``.

"""

if self.trigger_type is not microscope.TriggerType.SOFTWARE:

raise microscope.IncompatibleStateError(

"trigger type is not software"

)

_logger.debug("trigger by software")

self._do_trigger()

class Device(metaclass=abc.ABCMeta):

"""A base device class. All devices should subclass this class."""

def __init__(self) -> None:

self.enabled = False

self._settings: Dict[str, _Setting] = {}

def __del__(self) -> None:

self.shutdown()

def get_is_enabled(self) -> bool:

return self.enabled

def _do_disable(self):

"""Do any device-specific work on disable.

Subclasses should override this method rather than modify

`disable`.

"""

return True

def disable(self) -> None:

"""Disable the device for a short period for inactivity."""

self._do_disable()

self.enabled = False

def _do_enable(self):

"""Do any device specific work on enable.

Subclasses should override this method, rather than modify

`enable`.

"""

return True

def enable(self) -> None:

"""Enable the device."""

try:

self.enabled = self._do_enable()

except Exception as err:

_logger.debug("Error in _do_enable:", exc_info=err)

@abc.abstractmethod

def _do_shutdown(self) -> None:

"""Private method - actual shutdown of the device.

Users should be calling :meth:`shutdown` and not this method.

Concrete implementations should implement this method instead

of `shutdown`.

"""

raise NotImplementedError()

def initialize(self) -> None:

"""Initialize the device.

If devices have this method (not required, and many don't),

then they should call it as part of the initialisation, i.e.,

they should call it on their `__init__` method.

"""

pass

def shutdown(self) -> None:

"""Shutdown the device.

Disable and disconnect the device. This method should be

called before destructing the device object, to ensure that

the device is actually shutdown.

After `shutdown`, the device object is no longer usable and

calling any other method is undefined behaviour. The only

exception `shutdown` itself which can be called consecutively,

and after the first time will have no effect.

A device object that has been shutdown can't be reinitialised.

Instead of reusing the object, a new one should be created

instead. This means that `shutdown` will leave the device in

a state that it can be reconnected.

.. code-block:: python

device = SomeDevice()

device.shutdown()

# Multiple calls to shutdown are OK

device.shutdown()

device.shutdown()

# After shutdown, everything else is undefined behaviour.

device.enable() # undefined behaviour

device.get_setting("speed") # undefined behaviour

# To reinitialise the device, construct a new instance.

device = SomeDevice()

.. note::

While `__del__` calls `shutdown`, one should not rely on

it. Python does not guarante that `__del__` will be

called when the interpreter exits so if `shutdown` is not

called explicitely, the devices might not be shutdown.

"""

try:

self.disable()

except Exception as e:

_logger.warning("Exception in disable() during shutdown: %s", e)

_logger.info("Shutting down ... ... ...")

self._do_shutdown()

_logger.info("... ... ... ... shut down completed.")

def add_setting(

self,

name,

dtype,

get_func,

set_func,

values,

readonly: Optional[Callable[[], bool]] = None,

) -> None:

"""Add a setting definition.

Args:

name: the setting's name.

dtype: a data type from `"int"`, `"float"`, `"bool"`,

`"enum"`, or `"str"` (see `DTYPES`).

get_func: a function to get the current value.

set_func: a function to set the value.

values: a description of allowed values dependent on

dtype, or function that returns a description.

readonly: an optional function to indicate if the setting

is readonly. A setting may be readonly temporarily,

so this function will return `True` or `False` to

indicate its current state. If set to no `None`

(default), then its value will be dependent on the

value of `set_func`.

A client needs some way of knowing a setting name and data

type, retrieving the current value and, if settable, a way to

retrieve allowable values, and set the value. We store this

info in a dictionary. I considered having a `Setting1 class

with getter, setter, etc., and adding `Setting` instances as

device attributes, but Pyro does not support dot notation to

access the functions we need (e.g. `Device.some_setting.set`),

so I'd have to write access functions, anyway.

"""

if dtype not in DTYPES:

raise ValueError("Unsupported dtype.")

elif not (isinstance(values, DTYPES[dtype]) or callable(values)):

raise TypeError(

"Invalid values type for %s '%s': expected function or %s"

% (dtype, name, DTYPES[dtype])

)

else:

self._settings[name] = _Setting(

name, dtype, get_func, set_func, values, readonly

)

def get_setting(self, name: str):

"""Return the current value of a setting."""

try:

return self._settings[name].get()

except Exception as err:

_logger.error("in get_setting(%s):", name, exc_info=err)

raise

def get_all_settings(self):

"""Return ordered settings as a list of dicts."""

# Fetching some settings may fail depending on device state.

# Report these values as 'None' and continue fetching other settings.

def catch(f):

try:

return f()

except Exception as err:

_logger.error("getting %s: %s", f.__self__.name, err)

return None

return {k: catch(v.get) for k, v in self._settings.items()}

def set_setting(self, name: str, value) -> None:

"""Set a setting."""

try:

self._settings[name].set(value)

except Exception as err:

_logger.error("in set_setting(%s):", name, exc_info=err)

raise

def describe_setting(self, name: str):

"""Return ordered setting descriptions as a list of dicts."""

return self._settings[name].describe()

def describe_settings(self):

"""Return ordered setting descriptions as a list of dicts."""

return [(k, v.describe()) for (k, v) in self._settings.items()]

def update_settings(self, incoming, init: bool = False):

"""Update settings based on dict of settings and values."""

if init:

# Assume nothing about state: set everything.

my_keys = set(self._settings.keys())

their_keys = set(incoming.keys())

update_keys = my_keys & their_keys

if update_keys != my_keys:

missing = ", ".join([k for k in my_keys - their_keys])

msg = (

"update_settings init=True but missing keys: %s." % missing

)

_logger.debug(msg)

raise Exception(msg)

else:

# Only update changed values.

my_keys = set(self._settings.keys())

their_keys = set(incoming.keys())

update_keys = set(

key

for key in my_keys & their_keys

if self.get_setting(key) != incoming[key]

)

results = {}

# Update values.

for key in update_keys:

if key not in my_keys or not self._settings[key].set:

# Setting not recognised or no set function implemented

results[key] = NotImplemented

update_keys.remove(key)

continue

if self._settings[key].readonly():

continue

self._settings[key].set(incoming[key])

# Read back values in second loop.

for key in update_keys:

results[key] = self._settings[key].get()

return results

def keep_acquiring(func):

"""Wrapper to preserve acquiring state of data capture devices."""

def wrapper(self, *args, **kwargs):

if self._acquiring:

self.abort()

result = func(self, *args, **kwargs)

self._do_enable()

else:

result = func(self, *args, **kwargs)

return result

return wrapper

class DataDevice(Device, metaclass=abc.ABCMeta):

"""A data capture device.

This class handles a thread to fetch data from a device and dispatch

it to a client. The client is set using set_client(uri) or (legacy)

receiveClient(uri).

Derived classed should implement:

* :meth:`abort` (required)

* :meth:`_fetch_data` (required)

* :meth:`_process_data` (optional)

Derived classes may override ``__init__``, ``enable`` and

``disable``, but must ensure to call this class's implementations

as indicated in the docstrings.

"""

def __init__(self, buffer_length: int = 0, **kwargs) -> None:

"""Derived.__init__ must call this at some point."""

super().__init__(**kwargs)

# A thread to fetch and dispatch data.

self._fetch_thread = None

# A flag to control the _fetch_thread.

self._fetch_thread_run = False

# A flag to indicate that this class uses a fetch callback.

self._using_callback = False

# Clients to which we send data.

self._clientStack = []

# A set of live clients to avoid repeated dispatch to disconnected client.

self._liveClients = set()

# A thread to dispatch data.

self._dispatch_thread = None

# A buffer for data dispatch.

self._dispatch_buffer = queue.Queue(maxsize=buffer_length)

# A flag to indicate if device is ready to acquire.

self._acquiring = False

# A condition to signal arrival of a new data and unblock grab_next_data

self._new_data_condition = threading.Condition()

def __del__(self):

self.disable()

super().__del__()

# Wrap set_setting to pause and resume acquisition.

set_setting = keep_acquiring(Device.set_setting)

@abc.abstractmethod

def abort(self) -> None:

"""Stop acquisition as soon as possible."""

self._acquiring = False

def enable(self) -> None:

"""Enable the data capture device.

Ensures that a data handling threads are running. Implement

device specific code in `_do_enable`.

"""

_logger.debug("Enabling ...")

# Call device-specific code.

try:

result = self._do_enable()

except Exception as err:

_logger.debug("Error in _do_enable:", exc_info=err)

self.enabled = False

raise err

if not result:

_logger.warning("Failed to enable but no error was raised")

self.enabled = False

else:

self.enabled = True

if self._using_callback:

_logger.debug("Setup with callback, disabling fetch thread")

if self._fetch_thread:

self._fetch_thread_run = False

else:

_logger.debug("Setting up fetch thread")

if not self._fetch_thread or not self._fetch_thread.is_alive():

self._fetch_thread = Thread(target=self._fetch_loop)

self._fetch_thread.daemon = True

self._fetch_thread.start()

if self._dispatch_thread and self._dispatch_thread.is_alive():

_logger.debug("Found live dispatch thread.")

else:

_logger.debug("Setting up dispatch thread")

self._dispatch_thread = Thread(target=self._dispatch_loop)

self._dispatch_thread.daemon = True

self._dispatch_thread.start()

_logger.debug("... enabled.")

def disable(self) -> None:

"""Disable the data capture device.

Implement device-specific code in `_do_disable`.

"""

self.enabled = False

if self._fetch_thread:

if self._fetch_thread.is_alive():

_logger.debug("Found fetch thread alive. Joining.")

self._fetch_thread_run = False

self._fetch_thread.join()

_logger.debug("Fetch thread is dead.")

super().disable()

@abc.abstractmethod

def _fetch_data(self) -> None:

"""Poll for data and return it, with minimal processing.

If the device uses buffering in software, this function should

copy the data from the buffer, release or recycle the buffer,

then return a reference to the copy. Otherwise, if the SDK

returns a data object that will not be written to again, this

function can just return a reference to the object. If no

data is available, return `None`.

"""

raise NotImplementedError()

def _process_data(self, data):

"""Do any data processing and return data."""

return data

def _send_data(self, client, data, timestamp):

"""Dispatch data to the client."""

_logger.debug("sending data to client")

try:

# Cockpit will send a client with receiveData and expects

# two arguments (data and timestamp). But we really want

# to use Python's Queue and instead of just the timestamp

# we should be sending some proper metadata object. We

# don't have that proper metadata class yet so just send

# the image data as a numpy ndarray for now.

if hasattr(client, "put"):

client.put(data)

else:

client.receiveData(data, timestamp)

except (

Pyro4.errors.ConnectionClosedError,

Pyro4.errors.CommunicationError,

):

# Client not listening

_logger.info(

"Removing %s from client stack: disconnected.", client._pyroUri

)

self._clientStack = list(filter(client.__ne__, self._clientStack))

self._liveClients = self._liveClients.difference([client])

def _dispatch_loop(self) -> None:

"""Process data and send results to any client."""

while True:

_logger.debug("Getting data from dispatch buffer")

client, data, timestamp = self._dispatch_buffer.get(block=True)

if client not in self._liveClients:

_logger.debug("Client not in liveClients so ignoring data.")

continue

err = None

if isinstance(data, Exception):

standard_exception = Exception(str(data).encode("ascii"))

try:

self._send_data(client, standard_exception, timestamp)

except Exception as e:

err = e

else:

try:

self._send_data(

client, self._process_data(data), timestamp

)

except Exception as e:

err = e

if err:

# Raising an exception will kill the dispatch loop. We need

# another way to notify the client that there was a problem.

_logger.error("in _dispatch_loop:", exc_info=err)

self._dispatch_buffer.task_done()

def _fetch_loop(self) -> None:

"""Poll source for data and put it into dispatch buffer."""

self._fetch_thread_run = True

while self._fetch_thread_run:

_logger.debug("Fetching data from device.")

try:

data = self._fetch_data()

except Exception as e:

_logger.error("in _fetch_loop:", exc_info=e)

# Raising an exception will kill the fetch loop. We need

# another way to notify the client that there was a problem.

timestamp = time.time()

self._put(e, timestamp)

data = None

if data is not None:

_logger.debug("Fetch data to be put into dispatch buffer.")

# TODO Add support for timestamp from hardware.

timestamp = time.time()

self._put(data, timestamp)

else:

_logger.debug("Fetched no data from device.")

time.sleep(0.001)

@property

def _client(self):

"""A getter for the current client."""

return (self._clientStack or [None])[-1]

@_client.setter

def _client(self, val):

"""Push or pop a client from the _clientStack."""

if val is None:

self._clientStack.pop()

else:

self._clientStack.append(val)

self._liveClients = set(self._clientStack)

def _put(self, data, timestamp) -> None:

"""Put data and timestamp into dispatch buffer with target dispatch client."""

self._dispatch_buffer.put((self._client, data, timestamp))

def set_client(self, new_client) -> None:

"""Set up a connection to our client.

Clients now sit in a stack so that a single device may send

different data to multiple clients in a single experiment.

The usage is currently::

device.set_client(client) # Add client to top of stack

# do stuff, send triggers, receive data

device.set_client(None) # Pop top client off stack.

There is a risk that some other client calls ``None`` before

the current client is finished. Avoiding this will require

rework here to identify the caller and remove only that caller

from the client stack.

"""

if new_client is not None:

if isinstance(new_client, (str, Pyro4.core.URI)):

self._client = Pyro4.Proxy(new_client)

else:

self._client = new_client

else:

self._client = None

# _client uses a setter. Log the result of assignment.

if self._client is None:

_logger.info("Current client is None.")

else:

_logger.info("Current client is %s.", str(self._client))

@keep_acquiring

def update_settings(self, settings, init: bool = False) -> None:

"""Update settings, toggling acquisition if necessary."""

super().update_settings(settings, init)

# noinspection PyPep8Naming

def receiveClient(self, client_uri: str) -> None:

"""A passthrough for compatibility."""

self.set_client(client_uri)

def grab_next_data(self, soft_trigger: bool = True):

"""Returns results from next trigger via a direct call.

Args:

soft_trigger: calls :meth:`trigger` if `True`, waits for

hardware trigger if `False`.

"""

if not self.enabled:

raise microscope.DisabledDeviceError("Camera not enabled.")

self._new_data_condition.acquire()

# Push self onto client stack.

self.set_client(self)

# Wait for data from next trigger.

if soft_trigger:

self.trigger()

self._new_data_condition.wait()

# Pop self from client stack

self.set_client(None)

# Return the data.

return self._new_data

# noinspection PyPep8Naming

def receiveData(self, data, timestamp) -> None:

"""Unblocks grab_next_frame so it can return."""

with self._new_data_condition:

self._new_data = (data, timestamp)

self._new_data_condition.notify()

class Camera(TriggerTargetMixin, DataDevice):

"""Adds functionality to :class:`DataDevice` to support cameras.

Defines the interface for cameras. Applies a transform to

acquired data in the processing step.

"""

def __init__(self, **kwargs) -> None:

super().__init__(**kwargs)

# Transforms to apply to data (fliplr, flipud, rot90)

# Transform to correct for readout order.

self._readout_transform = (False, False, False)

# Transform supplied by client to correct for system geometry.

self._client_transform = (False, False, False)

# Result of combining client and readout transforms

self._transform = (False, False, False)

self.add_setting("roi", "tuple", self.get_roi, self.set_roi, None)

def _process_data(self, data):

"""Apply self._transform to data."""

flips = (self._transform[0], self._transform[1])

rot = self._transform[2]

# Choose appropriate transform based on (flips, rot).

# Do rotation

data = np.rot90(data, rot)

# Flip

data = {

(0, 0): lambda d: d,

(0, 1): np.flipud,

(1, 0): np.fliplr,

(1, 1): lambda d: np.fliplr(np.flipud(d)),

}[flips](data)

return super()._process_data(data)

@property

def shuttering_mode(self) -> microscope.ElectronicShutteringMode:

"""Return the electronic shuttering mode."""

return self._get_shuttering_mode()

@abc.abstractmethod

def _get_shuttering_mode(self) -> microscope.ElectronicShutteringMode:

"""Return the electronic shuttering mode."""

pass

@shuttering_mode.setter

def shuttering_mode(self, mode: microscope.ElectronicShutteringMode):

"""Set the electronic shuttering mode."""

self._set_shuttering_mode(mode)

def _set_shuttering_mode(self, mode: microscope.ElectronicShutteringMode):

"""Set the electronic shuttering mode."""

raise NotImplementedError()

def get_transform(self) -> Tuple[bool, bool, bool]:

"""Return the current transform without readout transform."""

return self._client_transform

def _update_transform(self):

"""Update transform (after setting the client or readout transform)."""

lr, ud, rot = (

self._readout_transform[i] ^ self._client_transform[i]

for i in range(3)

)

if self._readout_transform[2] and self._client_transform[2]:

lr = not lr

ud = not ud

self._transform = (lr, ud, rot)

def set_transform(self, transform: Tuple[bool, bool, bool]) -> None:

"""Set client transform and update resultant transform."""

self._client_transform = transform

self._update_transform()

def _set_readout_transform(self, new_transform):

"""Set readout transform and update resultant transform."""

self._readout_transform = [bool(int(t)) for t in new_transform]

self._update_transform()

@abc.abstractmethod

def set_exposure_time(self, value: float) -> None:

"""Set the exposure time on the device in seconds."""

pass

def get_exposure_time(self) -> float:

"""Return the current exposure time in seconds."""

pass

def get_cycle_time(self) -> float:

"""Return the cycle time in seconds."""

pass

@abc.abstractmethod

def _get_sensor_shape(self) -> Tuple[int, int]:

"""Return a tuple of `(width, height)` indicating shape in pixels."""

pass

def get_sensor_shape(self) -> Tuple[int, int]:

"""Return a tuple of `(width, height)` corrected for transform."""

shape = self._get_sensor_shape()

if self._transform[2]:

# 90 degree rotation

shape = (shape[1], shape[0])

return shape

@abc.abstractmethod

def _get_binning(self) -> microscope.Binning:

"""Return the current binning."""

pass

def get_binning(self) -> microscope.Binning:

"""Return the current binning corrected for transform."""

binning = self._get_binning()

if self._transform[2]:

# 90 degree rotation

binning = microscope.Binning(binning[1], binning[0])

return binning

@abc.abstractmethod

def _set_binning(self, binning: microscope.Binning):

"""Set binning along both axes. Return `True` if successful."""

pass

def set_binning(self, binning: microscope.Binning) -> None:

"""Set binning along both axes. Return `True` if successful."""

h_bin, v_bin = binning

if self._transform[2]:

# 90 degree rotation

binning = microscope.Binning(v_bin, h_bin)

else:

binning = microscope.Binning(h_bin, v_bin)

return self._set_binning(binning)

@abc.abstractmethod

def _get_roi(self) -> microscope.ROI:

"""Return the ROI as it is on hardware."""

raise NotImplementedError()

def get_roi(self) -> microscope.ROI:

"""Return current ROI."""

roi = self._get_roi()

if self._transform[2]:

# 90 degree rotation

roi = microscope.ROI(roi[1], roi[0], roi[3], roi[2])

return roi

@abc.abstractmethod

def _set_roi(self, roi: microscope.ROI):

"""Set the ROI on the hardware. Return `True` if successful."""

return False

def set_roi(self, roi: microscope.ROI) -> None:

"""Set the ROI according to the provided rectangle.

Return True if ROI set correctly, False otherwise.

"""

maxw, maxh = self.get_sensor_shape()

binning = self.get_binning()

left, top, width, height = roi

if not width: # 0 or None

width = maxw // binning.h

if not height: # 0 or None

height = maxh // binning.v

if self._transform[2]:

roi = microscope.ROI(left, top, height, width)

else:

roi = microscope.ROI(left, top, width, height)

return self._set_roi(roi)

class SerialDeviceMixin(metaclass=abc.ABCMeta):