camfi.wingbeat module

Implements wingbeat frequency measurement from annotated images of flying insects.

class camfi.wingbeat.BcesEM(*, x: numpy.ndarray, y: numpy.ndarray, n_classes: pydantic.types.PositiveInt, xerr: Union[float, numpy.ndarray] = 0.0, yerr: Union[float, numpy.ndarray] = 0.0, cov: Union[float, numpy.ndarray] = 0.0, class_mask: Union[None, int, numpy.random._generator.Generator, numpy.ndarray] = None, prob_class: numpy.ndarray = None)

Bases: pydantic.main.BaseModel

Implements an expectation-maximisation algorithm for fitting multiple BCES linear regresssions to a dataset.

Parameters

  • x (np.ndarray) – Independent variable values.

  • y (np.ndarray) – Dependent variable values. Should have same shape as x.

  • n_classes (int) – Number of classes (i.e. number of regressions in multiple regression).

  • xerr (Union[float, np.ndarray]) – Error of independent variable measurments. Should have same shape as x. If float is given, it will be converted to an array.

  • yerr (Union[float, np.ndarray]) – Error of dependent variable measurements. Should have same shape as x. If float is given, it will be converted to an array.

  • cov (Union[float, np.ndarray]) – Covariance of independent and dependent variable measurements. Should have same shape as x. If float is given, it will be converted to an array.

  • class_mask (Union[None, int, np.random.Generator, np.ndarray]) – Array of integers defining classes of measurements. Accepts values from the set {0, 1, …, n_classes - 1}. Should have same shape as x. Alternatively, give a seed for a random number generator (or the Generator itself), and class_mask will be generated.

  • prob_class (np.ndarray) – Marginal probabilities of each class. Should have shape (n_classes,), and sum to 1.

fit(max_iterations: int = 100)

Performs expectation-maximisation to fit data to self.n_classes BCES linear regression lines. Modifies self.class_mask and self.prob_class.

Parameters

max_iterations (int) – Maximum number of EM-iterations.

Returns

estimates – list of self.n_classes BcesResult instances.

Return type

list[BcesResult]

fit_bces()

Fits BCES linear regressions to the data, subdivided into self.n_classes classes by self.class_mask.

Returns

  • estimates (list[BcesResult]) – list of self.n_classes BcesResult instances.

  • err (np.ndarray) – Array of weighted errors for each measurement from each regression line. Has shape (n_classes, len(x)).

classmethod from_region_dataframe(regions: pandas.core.frame.DataFrame, n_classes: int, seed: Optional[int] = None) → camfi.wingbeat.BcesEM

Initialises a BcesEM object from a regions dataframe.

Parameters

  • regions (pd.DataFrame) – DataFrame containing wingbeat-extracted polylines. SNR threshold should already have been applied. Must contain columns “best_peak”, “et_up”, “et_dn”, and “blur_length”.

  • n_classes (int) – Number of target classes.

  • seed (Optional[int]) – Sets the seed for initialisation of class_mask.

Returns

bces_em – Model to fit by calling bces_em.fit().

Return type

BcesEM

class camfi.wingbeat.BcesResult(*, gradient: float, y_intercept: float, gradient_stderr: float, y_intercept_stderr: float, cov_xy: float)

Bases: pydantic.main.BaseModel

Stores parameters of one BCES linear regression.

estimateslist[tuple[float, float, float, float, float]]

list of (gradient, y_intercept, gradient_stderr, y_intercept_stderr, cov_xy) tuples of estimates and standard errors. Has length n_classes.

errnp.ndarray

Array of weighted errors for each measurement from each regression line. Has shape (n_classes, len(x)).

Parameters

  • gradient (float) – Gradient Estimate.

  • y_intercept (float) – Intercept estimate.

  • gradient_stderr (float) – Standard error of the gradient esitmate.

  • y_intercept_stderr (float) – Standard error of the y_intercept estimate.

  • cov_xy (float) – Covariance estimate.

