Core Framework

The core framework provides the fundamental components for building social interaction applications.

Core Components

component_python2.py

This module contains the SICComponent class, which is the base class for all components in the Social Interaction Cloud.

class sic_framework.core.component_python2.SICComponent(ready_event=None, stop_event=None, log_level=10, conf=None, input_channel=None, output_channel=None, req_reply_channel=None, client_id='', redis=None)

Bases: object

Abstract class for Components that provide essential functions for Social Interaction Cloud applications.

Parameters:

• ready_event (threading.Event, optional) – Threading event to signal when the component is ready. If None, creates a new Event.

• stop_event (threading.Event, optional) – Threading event to signal when the component should stop. If None, creates a new Event.

• log_level (int, optional) – The logging verbosity level (e.g., DEBUG, INFO, WARNING, ERROR).

• conf (dict, optional) – Configuration parameters for the component. If None, uses default configuration.

COMPONENT_STARTUP_TIMEOUT = 30

Timeout in seconds for component startup.

This controls how long a SICConnector should wait when requesting a component to start. Increase this value for components that need more time to initialize (e.g., robots that need to stand up or models that need to load to GPU).

__init__(ready_event=None, stop_event=None, log_level=10, conf=None, input_channel=None, output_channel=None, req_reply_channel=None, client_id='', redis=None)

classmethod get_component_name()

Get the display name of this component.

Returns the name of the subclass that implements this class (e.g. “DesktopCameraSensor”)

Returns:

The component’s display name (typically the class name)

Return type:

str

start()

Start the component. This method registers a request handler, signals the component is ready, and logs that the component has started.

Subclasses should call this method from their overridden start() method to get the framework’s default startup behavior.

stop(*args)

Stop the component.

Closes the Redis connection and sets the stop event.

Parameters:

args (tuple) – Additional arguments (not used)

set_config(new=None)

Set the configuration for this component.

Calls _parse_conf() to parse the configuration message.

Parameters:

new (SICConfMessage, optional) – The new configuration. If None, uses the default configuration.

on_request(request)

Define the handler for Component specific requests. Must return a SICMessage as a reply to the request.

Parameters:

request (SICRequest) – The request for this component.

Returns:

The reply

Return type:

SICMessage

on_message(message='')

Define the handler for input messages.

Parameters:

message (SICMessage) – The message to handle.

Returns:

The reply

Return type:

SICMessage

output_message(message)

Send a message on the output channel of this component.

Stores the component name in the message to allow for debugging.

Parameters:

message (SICMessage) – The message to send.

abstractmethod static get_inputs()

Define the input types the component needs as a list.

Must be implemented by the subclass.

Returns:

list of SIC messages

Return type:

List[Type[SICMessage]]

abstractmethod static get_output()

Define the output type of the component.

Must be implemented by the subclass.

Returns:

SIC message

Return type:

Type[SICMessage]

static get_conf()

Define the expected configuration of the component using SICConfMessage.

Returns:

a SICConfMessage or None

Return type:

SICConfMessage

_start()

Wrapper for the user-implemented start method that provides error handling and logging.

This method calls the user’s start() implementation and ensures any exceptions are properly logged before being re-raised to the caller.

_handle_message(message)

Handle incoming messages.

Calls the user-implemented on_message method to process the message.

Parameters:

message (SICMessage) – The message to handle.

Returns:

The reply to the message.

Return type:

SICMessage

_handle_request(request)

Handle control requests such as SICPingRequests and SICStopRequest by calling generic Component methods. Component specific requests are passed to the normal on_request handler.

Parameters:

request (SICRequest) – The request to handle.

Returns:

The reply to the request.

Return type:

SICMessage

_parse_conf(conf)

Parse configuration messages (SICConfMessage).

This method is called by set_config() to parse the configuration message.

Parameters:

conf (SICConfMessage) – Configuration message to parse

_get_timestamp()

