media package

Subpackages

• media.controllers package
  • Submodules
  • media.controllers.brightnesscontroller module
    • BrightnessController
      • BrightnessController.add_rules()
      • BrightnessController.brightness()
      • BrightnessController.get_brightness()
      • BrightnessController.put_brightness()
  • media.controllers.configurationcontroller module
    • ConfigurationController
      • ConfigurationController.add_rules()
      • ConfigurationController.configuration()
      • ConfigurationController.get_configuration()
      • ConfigurationController.put_configuration()
  • media.controllers.controller module
    • Controller
      • Controller.add_rules()
      • Controller.response()
  • Module contents

Submodules

media.headpose module

Head Pose Class

class media.headpose.HeadPose

Bases: object

A class to estimate and track head pose using Kalman Filters and a pre-trained model.

calculate_direction(point: Tuple[float, float], ref: Tuple[float, float], angle: float, distance: float = 100.0) → Tuple[float, float]

Calculate a new point B that moves from ‘point’ in the given ‘angle’ clockwise relative to the ‘ref’ point.

Parameters:

• point (Tuple[float, float]) – (x1, y1) starting point.

• ref (Tuple[float, float]) – (xr, yr) reference point.

• angle (float) – Angle in degrees, measured clockwise from the vector (ref -> point).

• distance (float) – Distance to move in the given direction.

Returns:

New point (x2, y2).

Return type:

Tuple[float, float]

get_head_pose(simcc: Tuple, resolution: Tuple[int, int], landmarks_length: int, num_landmarks: int, face: dict, filter_output: bool = True) → Tuple[float | None, float]

Get the head pose (yaw) from the given SimCC landmarks and face bounding box.

Parameters:

• simcc (Tuple) – The SimCC x and y coordinates.

• resolution (Tuple[int, int]) – The resolution of the input image.

• landmarks_length (int) – The length of the landmarks.

• num_landmarks (int) – The number of landmarks.

• face (dict) – The face bounding box and ID.

• filter_output (bool) – Whether to filter the output using a Kalman filter.

Returns:

The estimated yaw of the head pose and the average confidence.

Return type:

Tuple[Optional[float], float]

class media.headpose.Kalman1DFilter(process_variance: float = 1e-05, measurement_variance: float = 0.01)

Bases: object

A simple 1D Kalman Filter.

update(measurement: float) → float

Update the filter with a new measurement and return the filtered value.

Parameters:

measurement (float) – The new measurement to update the filter with.

Returns:

The filtered value.

Return type:

float

class media.headpose.Kalman1DTracker(max_missed_frames: int = 10)

Bases: object

A tracker that uses multiple 1D Kalman Filters to track multiple objects.

cleanup(frame_number: int | None = None) → None

Remove stale Kalman filters if an object has not been seen for too long.

Parameters:

frame_number (int, optional) – The current frame number. If None, the internal counter will be used.

get_prediction(object_id: int) → float | None

Get the predicted value for an object if it exists.

Parameters:

object_id (int) – The ID of the object to get the prediction for.

Returns:

The predicted value of the object, or None if the object does not exist.

Return type:

Optional[float]

update(object_id: int, measurement: float, frame_number: int | None = None) → float

Update or create a Kalman filter for the given object ID.

Parameters:

• object_id (int) – The ID of the object to update.

• measurement (float) – The new measurement for the object.

• frame_number (int, optional) – The current frame number. If None, it will be incremented internally.

Returns:

The filtered position of the object.

Return type:

float

class media.headpose.SimCCLandmark

Bases: Structure

A structure to hold the confidence and coordinates of a landmark.

confidence

Structure/Union member

x

Structure/Union member

y

Structure/Union member

media.logger module

Service Logger

class media.logger.CustomFormatter(format_str)

Bases: Formatter

Custom formatter for logging

format(record)

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

class media.logger.Logger

Bases: object

This class contains the methods used to configure the service logger.

LOGGER_NAME = 'media'

classmethod get_logger()

Get logger with the given name

Returns:

A logger object

Return type:

logging

classmethod init(log_level=20, log_file: str = None)