__ge__(other, NotImplemented=NotImplemented)

Return a >= b. Computed by @total_ordering from (not a < b).

__gt__(other, NotImplemented=NotImplemented)

Return a > b. Computed by @total_ordering from (not a < b) and (a != b).

__le__(other, NotImplemented=NotImplemented)

Return a <= b. Computed by @total_ordering from (a < b) or (a == b).

__lt__(other: camfi.wingbeat.BcesResult) → bool

Return self<value.

class camfi.wingbeat.WingbeatExtractor(*, device: str = 'cpu', backup_device: str = None, scan_distance: pydantic.types.PositiveInt = 50, max_pixel_period: pydantic.types.PositiveInt = None, force_load_exif_metadata: bool = False, metadata: camfi.datamodel.via.ViaMetadata, root: pathlib.Path, line_rate: pydantic.types.PositiveFloat, location: str = None, datetime_corrector: Callable[[datetime.datetime], datetime.datetime] = None, supplementary_figure_plotter: camfi.wingbeat.WingbeatSuppFigPlotter = None)

Bases: camfi.wingbeat.WingbeatExtractorConfig

Class for measuring wingbeat frequencies of annotated flying insects in an image. A new instance of WingbeatExtractor should be used for each distinct image file.

Parameters

  • metadata (ViaMetadata) – Containing annotations of flying insects, as well as file-level image metadata. If file-level metadata is missing (specifically, exposure_time), this will be read from the image file.

  • root (Path) – Path to root directory containing all image directories.

  • line_rate (PositiveFloat) – Rolling shutter line rate of the camera used to take the photo, in lines per second. This must be measured separately. See https://camfi.readthedocs.io/en/latest/usage/notebooks/camera_calibration.html for a guide on measuring the rolling shutter line rate.

  • device (str) – Some steps can run on the GPU, which can give speedups of over 4x. To enable, set e.g. device=”cuda”.

  • backup_device (Optional[str]) – If a step raises a RuntimeError, it can be re-attempted on an alternative device. It is recommended to set to “cpu” if running on a GPU with limited memory.

  • scan_distance (PositiveInt) – Optional parameter used in WingbeatExtractor.process_blur. This defines the maximum perpendicular distance from the polyline annotation of pixels included in the rotated, cropped, and straightened region of interest images.

  • max_pixel_period (Optional[PositiveInt]) – Optional parameter used in WingbeatExtractor.process_blur. By default, autocorrelation is calculated up to half the length of each motion blur. For speed of execution or to reduce memory footprint, a maximum value can be set.

  • force_load_exif_metadata (bool) – If True, EXIF metadata will be read from the image file, regardless of whether exposure_time is already set in metadata.file_attributes. By default, EXIF metadata will only be read if it is missing from metadata.file_attributes.

  • location (Optional[str]) – Sets location string when loading EXIF metadata, placed in metadata.file_attributes.location. Has no effect if EXIF metadata is not loaded. Recommended to use in conjunction with force_load_exif_metadata.

  • datetime_corrector (Optional[DatetimeCorrector]) – If provided, will be called while loading EXIF metadata to obtain a corrected timestamp, which is placed in metadata.file_attributes.datetime_corrected. Has no effect if EXIF metadata is not loaded. Recommended to use in conjunction with force_load_exif_metadata.

  • supplementary_figure_plotter (Optional[WingbeatSuppFigPlotter]) – If set, will be called to plot supplementary figures during self.process_blur. A custom implementation of WingbeatSuppFigPlotter may be used, or one from camfi.plotting.

__eq__(other)

Return self==value.

__hash__()

Return hash(self).

property exposure_time

Gets exposure time either from self.metadata.file_attributes, or from the EXIF metadata of the image file. Caches output so file is only read once for life of WingbeatExtractor instance.

Returns

expoosure_time – Exposure time of photograph in seconds.