Get the current timestamp from the Redis server.

Returns:

The current timestamp

Return type:

float

sensor_python2.py

This module contains the SICSensor class, which is the base class for all sensors in the Social Interaction Cloud.

class sic_framework.core.sensor_python2.SICSensor(ready_event=None, stop_event=None, log_level=10, conf=None, input_channel=None, output_channel=None, req_reply_channel=None, client_id='', redis=None)

Bases: SICComponent

Abstract class for Sensors that provide data for the Social Interaction Cloud.

Start method calls the _produce method which calls the execute method in a loop.

Sensors must implement the execute method individually.

start()

Start the Sensor. Calls the _produce method to start producing output.

abstractmethod execute()

Main function of the sensor.

Must be implemented by the subclass.

Returns:

A SICMessage

Return type:

SICMessage

_produce()

Call the execute method in a loop until the stop event is set.

The output of the execute method is sent on the output channel.

actuator_python2.py

This module contains the SICActuator class.

class sic_framework.core.actuator_python2.SICActuator(ready_event=None, stop_event=None, log_level=10, conf=None, input_channel=None, output_channel=None, req_reply_channel=None, client_id='', redis=None)

Bases: SICComponent

Abstract class for Actuators that provide physical actions for the Social Interaction Cloud.

Actuators must implement the execute method individually.

on_request(request)

Handle a request from the client. Calls the execute method.

Parameters:

request (SICRequest) – input messages

Return type:

SICMessage

abstractmethod execute(request)

Main function of the Actuator. Must return a SICMessage as a reply to the user.

Must be implemented by the subclass.

Parameters:

request (SICRequest) – input messages

Return type:

SICMessage

service_python2.py

This module contains the SICService class, which is the base class for all services in the Social Interaction Cloud.

class sic_framework.core.service_python2.MessageQueue(logger)

Bases: deque

A message buffer, with logging to notify of excessive dropped messages.

Parameters:

logger (logging.Logger) – The logger to use for logging dropped messages.

__init__(logger)

appendleft(x)

Add an element to the left side of the deque.

exception sic_framework.core.service_python2.PopMessageException

Bases: ValueError

An exception to raise whenever the conditions to pop messages from the input message buffers were not met.

class sic_framework.core.service_python2.SICMessageDictionary

Bases: object

A dictionary container for messages, indexable by the message type and possibly an origin.

__init__()

set(message)

Add a message to the dictionary.

Parameters:

message (SICMessage) – The message to add.

get(type, source_component=None)

Get a message from the dictionary.

Parameters:

• type (SICMessage) – The type of message to get.

• source_component – The component that sent the message.

class sic_framework.core.service_python2.SICService(*args, **kwargs)

Bases: SICComponent

Abstract class for Services that transform and perform operations on data.

Some Services take multiple inputs and must synchronize the data based on timestamp. This class provides the logic to synchronize the data if multiple inputs are expected.

MAX_MESSAGE_BUFFER_SIZE = 10

MAX_MESSAGE_AGE_DIFF_IN_SECONDS = 0.5

__init__(*args, **kwargs)

start()

Start the Service. Calls the _listen method to start listening for input messages.

abstractmethod execute(inputs)

Process the input messages and return a SICMessage.

Main function of the Service. Must be implemented by the subclass.

Parameters:

inputs (SICMessageDictionary) – dict of input messages from other components

Returns:

A SICMessage or None

Return type:

SICMessage | None

_pop_messages()

Collect all input SICdata messages gathered in the buffers into a dictionary to use in the execute method. Make sure all input messages are aligned to the newest timestamp to synchronise the input data. If multiple channels contain the same type, give them an index in the service_input dict.

If the buffers do not contain an aligned set of messages, a PopMessageException is raised. :raises: PopMessageException :return: tuple of dictionary of messages and the shared timestamp

on_message(message)

Collect an input message and put it into the appropriate buffer.

