2.9 dev numpy types

# Developer guidelines

The Raspberry Pi image we use currently ships with Python 3.7.3. For local development, please use PyEnv or similar to make sure you're running on this version. For example, Windows users can use [Scoop](https://scoop.sh/) to install specific Python versions.

## Installation

* `git clone https://gitlab.com/openflexure/openflexure-microscope-server.git`

* `cd openflexure-microscope-server`

* (Optional) Set local Python version

  * `pyenv init`

  * `pyenv install 3.7.3`

  * `pyenv local 3.7.3`

* `poetry install`

  * Building the static interface will require a valid Node.js installation

  * To build on a Raspberry Pi:

import logging

import time

from contextlib import contextmanager

from typing import Callable, List, Optional, Tuple

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

import numpy as np

from labthings import current_action, fields, find_component

        self.camera.stream.stop_tracking()

        self.camera.stream.reset_tracking()

    def focus_rel(self, dz: int, backlash: bool = False, **kwargs):

    def focus_rel(self, dz: int, backlash: bool = False, **kwargs) -> Tuple[int, int]:

        # Store the start time and position

        self.camera.stream.start_tracking()

        self.stage_times.append(time.time())

        self.camera.stream.reset_tracking()

        # Index of the data for this movement

        data_index = len(self.stage_positions) - 2

        data_index: int = len(self.stage_positions) - 2

        # Final z position after move

        final_z_position = self.stage_positions[-1][2]

        final_z_position: int = self.stage_positions[-1][2]

        return data_index, final_z_position

    def move_data(self, istart: int, istop: Optional[int] = None):

    def move_data(

        self, istart: int, istop: Optional[int] = None

    ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:

        """Extract sharpness as a function of (interpolated) z"""

        if istop is None:

            istop = istart + 2

        jpeg_times = np.array(self.jpeg_times)

        jpeg_sizes = np.array(self.jpeg_sizes)

        stage_times = np.array(self.stage_times)[istart:istop]

        stage_zs = np.array(self.stage_positions)[istart:istop, 2]

        jpeg_times: np.ndarray = np.array(self.jpeg_times)  # np.ndarray[float]

        jpeg_sizes: np.ndarray = np.array(self.jpeg_sizes)  # np.ndarray[int]

        stage_times: np.ndarray = np.array(self.stage_times)[

            istart:istop

        ]  # np.ndarray[float]

        stage_zs: np.ndarray = np.array(self.stage_positions)[

            istart:istop, 2

        ]  # np.ndarray[int]

        try:

            start = np.argmax(jpeg_times > stage_times[0])

            stop = np.argmax(jpeg_times > stage_times[1])

            start: int = int(np.argmax(jpeg_times > stage_times[0]))

            stop: int = int(np.argmax(jpeg_times > stage_times[1]))

        except ValueError as e:

            if np.sum(jpeg_times > stage_times[0]) == 0:

                raise ValueError(

            stop = len(jpeg_times)

            logging.debug("changing stop to %s", (stop))

        jpeg_times = jpeg_times[start:stop]

        jpeg_zs = np.interp(jpeg_times, stage_times, stage_zs)

        jpeg_zs: np.ndarray = np.interp(

            jpeg_times, stage_times, stage_zs

        )  # np.ndarray[float]

        return jpeg_times, jpeg_zs, jpeg_sizes[start:stop]

    def sharpest_z_on_move(self, index: int):

    def sharpest_z_on_move(self, index: int) -> int:

        """Return the z position of the sharpest image on a given move"""

        _, jz, js = self.move_data(index)

        if len(js) == 0:

            )

        return jz[np.argmax(js)]

    def data_dict(self):

    def data_dict(self) -> Dict[str, np.ndarray]:

        """Return the gathered data as a single convenient dictionary"""

        data = {}

        for k in ["jpeg_times", "jpeg_sizes", "stage_times", "stage_positions"]:

@contextmanager

def monitor_sharpness(microscope: Microscope):

    m = JPEGSharpnessMonitor(microscope)

    m: JPEGSharpnessMonitor = JPEGSharpnessMonitor(microscope)

    m.start()

    try:

        yield m

        m.stop()

def sharpness_sum_lap2(rgb_image: np.ndarray):

def sharpness_sum_lap2(rgb_image: np.ndarray) -> np.float:

    """Return an image sharpness metric: sum(laplacian(image)**")"""

    image_bw = np.mean(rgb_image, 2)

    image_lap = ndimage.filters.laplace(image_bw)

    image_bw: np.float = np.mean(rgb_image, 2)

    image_lap: np.float = ndimage.filters.laplace(image_bw)

    return np.mean(image_lap.astype(np.float) ** 4)

def sharpness_edge(image: np.ndarray):

def sharpness_edge(image: np.ndarray) -> np.float:

    """Return a sharpness metric optimised for vertical lines"""

    gray = np.mean(image.astype(float), 2)

    n = 20

    edge = np.array([[-1] * n + [1] * n])

    gray: np.float = np.mean(image.astype(float), 2)

    n: int = 20

    edge: np.ndarray = np.array([[-1] * n + [1] * n])

    return np.sum(

        [np.sum(ndimage.filters.convolve(gray, W) ** 2) for W in [edge, edge.T]]

    )

    def measure_sharpness(

        self, microscope: Microscope, metric_fn: Callable = sharpness_sum_lap2

    ):

    ) -> np.float:

        """Measure the sharpness of the camera's current view."""

        if hasattr(microscope.camera, "array") and callable(

        dz: List[int],

        settle: float = 0.5,

        metric_fn: Callable = sharpness_sum_lap2,

    ):

    ) -> Tuple[List[int], List[np.float]]:

        """Perform a simple autofocus routine.

        The stage is moved to z positions (relative to current position) in dz,

        and at each position an image is captured and the sharpness function 

        highest.  No interpolation is performed.

        dz is assumed to be in ascending order (starting at -ve values)

        """

        camera = microscope.camera

        stage = microscope.stage

        camera: BaseCamera = microscope.camera

        stage: BaseStage = microscope.stage

        with set_properties(stage, backlash=256), stage.lock, camera.lock:

            sharpnesses = []

            positions = []

            sharpnesses: List[np.float] = []

            positions: List[int] = []

            # Some cameras may not have annotate_text. Reset if it does

            if getattr(camera, "annotate_text"):

            for _ in stage.scan_z(dz, return_to_start=False):

                if current_action() and current_action().stopped:

                    return

                    return [], []

                positions.append(stage.position[2])

                time.sleep(settle)

                sharpnesses.append(self.measure_sharpness(microscope, metric_fn))

            newposition = positions[np.argmax(sharpnesses)]

            newposition: int = positions[int(np.argmax(sharpnesses))]

            stage.move_rel((0, 0, newposition - stage.position[2]))

        return positions, sharpnesses

    def move_and_find_focus(self, microscope: Microscope, dz: int):

    def move_and_find_focus(self, microscope: Microscope, dz: int) -> int:

        """Make a relative Z move and return the peak sharpness position"""

        with monitor_sharpness(microscope) as m:

            m.focus_rel(dz)

            return m.sharpest_z_on_move(0)

    def move_and_measure(self, microscope: Microscope, dz: int):

    def move_and_measure(

        self, microscope: Microscope, dz: int

    ) -> Dict[str, np.ndarray]:

        """Make a relative Z move and return the sharpness data"""

        with monitor_sharpness(microscope) as m:

            m.focus_rel(dz)

            return m.data_dict()

    def fast_autofocus(self, microscope: Microscope, dz: int = 2000):

    def fast_autofocus(

        self, microscope: Microscope, dz: int = 2000

    ) -> Dict[str, np.ndarray]:

        """Perform a down-up-down-up autofocus"""

        with microscope.camera.lock, microscope.stage.lock:

            with monitor_sharpness(microscope) as m:

                # z: Final z position after move

                i, z = m.focus_rel(dz)

                # Get the z position with highest sharpness from the previous move (index i)

                fz = m.sharpest_z_on_move(i)

                fz: int = m.sharpest_z_on_move(i)

                # Move all the way to the start so it's consistent

                # Store final absolute z position from this return move

                i, z = m.focus_rel(-dz)

        target_z: int = 0,

        initial_move_up: bool = True,

        mini_backlash: int = 25,

    ):

    ) -> Dict[str, np.ndarray]:

        """Autofocus by measuring on the way down, and moving back up with feedback.

        This autofocus method is very efficient, as it only passes the peak once.

                    m.focus_rel(dz / 2)

                # move down

                logging.debug("Move down")

                i: int

                z: int

                i, z = m.focus_rel(-dz)

                # now inspect where the sharpest point is, and estimate the sharpness

                # (JPEG size) that we should find at the start of the Z stack

                jz: np.ndarray  # np.ndarray[float]

                js: np.ndarray  # np.ndarray[float]

                _, jz, js = m.move_data(i)

                best_z = jz[np.argmax(js)]

                best_z: int = jz[np.argmax(js)]

                # now move to the start of the z stack

                logging.debug("Move to the start of the z stack")

                # We've deliberately undershot - figure out how much further we should move based on the curve

                logging.debug("Calculate remining movement")

                current_js = m.camera.stream.last.size

                imax = np.argmax(js)  # we want to crop out just the bit below the peak

                imax: int = int(

                    np.argmax(js)

                )  # we want to crop out just the bit below the peak

                js = js[imax:]  # NB z is in DECREASING order

                jz = jz[imax:]

                inow = np.argmax(

                    js < current_js

                inow: int = int(

                    np.argmax(js < current_js)

                )  # use the curve we recorded to estimate our position

                # So, the Z position corresponding to our current sharpness value is zs[inow]

                # That means we should move forwards, by best_z - zs[inow]

                logging.debug("Correction move")

                correction_move = best_z + target_z - jz[inow]

                correction_move: int = best_z + target_z - jz[inow]

                logging.debug(

                    "Fast autofocus scan: correcting backlash by moving %s steps",

                    (correction_move),

def channels_from_bayer_array(bayer_array: np.ndarray) -> np.ndarray:

    """Given the 'array' from a PiBayerArray, return the 4 channels."""

    bayer_pattern: List[Tuple[int, int]] = [(0, 0), (0, 1), (1, 0), (1, 1)]

    channels: np.ndarray = np.zeros(

        (4, bayer_array.shape[0] // 2, bayer_array.shape[1] // 2),

        dtype=bayer_array.dtype,

    channels_shape: Tuple[int, ...] = (

        4,

        bayer_array.shape[0] // 2,

        bayer_array.shape[1] // 2,

    )

    channels: np.ndarray = np.zeros(channels_shape, dtype=bayer_array.dtype)

    for i, offset in enumerate(bayer_pattern):

        # We simplify life by dealing with only one channel at a time.

        channels[i, :, :] = np.sum(

from .tools import devtools_extension_v2

from .tools import DevToolsExtension

LABTHINGS_EXTENSIONS = [devtools_extension_v2]

LABTHINGS_EXTENSIONS = [DevToolsExtension]

from labthings.views import ActionView

class DevToolsExtension(BaseExtension):

    def __init__(self) -> None:

        super().__init__(

            "org.openflexure.dev.tools",

            version="0.1.0",

            description="Actions to cause various traumatic events in the microscope, used for testing.",

        )

        self.add_view(RaiseException, "/raise")

        self.add_view(SleepFor, "/sleep")

class RaiseException(ActionView):

    def post(self):

        raise Exception("The developer raised an exception")

        end = time.time()

        logging.info("Waking up!")

        return {"TimeAsleep": (end - start)}

devtools_extension_v2 = BaseExtension(

    "org.openflexure.dev.tools",

    version="0.1.0",

    description="Actions to cause various traumatic events in the microscope, used for testing.",

)

devtools_extension_v2.add_view(RaiseException, "/raise")

devtools_extension_v2.add_view(SleepFor, "/sleep")