DynAIkonTrap.filtering.motion_queue#

This module provides access to a MotionLabelledQueue, which is simply a queue for labelled consecutive sequences. The intended usage is to place frames of interest, labelled as determined by a motion filter, into the queue. The queue is ended when the maximum length is reached as determined by MotionQueueSettings.

The sequence is then analysed by the animal filter, loaded into the MotionLabelledQueue, and a callback called with only the animal frames from the motion sequence. Within a Sequence 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:

mf = MotionFilter(...)

mq = MotionLabelledQueue(
    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, MotionStatus.MOTION)
    else:
        mq.put(frame, -1.0, MotionStatus.STILL)

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

Classes

`Label`(value)	Categories into which a frame can fall
`LabelledFrame`(frame, index, priority[, ...])	A frame of motion and image data accompanied by some additional labels for the motion queue
`MotionLabelledQueue`(settings, ...)	A queue for sequences of motion to be analysed by the animal filter
`MotionStatus`(value)	Categories for the motion status of a frame
`Sequence`(smoothing_len, context_len)	Sequence of consecutive labelled frames.

class Label(value)#

Categories into which a frame can fall

ANIMAL = 1#

CONTEXT = 3#

EMPTY = 0#

HUMAN = 4#

UNKNOWN = 2#

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

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

frame: Frame#

index: int#

label: Label = 2#

motion_status: MotionStatus = 2#

priority: float#

class MotionLabelledQueue(settings: ProcessingSettings, animal_detector: AnimalFilter, framerate: int)#

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

Parameters:

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

close()#

end_motion_sequence()#: End the current sequence and prepare the next one. It is safe to call this repeatedly for consecutive empty frames. Calling this releases the sequence to be processed by the animal filter.

get() → Frame#

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

Returns:: An animal frame
Return type:: Frame

is_idle() → bool#: Allows checking if the motion queue is currently waiting for new frames to arrive. May be removed in future.

put(frame: Frame, motion_score: float, motion_status: MotionStatus)#

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

Parameters:

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
status (MotionStatus) – status of motion detected in this frame

class MotionStatus(value)#

Categories for the motion status of a frame

MOTION = 1#

STILL = 0#

UNKNOWN = 2#

class Sequence(smoothing_len: int, context_len: int)#

Sequence of consecutive labelled frames. Frames may be “still” or contain motion. 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.

Parameters:: smoothing_len (int) – Number of frames by which to smooth animal detections in either direction

add_context()#: Add context labels to either side of the animal predictions. This should only be called just before the sequence is passed out of the motion labelled queue - ie after close_gaps()

close_gaps()#: Remove small gaps of missing animal predictions in the current sequence. This function removes unlikely gaps in animal detections using the smoothing_len.

get_animal_frames() → List[LabelledFrame]#

Retrieve only the animal frames from the sequence

Returns:: List of animal frames from this sequence
Return type:: List[LabelledFrame]

get_animal_or_context_frames() → List[LabelledFrame]#

Retrieve only the animal or context frames from the sequence

Returns:: List of animal or context frames from this sequence
Return type:: List[LabelledFrane]

get_first_animal_index() → int#

Finds and returns first index in the frame queue labelled as an animal

Returns:: Index (int) of first animal frame in this sequence.

get_highest_priority() → LabelledFrame#

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

Returns:: Frame to be analysed by the animal filtering stage
Return type:: LabelledFrame

get_last_animal_index() → int#

Finds and returns last index in the frame queue labeled as an animal

Returns:: Index (int) of last animal frame in this sequence.

has_motion() → bool#

Check if this sequence has a frame with motion status MOTION

Returns:: True if a frame in this sequence has a status indicating motion, False otherwise
Return type:: Bool

label_as_animal(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 sequence will also be labelled as animal according to the smoothing_len

Parameters:: frame (LabelledFrame) – The frame to be labelled as containing an animal

label_as_empty(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.

Parameters:: frame (LabelledFrame) – The frame to be labelled as being empty

label_as_human(frame: LabelledFrame)#

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

Parameters:: frame – (LabelledFrame): The frame to be labelled as containing a human

put(frame: Frame, motion_score: float, status: MotionStatus)#

Append the frame to this sequence

Parameters:

frame (Frame) – Frame to be put in this sequence
motion_score (float) – Output value for this frame from the motion filtering stage
status (MotionStatus) – status of motion detected in this frame