Sets the _new_data_event to signal that new data is available.

Parameters:

message (SICMessage) – The message to collect.

_listen()

Wait for new data and execute when possible.

Supporting Elements

component_manager_python2.py

This module contains the SICComponentManager class, used to start, stop, and manage components.

class sic_framework.core.component_manager_python2.SICStartComponentRequest(component_name, log_level, input_channel, client_id, conf=None)

Bases: SICRequest

A request from a user to start a component.

Parameters:

• component_name (str) – The name of the component to start.

• log_level (logging.LOGLEVEL) – The logging level to use for the component.

• conf (SICConfMessage) – The configuration the component.

__init__(component_name, log_level, input_channel, client_id, conf=None)

class sic_framework.core.component_manager_python2.SICNotStartedMessage(message)

Bases: SICMessage

A message to indicate that a component failed to start.

Parameters:

message (str) – The message to indicate the failure.

__init__(message)

class sic_framework.core.component_manager_python2.SICComponentStartedMessage(output_channel, request_reply_channel)

Bases: SICMessage

__init__(output_channel, request_reply_channel)

class sic_framework.core.component_manager_python2.SICComponentManager(component_classes, client_id='', auto_serve=True, name='')

Bases: object

A component manager to start, stop, and manage components.

Parameters:

• component_classes (list) – List of Components this manager can start.

• auto_serve (bool) – Whether to automatically start serving requests.

MAX_REDIS_SERVER_TIME_DIFFERENCE = 2

COMPONENT_START_TIMEOUT = 10

__init__(component_classes, client_id='', auto_serve=True, name='')

serve()

Listen for requests to start/stop components until signaled to stop running.

start_component(request)

Start a component on this host as requested by a user.

Parameters:

request (SICStartComponentRequest) – The request to start the component.

Returns:

The reply to the request.

Return type:

SICMessage

stop(*args)

Stop the component manager.

Closes the redis connection and stops all active components.

Parameters:

args (tuple) – Additional arguments to pass to the stop method.

_sync_time()

Sync the time of components with the time of the redis server.

WORK IN PROGRESS: Does not work! clock on devices is often not correct, so we need to correct for this

_handle_request(request)

Handle user requests such as starting/stopping components and pinging the component manager.

Parameters:

request (SICRequest) – The request to handle.

Returns:

The reply to the request.

Return type:

SICMessage

connector.py

This module contains the SICConnector class, the user interface to connect to components.

exception sic_framework.core.connector.ComponentNotStartedError

Bases: Exception

An exception to indicate that a component failed to start.

class sic_framework.core.connector.SICConnector(ip='localhost', log_level=20, conf=None, input_source=None)

Bases: object

The user interface to connect to components wherever they are running.

Parameters:

• ip (str, optional) – The IP address of the component to connect to.

• log_level (logging.LOGLEVEL, optional) – The logging level to use for the connector.

• conf (SICConfMessage, optional) – The configuration for the connector.

__init__(ip='localhost', log_level=20, conf=None, input_source=None)

property component_class

The component class this connector is for.

Returns:

The component class this connector is for

Return type:

type[SICComponent]

send_message(message)

Send a message to the component.

Parameters:

message (SICMessage) – The message to send.

register_callback(callback)

Subscribe a callback to be called when there is new data available.

Parameters:

callback (function) – the function to execute.

request(request, timeout=100.0, block=True)

Send a request to the Component.

Waits until the reply is received. If the reply takes longer than timeout seconds to arrive, a TimeoutError is raised. If block is set to false, the reply is ignored and the function returns immediately.

Parameters:

• request (SICRequest) – The request to send to the component.

• timeout (float) – A timeout in case the action takes too long.

• block (bool) – If false, immediately returns None after sending the request.

Returns:

the SICMessage reply from the component, or None if blocking=False

Return type:

SICMessage | None

stop()

Send a stop request to the component and close the redis connection.

get_input_channel()

