Code format

#!/usr/bin/env python

from gevent import monkey

monkey.patch_all()

import time

import traceback

from contextlib import contextmanager

@contextmanager

def handle_extension_error(extension_name):

    """'gracefully' log an error if an extension fails to load."""

            f"Exception loading builtin extension picamera_autocalibrate: \n{traceback.format_exc()}"

        )

with handle_extension_error("autofocus"):

    from .autofocus import autofocus_extension_v2

with handle_extension_error("scan"):

    def start(self):

        "Start monitoring sharpness by looking at JPEG size"

        if not self.camera.stream_active:

            logging.warn("Autofocus sharpness monitor was started but the camera isn't streaming.  Attempting to start the stream...")

            logging.warn(

                "Autofocus sharpness monitor was started but the camera isn't streaming.  Attempting to start the stream..."

            )

            self.camera.start_stream_recording()

        self.background_thread = Thread(target=self._measure_jpegs)

        self.background_thread.start()

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

        except ValueError as e:

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

                raise ValueError("No images were captured during the move of the stage.  Perhaps the camera is not streaming images?")

                raise ValueError(

                    "No images were captured during the move of the stage.  Perhaps the camera is not streaming images?"

                )

            else:

                raise e

        if stop < 1:

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

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

        if len(js) == 0:

            raise ValueError("No images were captured during the move of the stage.  Perhaps the camera is not streaming images?")

            raise ValueError(

                "No images were captured during the move of the stage.  Perhaps the camera is not streaming images?"

            )

        return jz[np.argmax(js)]

    def data_dict(self):

def fast_autofocus(microscope, dz=2000, backlash=None):

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

    with monitor_sharpness(microscope) as m, microscope.camera.lock, microscope.stage.lock:

    with monitor_sharpness(

        microscope

    ) as m, microscope.camera.lock, microscope.stage.lock:

        i, z = m.focus_rel(-dz / 2)

        i, z = m.focus_rel(dz)

        fz = m.sharpest_z_on_move(i)

            might slightly hurt accuracy, but is unlikely to be a big issue.  Too big

            may cause you to overshoot, which is a problem.

    """

    with monitor_sharpness(microscope) as m, microscope.camera.lock, microscope.stage.lock:

    with monitor_sharpness(

        microscope

    ) as m, microscope.camera.lock, microscope.stage.lock:

        # Ensure the MJPEG stream has started

        microscope.camera.start_stream_recording()

        if microscope.has_real_stage():

            logging.debug("Running autofocus...")

            task = taskify(fast_up_down_up_autofocus)(microscope, dz=dz, mini_backlash=backlash)

            task = taskify(fast_up_down_up_autofocus)(

                microscope, dz=dz, mini_backlash=backlash

            )

            # return a handle on the autofocus task

            return task

autofocus_extension_v2 = BaseExtension(

    "org.openflexure.autofocus", version="2.0.0",

    description="Actions to move the microscope in Z and pick the point with the sharpest image."

    "org.openflexure.autofocus",

    version="2.0.0",

    description="Actions to move the microscope in Z and pick the point with the sharpest image.",

)

autofocus_extension_v2.add_method(fast_autofocus, "fast_autofocus")

autofocus_extension_v2.add_method(fast_up_down_up_autofocus, "fast_up_down_up_autofocus")

autofocus_extension_v2.add_method(

    fast_up_down_up_autofocus, "fast_up_down_up_autofocus"

)

autofocus_extension_v2.add_method(autofocus, "autofocus")

autofocus_extension_v2.add_view(MeasureSharpnessAPI, "/measure_sharpness")

import numpy as np

class AttributeDict(dict):

    """This class extends a dictionary to have a "create" method for

    compatibility with h5py attrs objects."""

            if isinstance(self[k], np.ndarray):

                self[k] = np.copy(self[k])

def ensure_attribute_dict(obj, copy=False):

    """Given a mapping that may or not be an AttributeDict, return an

    AttributeDict object that either is, or copies the data of, the input."""

            out.copy_arrays()

        return out

def ensure_attrs(obj):

    """Return an ArrayWithAttrs version of an array-like object, may be the

    original object if it already has attrs."""

    if hasattr(obj, 'attrs'):

    if hasattr(obj, "attrs"):

        return obj  # if it has attrs, do nothing

    else:

        return ArrayWithAttrs(obj)  # otherwise, wrap it

class ArrayWithAttrs(np.ndarray):

    """A numpy ndarray, with an AttributeDict accessible as array.attrs.

    def __array_finalize__(self, obj):

        # this is called by numpy when the object is created (__new__ may or

        # may not get called)

        if obj is None: return # if obj is None, __new__ was called - do nothing

        if obj is None:

            return  # if obj is None, __new__ was called - do nothing

        # if we didn't create the object with __new__,  we must add the attrs

        # dictionary.  We copy this from the source object if possible (while

        # ensuring it's the right type) or create a new, empty one if not.

        # NB we don't use ensure_attribute_dict because we want to make sure the

        # dict object is *copied* not merely referenced.

        self.attrs = ensure_attribute_dict(getattr(obj, 'attrs', {}), copy=True)

        self.attrs = ensure_attribute_dict(getattr(obj, "attrs", {}), copy=True)