initialize the service logger

Parameters:

• log_level (optional) – Logging level as in logging. Defaults to logging.INFO.

• log_file (str, optional) – Log file path. Defaults to None.

media.main module

Server entry point

media.main.main()

Main application

media.main.parse_args()

Parse arguments

media.mediamanager module

Media Manager Class

class media.mediamanager.MediaManager(config_file: str, bev_calibration_file: str, redis_host: str = 'localhost', redis_port: int = 6379)

Bases: object

Media Manager class manage media stream and configuration.

bev_probe(pad, info)

Probe callback to draw detection circles

get_brightness()

Get the brightness configuration

Returns:

An instance of Brightness if

Return type:

Brightness or None

configuration is available, else None.

get_configuration()

Get the current configuration

Returns:

The current configuration. None if no configuration file was used or configured through the request.

Return type:

Configuration or None

get_pipeline_description()

Parse configuration and defines pipeline description

Returns:

A string with the pipeline description if the configuration can be parsed, None otherwise.

landmarks_probe(pad, info)

Probe callback to detect face landmarks

on_reconnect_callback()

Callback to reconnect bev and landmarks pad probes

set_brightness(config: rrmsutils.models.media.brightness.Brightness)

Change the output stream brightness.

Parameters:

config (Brightness)

Returns:

True if the configuration is successfully applied, False if the value could not be configured.

Return type:

True or False

set_configuration(config: rrmsutils.models.media.configuration.Configuration)

Sets Media configuration: defines inputs to use and output setup.

Parameters:

config (Configuration) – The configuration to apply.

Returns:

True if the configuration is applied correctly, False otherwise.

start()

Configures and starts media stream.

media.mediamanager.glist_iterator(list_: Any, class_: Any) → Any

Meta GList iterator helper

media.server module

Base HTTP server using Flask.

class media.server.Server(controllers: list, host: str = '127.0.0.1', port: int = 5051)

Bases: object

Flask server

start()

Run the server with given port.

media.streamhandler module

Stream Handler Class

class media.streamhandler.StreamHandler(description: str, retry: bool = True, retry_delay: int = 5, on_reconnect_callback=None)

Bases: object

Stream Handler class, sets a Gstreamer pipeline, change its status, updates and gets the properties of the element specified

attach_probe(element_name, callback)

Attach callback to element probe at src pad

Parameters:

• element_name (str) – Pipeline element to attach callback to

• callback – Function callback to call for each buffer.

Returns:

True if can attach callback properly, False otherwise.

Return type:

True or False

get_caps(element_name)

Gets the value of an element caps property in the pipeline

Parameters:

• element_name (str) – Pipeline element to get the caps

• from (property)

Returns:

The value of the element caps property asked for. None if ‘element_name’ doesn’t exist in the pipeline or the property get fails.

Return type:

property (str) or None

get_property(element_name, property_name)

Gets the value of an elements property in the pipeline

Parameters:

• element_name (str) – Pipeline element to get the property from

• property_name (str) – Property in the pipeline element to get the value from

Returns:

property; the value of the elements property asked for,

None if the element doesn’t exist in the pipeline.

Return type:

property or None

set_caps(element_name, caps)

Set the ‘caps’ in the pipeline ‘element_name’

Parameters:

• element_name (str) – Pipeline element to be changed

• caps (str) – Caps to be changed in the caps property of pipeline element

Returns:

True if the caps property is successfully set. False if ‘element_name’ is not present in the pipeline or the property set fails.

Return type:

True or False

set_property(element_name, property_name, value)

Set the ‘property_name’ in the pipeline ‘element_name’ to the specified ‘value’

Parameters:

• element_name (str) – Pipeline element to be changed

• property_name (str) – Property in the pipeline element to be changed

• value (str or float) – Value to be set in the property that belongs to the element in the pipeline

Returns:

True if the pipeline element actually is in the pipeline.

False if doesn’t.

Return type:

True or False

start()

Stars the stream :returns: True is start is successful, False if not :rtype: True or False

stop()

Stops the stream :returns: True is stop is successful, False if not :rtype: True or False

Module contents