Get the input channel of the component.

get_output_channel()

Get the output channel of the component.

_ping()

Ping the component to check if it is alive.

Returns:

True if the component is alive, False otherwise.

Return type:

bool

_start_component()

Request the component to be started.

Returns:

The component we requested to be started

Return type:

SICComponent

Message System

sic_redis.py

A wrapper around Redis to provide a simpler interface for sending SICMessages, using two different APIs. The non-blocking (asynchronous) API is used for messages which are simply broadcasted and do not require a reply. The blocking (synchronous) API is used for requests, from which a reply is expected when the action is completed.

Example Usage: Non-blocking (asynchronous):

## DEVICE A

r.register_message_handler(“my_channel”, do_something_fn)

## DEVICE B

r.send_message(“my_channel”, SICMessage(“abc”))

Blocking (synchronous):

## DEVICE A

def do_reply(channel, request):

return SICMessage()

r.register_request_handler(“my_channel”, do_reply)

## DEVICE B

reply = r.request(“my_channel”, NamedRequest(“req_handling”), timeout=5)

# here the reply is received and stored in the variable ‘reply’.

class sic_framework.core.sic_redis.CallbackThread(function, pubsub, thread)

Bases: object

A thread that is used to listen to a channel and call a function when a message is received.

Parameters:

• function (function) – The function to call when a message is received.

• pubsub (redis.pubsub.PubSub) – The pubsub object to listen to.

• thread (threading.Thread) – The thread itself

__init__(function, pubsub, thread)

sic_framework.core.sic_redis.cleanup_on_exit()

Cleanup on exit. Close all Redis connections.

sic_framework.core.sic_redis.get_redis_db_ip_password()

Get the Redis database IP and password from environment variables. If not set, use default values.

Returns:

The Redis database IP and password.

Return type:

tuple[str, str]

class sic_framework.core.sic_redis.SICRedis(parent_name=None)

Bases: object

A custom version of Redis that provides a clear blocking and non-blocking API.

Parameters:

parent_name (str) – The name of the module that uses this Redis connection, for easier debugging.

__init__(parent_name=None)

register_message_handler(channels, callback, ignore_requests=True)

Subscribe a callback function to one or more channels, start a thread to monitor for new messages.

By default, ignores SICRequests. Registering request handlers calls this function but sets ignore_requests to False.

Parameters:

• callback (function) – a function expecting a SICMessage and a channel argument to process the messages received on channel

• channels (str or list[str]) – channel or channels to listen to.

• ignore_requests (bool) – Flag to control whether the message handler should also trigger the callback if the message is a SICRequest

Returns:

The CallbackThread object containing the the thread that is listening to the channel.

unregister_callback(callback_thread)

Unhook a callback by unsubscribing from Redis and stopping the thread. Will unregister all hooks if multiple hooks are created.

Parameters:

callback_thread (CallbackThread) – The CallbackThread to unregister.

send_message(channel, message)

Send a SICMessage on the provided channel to any subscribers.

Parameters:

• channel (str) – The Redis pubsub channel to communicate on.

• message (SICMessage) – The message to send.

Returns:

The number of subscribers that received the message.

Return type:

int

request(channel, request, timeout=5, block=True)

Send a request, and wait for the reply on the same channel. If the reply takes longer than timeout seconds to arrive, a TimeoutError is raised. If block is set to false, the reply is ignored and the function returns immediately.

Parameters:

• channel (str) – The Redis pubsub channel to communicate on.

• request (SICRequest) – The SICRequest

• timeout (float) – Timeout in seconds in case the reply takes too long.

• block (bool) – If false, immediately returns None after sending the request.

Returns:

the SICMessage reply

register_request_handler(channel, callback)

Register a function to listen to SICRequest’s (and ignore SICMessages). Handler must return a SICMessage as a reply. Will block receiving new messages until the callback is finished.

Parameters:

• channel (str) – The Redis pubsub channel to communicate on.

• callback (function) – function to run upon receiving a SICRequest. Must return a SICMessage reply

time()

Get the current time from the Redis server.

Returns:

The current time in seconds since the Unix epoch.

Return type:

tuple[int, int]

close()

Cleanup function to stop listening to all callback channels and disconnect Redis.

_reply(channel, request, reply)

Send a reply to a specific request. This is done by sending a SICMessage to the same channel, where the requesting thread/client is waiting for the reply.

Called by request handlers.

Parameters:

• channel (str) – The Redis pubsub channel to communicate on.

• request (SICRequest) – The SICRequest

• reply (SICMessage) – The SICMessage reply to send back to the requesting client.

static parse_pubsub_message(pubsub_msg)

Convert a redis pub/sub message to a SICMessage (sub)class. :param pubsub_msg: :return:

get_data_stream_map()

Get the data stream map from redis.

Returns a dictionary of data stream id to data stream information.

get_reservation_map()

Get the reservation map from redis.

Returns a dictionary of component id to client id.

get_data_stream(data_stream_id)

Get a specific data stream from redis.

Returns the data stream as a dictionary.

get_reservation(device_id)

Get a specific reservation from redis.

Returns the client id that has reserved the device.

set_data_stream(data_stream_id, data_stream_info)

Add a data stream in redis.

Parameters:

• data_stream_id – The id of the data stream

• data_stream_info – A dictionary containing the component id, input channel, and the client id its associated with

unset_data_stream(data_stream_id)

Remove a data stream in redis.

Parameters:

data_stream_id – The id of the data stream

set_reservation(device_id, client_id)

Add a reservation for a component in redis.

Parameters:

• device_id – The id of the device

• client_id – The id of the client reserving the component

Returns:

The number of keys set

unset_reservation(device_id)

Remove a reservation for a device in redis.

remove_client(client_id)

Remove a client’s reservations and data streams from redis.

Used if a client disconnects from the SIC server and their reservations and data streams are not removed properly.

Parameters:

client_id – The id of the client

ping_client(client_id)

Ping a client to see if they are still connected.

Parameters:

client_id – The id of the client

Returns:

True if the client is connected, False otherwise

message_python2.py

This module contains the SICMessage class, which is the base class for all messages in the SIC framework.

class sic_framework.core.message_python2.SICMessage

Bases: object

The abstract message structure to pass messages around the SIC framework. Supports python types, numpy arrays and JPEG compression using libturbo-jpeg.

Parameters:

• _compress_images (bool) – Whether to compress images.

• _request_id (int) – The request id of the message.

• _timestamp (float) – The timestamp of the message.

• _previous_component_name (str) – The name of the previous component that created the message.

classmethod get_message_name()

The pretty name of this message class.

Returns:

The name of the message class.

Return type:

str

classmethod deserialize(byte_string)

Convert object from its bytes representation, compatible between python2 and python3 and with support for numpy arrays.

Parameters:

byte_string (bytes) – The byte string to deserialize.

Returns:

The deserialized object.

Return type:

object

static _np2base(inp)

Convert numpy arrays to byte arrays.

Parameters:

inp (np.ndarray) – a numpy array

Returns:

the byte string

static _base2np(inp)

Convert back from byte arrays to numpy arrays.

Parameters:

inp (bytes) – a byte string

Returns:

the numpy array

static np2jpeg(inp)

Convert numpy array to JPEG bytes.

Parameters:

inp (np.ndarray) – a numpy array

Returns:

the JPEG bytes

static jpeg2np(inp)

Convert JPEG bytes to numpy array.

Parameters:

inp (bytes) – a JPEG bytes

Returns:

the numpy array

get_previous_component_name()

Get the name of the previous component that created the message.

Returns:

The name of the previous component.

Return type:

str

serialize()

Convert this object to its bytes representation, compatible between python2 and python3 and with support for numpy arrays.