def attribute_bundler(attrs):

    """Return a function that bundles the supplied attributes with an array."""

    def bundle_attrs(array):

        return ArrayWithAttrs(array, attrs=attrs)

from .camera_stage_tracker import Tracker, move_until_motion_detected

import logging

def displacements(positions):

    """Calculate the absolute distance of each point from the first point."""

    return norm(positions - positions[0, :][np.newaxis, :], axis=1)

def direction_from_points(points):

    """Given an Nx2 array of points, figure out the principal component.

    eigenvalues, eigenvectors = np.linalg.eig(np.cov(points.T))

    return eigenvectors[:, np.argmax(eigenvalues)]

def apply_backlash(x, backlash=0, start_unwound=True):

    """Apply a basic model of backlash to a set of coordinates.

            y[i] = y[i - 1]

    return y

def fit_backlash(moves):

    """Given a set of linear moves forwards and back, estimate backlash.

    assert tracker.stage_positions.shape[0] == 1

    original_stage_pos = tracker.stage_positions[-1, :]

    direction = direction / np.sum(direction**2)**0.5 # ensure "direction" is normalised

    direction = (

        direction / np.sum(direction ** 2) ** 0.5

    )  # ensure "direction" is normalised

    logging.info("Moving the stage until we see motion...")

    # Move the stage until we can see a significant amount of motion

    i, m = move_until_motion_detected(

        tracker, move, direction, threshold=tracker.max_safe_displacement * 0.2)

        tracker, move, direction, threshold=tracker.max_safe_displacement * 0.2

    )

    logging.info("Moving the stage to the edge of the field of view...")

    i, m = move_until_motion_detected(

        tracker, move, direction, 

        tracker,

        move,

        direction,

        threshold=tracker.max_safe_displacement * 0.7,

        multipliers=m / 2.0 * np.arange(20),

        detect_cumulative_motion=True)

        detect_cumulative_motion=True,

    )

    exponential_moves = tracker.history

    # Include this final step, and make a rough estimate of the scaling from stage to image

        move(starting_stage_pos - sensible_step * (i + 1))

        # print(".", end="")

        stage_pos, image_pos = tracker.append_point()

        if (i > 3 and tracker.moving_away_from_centre 

            and norm(image_pos) > 0.65 * tracker.max_safe_displacement):

        if (

            i > 3

            and tracker.moving_away_from_centre

            and norm(image_pos) > 0.65 * tracker.max_safe_displacement

        ):

            break  # Stop once we have moved far enough

    logging.info("Moving the stage forwards to measure backlash (2/2)")

        move(starting_stage_pos + sensible_step * (i + 1))

        # print(".", end="")

        stage_pos, image_pos = tracker.append_point()

        if (i > 3 and tracker.moving_away_from_centre 

            and norm(image_pos) > 0.65 * tracker.max_safe_displacement):

        if (

            i > 3

            and tracker.moving_away_from_centre

            and norm(image_pos) > 0.65 * tracker.max_safe_displacement

        ):

            break  # Stop once we have moved far enough

    linear_moves = tracker.history

    try:

        res = fit_backlash(linear_moves)

        backlash_correction = sensible_step / norm(sensible_step) * res["backlash"] * 1.5

        backlash_correction = (

            sensible_step / norm(sensible_step) * res["backlash"] * 1.5

        )

        # Finally, move back to the starting position, doing backlash-corrected moves.

        logging.info("Moving back to the start, correcting for backlash...")

        backlash_corrected_moves = tracker.history

        move(original_stage_pos - backlash_correction)

    except ValueError:

        return {"exponential_moves": exponential_moves, "linear_moves": linear_moves,}

        return {"exponential_moves": exponential_moves, "linear_moves": linear_moves}

    finally:

        # Reset position

        move(original_stage_pos)

    logging.info(f"Estimated backlash {res['backlash']:.0f} steps")

    logging.info(f"Stage-to-image ratio {np.abs(res['pixels_per_step']):.3f} pixels/step")

    logging.info(f"Residuals were about {res['fractional_error']:.2f} times the step size")

    res.update({

    logging.info(

        f"Stage-to-image ratio {np.abs(res['pixels_per_step']):.3f} pixels/step"

    )

    logging.info(

        f"Residuals were about {res['fractional_error']:.2f} times the step size"

    )

    res.update(

        {

            "exponential_moves": exponential_moves,

            "linear_moves": linear_moves,

        "backlash_corrected_moves": backlash_corrected_moves

    })

            "backlash_corrected_moves": backlash_corrected_moves,

        }

    )

    return res

def plot_1d_backlash_calibration(results):

    """Plot the results of a calibration run"""

    from matplotlib import pyplot as plt

    f, ax = plt.subplots(1, 2)

    for k in ["exponential", "linear", "backlash_corrected"]:

        moves = results[k + "_moves"]

        if moves is not None:

            ax[0].plot(moves[1][:,0], moves[1][:,1], 'o-')

            ax[0].plot(moves[1][:, 0], moves[1][:, 1], "o-")

    ax[0].set_aspect(1, adjustable="datalim")

    image_direction = results["image_direction"]

        image_1d = np.sum(image_pos * image_direction[np.newaxis, :], axis=1)

        return stage_1d, image_1d

    ax[1].plot(*convert_moves(results["exponential_moves"]), 'o-')

    ax[1].plot(*convert_moves(results["exponential_moves"]), "o-")

    stage_pos, image_pos = convert_moves(results["linear_moves"])

    model = apply_backlash(stage_pos, results["backlash"])

    model *= results["pixels_per_step"]

    model += np.mean(image_pos) - np.mean(model)

    ax[1].plot(stage_pos, model, '-')

    ax[1].plot(stage_pos, image_pos, 'o')

    ax[1].plot(stage_pos, model, "-")

    ax[1].plot(stage_pos, image_pos, "o")

    if results["backlash_corrected_moves"] is not None:

        ax[1].plot(*convert_moves(results["backlash_corrected_moves"]), '+')

        ax[1].plot(*convert_moves(results["backlash_corrected_moves"]), "+")

    return f, ax

def image_to_stage_displacement_from_1d(calibrations):

    """Combine X and Y calibrations

        c_blash = np.abs(cal["backlash"] * cal["stage_direction"])

        backlash[backlash < c_blash] = c_blash[backlash < c_blash]

    A, res, rank, s = np.linalg.lstsq(image_vectors, stage_vectors) # we solve image*A = stage

    A, res, rank, s = np.linalg.lstsq(

        image_vectors, stage_vectors

    )  # we solve image*A = stage

    return {

        "image_to_stage_displacement": A,

        "backlash_vector": backlash,