Return type

PositiveFloat

extract_wingbeats() → int

Calls self.process_blur on the shape_attributes of all polyline regions in self.metadata, replacing the region_attributes of those regions with ones containing wingbeat data.

Operates in place.

Returns

polylines_processed – Number of polyline annotations processed.

Return type

int

property image

Loads image from file and converts it to a greyscale tensor. Output is cached (so image is only loaded once for the life of the WingbeatExtractor instance).

Returns

image – Image tensor with shape [height, width].

Return type

torch.Tensor

process_blur(polyline: camfi.datamodel.geometry.PolylineShapeAttributes, score: Optional[float] = None) → camfi.datamodel.via_region_attributes.ViaRegionAttributes

Performs the camfi algorithm to takes a measurement of wingbeat frequency from a flying insect motion blur which has been annotated with a polyline.

Parameters

  • polyline (PolylineShapeAttributes) – Polyline annotation following the path of the flying insect’s motion blur.

  • score (Optional[float]) – Score parameter to be passed to ViaRegionAttributes constructor (should set if processing an annotation which was generated automatically, so that the score is reflected in the output).

Returns

region_attributes – With all fields set (including score iff a value was given).

Return type

ViaRegionAttributes

class camfi.wingbeat.WingbeatExtractorConfig(*, device: str = 'cpu', backup_device: str = None, scan_distance: pydantic.types.PositiveInt = 50, max_pixel_period: pydantic.types.PositiveInt = None, force_load_exif_metadata: bool = False)

Bases: pydantic.main.BaseModel

Contains configurable parameters for WingbeatExtractor. Provides a creation methods for WingbeatExtractor. See WingbeatExtractor for futher documentation.

get_wingbeat_extractor(**kwargs)

Instantiates WingbeatExtractor.

Parameters

**kwargs – Passed to WingbeatExtractor

Returns

wingbeat_extractor – WingbeatExtractor with fields taken from self and kwargs.

Return type

WingbeatExtractor

class camfi.wingbeat.WingbeatSuppFigPlotter(*, root: pathlib.Path, image_filename: pathlib.Path, suffix: str = '.png', annotation_idx: pydantic.types.NonNegativeInt = 0)

Bases: abc.ABC, pydantic.main.BaseModel

Defines an interface for producing suplementary figures from the wingbeat extraction procedure, used by WingbeatExtractor. This abstract base class is agnostic to which plotting library is used. Subclasses will naturally have to use a particular plotting library, such as matplotlib.

Concrete subclasses must implement the .__call__ method.

Parameters

  • root (Path) – Root directory to put supplementary figures in.

  • image_filename (Path) – Relative path to image file. self.get_filepath uses this to compute the path where the supplementary figure will be written (note that if annotation_idx is set, then the filename will be modified to include the annotation index - useful for plotting multple supplementary figures relating to annotations from one image).

  • suffix (str) – File suffix for output plots. Defaults to “.png”, but could be “.pdf”, “.html”, or “.eps”, etc. depending on the subclass of WingbeatSuppFigPlotter.

  • annotation_idx (NonNegativeInt) – Used to infer the correct filename. Defaults to 0 (which is almost always what you would want when initialising a subclass of WingbeatSuppFigPlotter).

Examples

>>> class DummyWingbeatSuppFigPlotter(WingbeatSuppFigPlotter):
...     def __call__(
...         self,
...         region_attributes: ViaRegionAttributes,
...         region_of_interest: torch.Tensor,
...         mean_autocorrelation: torch.Tensor,
...     ) -> None:
...         self.get_filepath()
...         return None
>>> supplementary_figure_plotter = DummyWingbeatSuppFigPlotter(
...     root="foo", image_filename="bar/baz.jpg"
... )
>>> supplementary_figure_plotter.get_filepath() == Path("foo/bar/baz_0.png")
True

The annotation index has now been incremented by 1.