Returns:

‘bytes’ in python3, ‘str’ in python2 (which are roughly the same)

Return type:

bytes or str

static _pickle_load(byte_string)

Load a pickle object from a byte string.

The pickle loads call is different between python versions. To reduce code duplication, this function is created to contain only the difference.

Parameters:

byte_string (bytes) – The byte string to load.

Returns:

The loaded object.

Return type:

object

class sic_framework.core.message_python2.SICConfMessage

Bases: SICMessage

A type of message that carries configuration information for services.

class sic_framework.core.message_python2.SICRequest(request_id=None)

Bases: SICMessage

A type of message that must be met with a reply, a SICMessage with the same request id, on the same channel.

__init__(request_id=None)

class sic_framework.core.message_python2.SICControlMessage

Bases: SICMessage

Superclass for all messages that are related to component control

class sic_framework.core.message_python2.SICControlRequest(request_id=None)

Bases: SICRequest

Superclass for all requests that are related to component control

class sic_framework.core.message_python2.SICPingRequest(request_id=None)

Bases: SICControlRequest

A request for a ping to check if alive.

class sic_framework.core.message_python2.SICPongMessage

Bases: SICControlMessage

A pong to reply to a ping request.;

class sic_framework.core.message_python2.SICSuccessMessage

Bases: SICControlMessage

Special type of message to signal a request was successfully completed.

class sic_framework.core.message_python2.SICStopRequest(request_id=None)

Bases: SICControlRequest

Special type of message to signal a device it should stop as the user no longer needs it.

class sic_framework.core.message_python2.SICIgnoreRequestMessage

Bases: SICControlMessage

Special type of message with the request_response_id set to -1. This means it will not be automatically set to the id of the request this is a reply to, and in effect will not reply to the request as the user will ignore this reply.

class sic_framework.core.message_python2.CompressedImage(image)

Bases: object

Compress WxHx3 np arrays using libturbo-jpeg to speed up network transfer of images. This is LOSSY JPEG compression, which means the image is not exactly the same. Non-image array content will be destroyed by this compression.

__init__(image)

class sic_framework.core.message_python2.CompressedImageMessage(*args, **kwargs)

Bases: CompressedImage, SICMessage

See CompressedImage

__init__(*args, **kwargs)

class sic_framework.core.message_python2.CompressedImageRequest(*args, **kwargs)

Bases: CompressedImage, SICRequest

See CompressedImage

__init__(*args, **kwargs)

class sic_framework.core.message_python2.UncompressedImageMessage(image)

Bases: SICMessage

Message class to send images/np array without JPEG compression. The data is compressed using default np.save lossless compression. In other words: the data does not change after compression, but this is much slower than JPEGCompressedImageMessage

__init__(image)

class sic_framework.core.message_python2.Audio(waveform, sample_rate)

Bases: object

A message that containes a _byte representation_ of pulse-code modulated (PCM) 16-bit signed little endian integer waveform audio data.

Integers are represented as a python byte array because this is the expected and provided data format of common hardware audio hardware and libraries. For compatibility with other services ensure that your data follows EXACTLY this data type. This should be the most common format, but please check your data format.

You can convert to and from .wav files using the built-in module https://docs.python.org/2/library/wave.html

__init__(waveform, sample_rate)

class sic_framework.core.message_python2.AudioMessage(*args, **kwargs)

Bases: Audio, SICMessage

Message class to send audio data.

__init__(*args, **kwargs)

class sic_framework.core.message_python2.AudioRequest(*args, **kwargs)

Bases: Audio, SICRequest

Request class to send audio data.

__init__(*args, **kwargs)

class sic_framework.core.message_python2.Text(text)

Bases: object

A simple object with a string as text.

__init__(text)

class sic_framework.core.message_python2.TextMessage(*args, **kwargs)

Bases: Text, SICMessage

Message class to send text data.

__init__(*args, **kwargs)

