Module DynAIkonTrap.filtering.motion_queue

This module provides access to a "motion queue", which is simply a queue for sequences of consecutive motion. The intended usage is to place frames of motion, as determined by a motion filter, into the queue. When a frame of no motion arrives, this is not added to the queue and the motion sequence is ended.

The sequence is then analysed by the animal filter, loaded into the MotionQueue, and a callback called with only the animal frames from the motion sequence. Within a MotionSequence there is some simplistic "smoothing" of animal detections. This means even animal detectors that provide sporadic outputs in time, are transformed to a smooth system output.

Below is a simple outline example of how this can be used to print all animal frames:

camera = Camera()

mf = MotionFilter(...)

mq = MotionQueue(
    AnimalFilter(...), 
    print, 
    MotionQueueSettings(), 
    camera.framerate,
    )

while True:

    frame = camera.get() # Can raise Empty exception

    motion_score = mf.run_raw(frame.motion)
    motion_detected = motion_score >= motion_threshold

    if motion_detected:
        mq.put(frame, motion_score)
    else:
        # Safe to call repeatedly; will only end non-empty motion sequence
        mq.end_motion_sequence()

The modularity here means Different implementations for animal filtering and motion filtering stages can be used.

Classes

class Label (value, names=None, *, module=None, qualname=None, type=None, start=1)

Categories into which a frame can fall

Ancestors

• enum.Enum

Class variables

var ANIMAL

var EMPTY

var UNKNOWN

class LabelledFrame (frame: Frame, index: int, priority: float, label: Label = Label.UNKNOWN)

A frame of motion and image data accompanied by some additional labels for the motion queue

Class variables

var frame : Frame

var index : int

var label : Label

var priority : float

class MotionQueue (settings: MotionQueueSettings, animal_detector: AnimalFilter, framerate: int)

A queue for sequences of motion to be analysed by the animal filter

Args

settings : MotionQueueSettings

Settings for the queue

animal_detector : AnimalFilter

An initialised animal filter to apply to frames in the motion sequences

output_callback : Callable[[List[Frame]], Any]

Function to call with filtered frames

framerate : int

Framerate at which the frames were recorded

Methods

def close(self)

def end_motion_sequence(self)

End the current motion sequence and prepare the next one. To be called when there is a gap in motion. It is safe to call this repeatedly for consecutive empty frames. Calling this releases the motion sequence to be processed by the animal filter.

def get(self) ‑> Frame

Retrieve the next animal Frame from the motion queue's output

Returns

Frame

An animal frame

def is_idle(self) ‑> bool

Allows checking if the motion queue is currently waiting for new frames to arrive. May be removed in future.

def put(self, frame: Frame, motion_score: float)

Append the given frame to the current motion sequence. If the sequence exceeds the length limit, a new one is automatically started. This prevents excessively long motion sequences.

Args

frame : Frame

A frame of motion and image data to be analysed

motion_score : float

Output value for this frame from the motion filtering stage

class MotionSequence (smoothing_len: int)

Sequence of consecutive frames with motion deemed sufficient by a previous motion filtering stage. Smoothing is built in to smooth any animal detections over multiple frames. This can be done as the minimum number of frames in which an animal is likely to be present, can be reasoned about.

Args

smoothing_len : int

Number of frames by which to smooth animal detections in either direction

Methods

def close_gaps(self)

Remove small gaps of missing animal predictions in the current motion sequence. This should only be called just before the motion sequence is passed out of the motion queue. This function removes unlikely gaps in animal detections using the smoothing_len.

def get_animal_frames(self) ‑> List[LabelledFrame]

Retrieve only the animal frames from the motion sequence

Returns

List[LabelledFrame]

List of animal frames from this motion sequence

def get_highest_priority(self) ‑> LabelledFrame

Finds the frame with the highest priority in the motion sequence. This should be the next frame to be passed to the animal filtering stage.

Returns

LabelledFrame

Frame to be analysed by the animal filtering stage

def label_as_animal(self, frame: LabelledFrame)

Label a given frame as containing an animal. Intended to be called based on the output of the animal filter. Frames either side of this one in the current motion sequence will also be labelled as animal according to the smoothing_len

Args

frame : LabelledFrame

The frame to be labelled as containing an animal

def label_as_empty(self, frame: LabelledFrame)

Label the given frame as empty. Intended to be called based on the output of the animal filter. Only this frame is labelled as empty; no smoothing is applied.

Args

frame : LabelledFrame

The frame to be labelled as being empty

def put(self, frame: Frame, motion_score: float)

Append the frame to this motion sequence

Args

frame : Frame

Frame to be put in this motion sequence

motion_score : float

Output value for this frame from the motion filtering stage