>>> supplementary_figure_plotter.annotation_idx
1
>>> supplementary_figure_plotter.get_filepath() == Path("foo/bar/baz_1.png")
True
abstract __call__(region_attributes: camfi.datamodel.via_region_attributes.ViaRegionAttributes, region_of_interest: torch.Tensor, mean_autocorrelation: torch.Tensor) → None

Implementations produce a supplementary figure of a wingbeat extraction process, perhaps writing this to a file. Should call self.get_filepath() once only to get the path to the file where the figure should be written.

Parameters

  • region_attributes (ViaRegionAttributes) – With fields calculated (e.g. by WingbeatExtractor.process_blur)

  • region_of_interest (torch.Tensor) – Greyscale image Tensor displaying region of interest

  • mean_autocorrelation (torch.Tensor) – 1-d Tensor with values containing autocorrrelation along axis 1 of region_of_interest

get_filepath()

Computes the filepath for the supplementary figure and increments self.annotation_idx by 1.

Returns

filepath – Full path to supplementary figure file.

Return type

Path

camfi.wingbeat.autocorrelation(roi: torch.Tensor, max_pixel_period: pydantic.types.PositiveInt) → torch.Tensor

Calculates the autocorrelation along axis 1 of roi. Will run entirely on the device specified by roi.device, and is optimised for running on the GPU.

Parameters

  • roi (torch.Tensor) – Tensor of shape (width, blur_length).

  • max_pixel_period (Optional[PositiveInt]) – Maximum period to consider (choosing a smaller number will increase the speed of execution if running on cpu, and decrease memory consumption regardless of which device it’s running on). If None, calculated as roi.shape[1] // 2.

Returns

mean_autocorrelation – Tensor of shape (max_pixel_period,). Contains the autocorrelation with of the roi along axis 1, with integer step-sizes.

Return type

torch.Tensor

Examples

>>> from torch import arange, cos, sin, stack
>>> from math import pi
>>> theta = arange(0., 4. * pi, pi / 4)
>>> autocorrelation(stack([sin(theta), cos(theta)]), len(theta) // 2)
tensor([ 0.4375,  0.2500,  0.0000, -0.2500, -0.3750, -0.2500,  0.1250,  0.3750])
camfi.wingbeat.extract_all_wingbeats(via_project: camfi.datamodel.via.ViaProject, disable_progress_bar: Optional[bool] = True, **kwargs) → None

Extracts wingbeat data from all images in via_project, inserting that data into via_project (operates in place).

Parameters

  • via_project (ViaProject) – Contains metadata for each image to process.

  • disable_progress_bar (Optional[bool]) – If True (default), progress bar is disabled. If set to None, disable on non-TTY.

  • **kwargs – Passed to WingbeatExtractor constructor (once for each image).

camfi.wingbeat.find_best_peak(values: torch.Tensor) → tuple

Takes a Tensor of values (with 1 dimension), and finds the index of the best peak If peak finding fails, (None, None) is returned.

Parameters

values (torch.Tensor) – 1-D tensor of values to find peaks in.

Returns

  • best_peak (Optional[PositiveInt]) – Index of best peak.

  • snr (Optional[float]) – Score of peak.

Examples

>>> t = torch.zeros(100)
>>> t[0] = 1.  # The first peak is always ignored
>>> t[50] = 1.
>>> find_best_peak(t)
(50, inf)

>>> from torch import cos, linspace
>>> from math import pi
>>> t = (cos(linspace(0., 8. * pi, 32)) + 1) * linspace(0.5, 0., 32)
>>> t[4] = max(t[3], t[5]) + 0.01  # small local peak is not picked up
>>> t[2] = t[0] + 0.01  # The highest peak is not necessarily the best peak
>>> best_peak, score = find_best_peak(t)
>>> best_peak
8
>>> 0 < score < 10
True

If no peak is found, then (None, None) is returned >>> find_best_peak(torch.zeros(10)) (None, None)