class sic_framework.core.message_python2.TextRequest(*args, **kwargs)

Bases: Text, SICRequest

Request class to send text data.

__init__(*args, **kwargs)

class sic_framework.core.message_python2.BoundingBox(x, y, w, h, identifier=None, confidence=None)

Bases: object

Bounding box for identifying an object in an image.

(x,y) represents the top-left pixel of the bounding box, and (w,h) indicates the width and height. Identifier can be used implementation specific to for example indicate a specific object type or detected person. Confidence indicates the confidence of the detection mechanism.

__init__(x, y, w, h, identifier=None, confidence=None)

xywh()

Get the coordinates as a numpy array.

Returns:

The coordinates as a numpy array.

Return type:

np.ndarray

class sic_framework.core.message_python2.FakeTurboJpeg

Bases: object

decode(bytes)

encode(array)

class sic_framework.core.message_python2.BoundingBoxesMessage(bboxes)

Bases: SICMessage

Message class to send multiple bounding boxes.

__init__(bboxes)

class sic_framework.core.message.SICMessage

Bases: SICMessage

__init__() → None

class sic_framework.core.message.SICConfMessage

Bases: SICConfMessage

__init__() → None

Utilities

utils.py

This module contains utility functions for the SIC framework.

sic_framework.core.utils.get_ip_adress()

Get the IP address of the device.

Returns:

The IP address of the device.

Return type:

str

sic_framework.core.utils.ping_server(server, port, timeout=3)

Ping a server to check if it is reachable.

Parameters:

• server (str) – The server to ping.

• port (int) – The port to ping.

• timeout (float) – The timeout in seconds.

Returns:

True if the server is reachable, False otherwise.

Return type:

bool

sic_framework.core.utils.get_username_hostname_ip()

Get the username, hostname and IP address of the device.

Returns:

The username, hostname and IP address of the device.

Return type:

str

sic_framework.core.utils.ensure_binary(s, encoding='utf-8', errors='strict')

From a future six version. Coerce s to six.binary_type.

For Python 2:

• unicode -> encoded to str

• str -> str

For Python 3:

• str -> encoded to bytes

• bytes -> bytes

Parameters:

• s (str) – The string to convert.

• encoding (str) – The encoding to use.

• errors (str) – The error handling strategy.

Returns:

The converted string.

Return type:

str

sic_framework.core.utils.str_if_bytes(data)

Compatibility for the channel names between python2 and python3 a redis channel b’name’ differes from “name”

Parameters:

data (str or bytes) – The data to convert.

Returns:

The converted string.

Return type:

str

sic_framework.core.utils.random_hex(nbytes=8)

Generate a random hex string.

Parameters:

nbytes (int) – The number of bytes to generate.

Returns:

The random hex string.

Return type:

str

sic_framework.core.utils.is_sic_instance(obj, cls)

Return True if the object argument is an instance of the classinfo argument, or of a (direct, indirect, or virtual) subclass thereof.

isinstance does not work when pickling object, so a looser class name check is performed. https://stackoverflow.com/questions/620844/why-do-i-get-unexpected-behavior-in-python-isinstance-after-pickling :param obj: :param cls: :return:

sic_framework.core.utils.type_equal_sic(a, b)

Check if two objects are of the same type.

type(a) == type(b), but with support for objects transported across the network with pickle.

Parameters:

• a (object) – The first object.

• b (object) – The second object.

Returns:

sic_framework.core.utils.zip_directory(path)

Create a compressed zip file from the given directory path.

Parameters:

path (str) – The path to the directory to be zipped.

Returns:

The path to the created zip file.

Return type:

str

Raises:

FileNotFoundError: If the path doesn’t exist

sic_framework.core.utils.create_data_stream_id(component_id, input_source, length=16)

Hashes component info into a short, random-looking string.

Parameters:

• component_id (str) – Component identifier.

• input_source (str) – Input stream name.

• length (int) – Length of the resulting string (default 16).

Returns:

A base64-encoded truncated hash string.

Return type:

str

utils_cv2.py

This module contains utility functions for working with OpenCV.

sic_framework.core.utils_cv2.draw_bbox_on_image(bbox, img, color=(0, 255, 0))

Draws a bounding box on an image.

Parameters:

• bbox – a BoundingBox object

• img – numpy image

• color – tuple (r, g, b) with the color of the bounding box

Returns:

sic_logging.py

This module contains the SICLogging class, which is used to log messages to the Redis log channel and a local logfile.

sic_framework.core.sic_logging.get_log_channel(client_id='')

Get the global log channel. All components on any device should log to this channel.

class sic_framework.core.sic_logging.SICLogMessage(msg, client_id='')

Bases: SICMessage

__init__(msg, client_id='')

A wrapper for log messages to be sent over the SICRedis pubsub framework. :param msg: The log message to send to the user

exception sic_framework.core.sic_logging.SICRemoteError

Bases: Exception

An exception indicating the error happened on a remote device

class sic_framework.core.sic_logging.SICCommonLog

Bases: object

A class to subscribe to a Redis log channel and write all log messages to a logfile.

Pseudo singleton object. Does nothing when this file is executed during the import, but can subscribe to the log channel for the user with subscribe_to_redis_log once.

Parameters:

• redis (SICRedis) – The Redis instance to use for logging.

• logfile (str) – The file path to write the log to.

__init__()

subscribe_to_redis_log(client_id='')

Subscribe to the Redis log channel and display any messages on the terminal. This function may be called multiple times but will only subscribe once.

Returns:

None

_handle_redis_log_message(message)

Handle a message sent on a debug stream. Currently it’s just printed to the terminal.

Parameters:

message (SICLogMessage) – The message to handle.

_write_to_logfile(message)

Write a message to the logfile.

Parameters:

message (str) – The message to write to the logfile.

stop()

Stop the logging.

class sic_framework.core.sic_logging.SICRedisHandler(redis, client_id)

Bases: Handler

Facilities to log to Redis as a file-like object, to integrate with standard python logging facilities.

Parameters:

• redis (SICRedis) – The Redis instance to use for logging.

• client_id (str) – The client id of the device that is logging

__init__(redis, client_id)

Initializes the instance - basically setting the formatter to None and the filter list to empty.

emit(record)

Emit a log message to the Redis log channel.

Parameters:

record (logging.LogRecord) – The log record to emit.

readable()

Check if the stream is readable.

Returns:

False

Return type:

bool

writable()

Check if the stream is writable.

Returns:

True

Return type:

bool

write(msg)

Write a message to the Redis log channel.

Parameters:

msg (str) – The message to write to the Redis log channel.

flush()

Flush the stream.

class sic_framework.core.sic_logging.SICLogFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

Bases: Formatter

A formatter for SIC log messages.

LOG_COLORS = {10: '\x1b[92m', 20: '\x1b[94m', 30: '\x1b[93m', 40: '\x1b[91m', 50: '\x1b[101m\x1b[97m'}

RESET_COLOR = '\x1b[0m'

format(record)

Format a log message.

Parameters:

record (logging.LogRecord) – The log record to format.

formatException(exec_info)

Prepend every exception with a | to indicate it is not local.

Parameters:

exec_info (tuple) – The exception information.

Returns:

The formatted exception.

Return type:

str

sic_framework.core.sic_logging.get_sic_logger(name='', client_id='', redis=None, log_level=10)

Set up logging to the log output channel to be able to report messages to users.

Parameters:

• name (str) – A readable and identifiable name to indicate to the user where the log originated

• client_id (str) – The client id of the device that is logging

• redis (SICRedis) – The SICRedis object

• log_level (int) – The logger.LOGLEVEL verbosity level

Returns:

The logger.

Return type:

logging.Logger