RTIS Dev Documentation - Table of Content

• General Example
• Classes
  • RTISMeasurement
  • RTISSettings
  • MeasureExternalTriggerQueueThread
  • MeasureExternalTriggerCallbackThread
• Methods
  • open_connection
  • close_connection
  • set_recording_settings
  • set_processing_settings
  • get_current_settings_config_name_list
  • get_current_settings
  • clear_current_settings
  • get_settings
  • set_settings_from_class
  • get_premade_processing_settings_list
  • get_premade_recording_settings_list
  • get_microphone_layout_list
  • prepare_processing
  • unload_processing
  • get_raw_measurement
  • get_signal_measurement
  • get_processed_measurement
  • process_measurement
  • set_counter
  • set_behaviour
  • get_firmware_version
  • create_measure_external_trigger_queue
  • create_measure_external_trigger_callback
  • create_processing_workers
  • toggle_amplifier
  • toggle_external_triggers
  • custom_command
  • reset_device
  • set_log_mode
  • set_custom_logger

General Example

Here is a small example that goes over most basic steps:

import rtisdev
import matplotlib.pyplot as plt
import numpy as np

Open a connection to the RTIS Device over the default serial port:

success_connect = rtisdev.open_connection()

Set the recording settings with 163840 samples and a call sweep between 25 and 50 KHz:

config_uuid = rtisdev.set_recording_settings(microphoneSamples=163840, callMinimumFrequency=25000, callMaximumFrequency=50000)

Enable all processing steps and preload them with RTIS CUDA. This will produce a 2D energyscape with 91 directions with a maximum distance of 6m:

success_settings_processing = rtisdev.set_processing_settings(directions=91, maxRange=5, configName=config_uuid)

Get the used settings as a RTISSettings object:

settings = rtisdev.get_current_settings(configName=config_uuid)

Get an ACTIVE measurement (protect your ears!) in raw data:

measurement = rtisdev.get_raw_measurement(True, configName=config_uuid)

Store the raw data of that measurement as a binary file. This can be opened in another application for further work:

raw_data_sonar = measurement.rawData.tobytes()
file_handle_data = open("test_measurement_" + str(measurement.index) + ".bin", "wb")
file_handle_data.write(raw_data_sonar)
file_handle_data.close()

Process that raw measurement to an energyscape using the configuration chosen earlier:

processed_measurement = rtisdev.process_measurement(measurement, configName=config_uuid)

Get a new ACTIVE measurement (protect your ears!) in both raw and processed data formats directly:

new_processed_measurement = rtisdev.get_processed_measurement(True, configName=config_uuid)

Plot the 2D energyscape of this processed measurement using matplotlib:

plt.imshow(np.transpose(new_processed_measurement.processedData), cmap="hot", interpolation='nearest')
plt.xlabel("Directions (degrees)")
plt.ylabel("Range (meters)")
indexes_x = np.arange(0, new_processed_measurement.processedData.shape[0], 20)
labels_x = np.round(np.rad2deg(settings.directions[indexes_x, 0])).astype(int)
indexes_y = np.arange(0, new_processed_measurement.processedData.shape[1], 100)
labels_y = settings.ranges[indexes_y]
fmt_x = lambda x: "{:.0f}°".format(x)
fmt_y = lambda x: "{:.2f}m".format(x)
plt.xticks(indexes_x, [fmt_x(i) for i in labels_x])
plt.yticks(indexes_y, [fmt_y(i) for i in labels_y])
plt.title("RTIS Dev - 2D Energyscape Example")
ax = plt.gca()
ax.invert_yaxis()
ax.set_aspect("auto")
plt.show()

Get a new ACTIVE measurement (protect your ears!) in both raw and microphone signal format directly:

signal_measurement = rtisdev.get_signal_measurement(True, configName=config_uuid)

Plot the microphone signals of this measurement:

fig, axs = plt.subplots(8, 4, figsize=(10,16), constrained_layout = True)
for microphone_index_i in range(0, 8):
    for microphone_index_j in range(0, 4):
        axs[microphone_index_i, microphone_index_j].set_title(str(microphone_index_j+(microphone_index_i*4)+1))
        axs[microphone_index_i, microphone_index_j].plot(signal_measurement.processedData[microphone_index_j+(microphone_index_i*4),:])
        if microphone_index_j != 0:
            plt.setp(axs[microphone_index_i, microphone_index_j], yticklabels=[])
        if microphone_index_i != 7:
            plt.setp(axs[microphone_index_i, microphone_index_j], xticklabels=[])
        if microphone_index_i == 7:
            axs[microphone_index_i, microphone_index_j].set_xlabel("Time (Samples)")
        if microphone_index_j == 0:
            axs[microphone_index_i, microphone_index_j].set_ylabel("Amplitude")
plt.show()
fig.suptitle("RTIS Dev - Microphone Signals")

Classes

RTISMeasurement

class RTISMeasurement(id: str='', timestamp: float=0, behaviour: bool=False, index: int=0, rawData: np.ndarray=None, processedData: np.ndarray=None, configName: str='') [source]

Class storing all data and information on an RTIS device measurement.

RTISSettings

class RTISSettings(firmwareVersion: str, configName: str) [source]

Class describing all the processing and recording settings related to RTIS devices. Too many variables to describe here. Check the source-code for more information on which variables are available. Can be converted to a dictionary.

MeasureExternalTriggerQueueThread

class MeasureExternalTriggerQueueThread(*args, **kwargs) [source]

The class based on a Multiprocessing Process to start RTIS sonar measurements triggered by an external trigger. To set the data queue correctly use set_queue(dataQueue) function. the RTISMeasurement objects will then be put on this queue. To start the process use the start() function. To stop use the stop_thread() function. By default, using a signal.SIGINT exit (ex. using CTRL+C) will gracefully end the script.

Use create_measure_external_trigger_queue(dataQueue) to make an easy to use the class.

MeasureExternalTriggerCallbackThread

class MeasureExternalTriggerCallbackThread(*args, **kwargs) [source]

The class based on a Multiprocessing Process to start RTIS sonar measurements triggered by an external trigger. To set the callback function correctly use set_callback(callback) function. Your callback function should only have one argument, the RTISMeasurement data package. To start the process use the start() function. To stop use the stop_thread() function. By default, using a signal.SIGINT exit (ex. using CTRL+C) will gracefully end the script.

Use create_measure_external_trigger_callback(save_callback) to make an easy to use the class.

Methods

open_connection

def open_connection(port: str='/dev/ttyACM0', allowDebugMode: bool=False ) [source]

Connect to the port of the RTIS Hardware.

close_connection

def close_connection() [source]

Manually close the connection to the RTIS device. Normally, when your script ends without exceptions the connection will automatically be closed gracefully. This will also unload all RTIS CUDA workers.

set_recording_settings

def set_recording_settings(premade: str=None, jsonPath: str=None, callCustom: str=None, microphoneSamples: int=163840, microphoneSampleFrequency: int=4500000, callSampleFrequency: int=450000, callDuration: float=2.5, callMinimumFrequency: int=25000, callMaximumFrequency: int=50000, callEmissions: int=1, configName: str= '', applyToDevice: bool=True) [source]

Set the recording settings. All parameters are optional and most have default values. Please read their decription carefully.

Examples

You can get the available premade settings with get_premade_recording_settings_list(). Create settings from a premade setup:

config_uuid = rtisdev.set_recording_settings(premade="short_20_80")

Create settings from a json file. This expects a json to be available with a format such as seen below. Here we use auto-generated pulse call to emit. More examples can be found here. An example json:

{
    "microphoneSamples" : 294912,
    "microphoneSampleFrequency" : 4500000,
    "callSampleFrequency" : 450000,
    "callDuration" : 2.5,
    "callMinimumFrequency" : 25000,
    "callMaximumFrequency" : 50000,
    "callEmissions": 1
}

config_uuid = rtisdev.set_recording_settings(jsonPath="./myrecordingsettings.json")

Create settings from a json file. This expects a json to be available with a format such as seen below. Here we use manually generated call. It has to be available on the given path and have the right format. An example of such a custom call can be found hereflutter.csv:

{
    "microphoneSamples" : 16777216,
    "microphoneSampleFrequency" : 4500000,
    "callSampleFrequency" : 450000,
    "callCustom": "mycall.csv",
    "callEmissions": 1
}

config_uuid = rtisdev.set_recording_settings(jsonPath="./myrecordingsettings.json")

Create full custom settings with the arguments. All arguments that aren't filled in will use default values:

config_uuid = rtisdev.set_recording_settings(microphoneSamples=294912, callDuration=3, callMinimumFrequency=25000, callMaximumFrequency=80000)

Load in manually generated call. This requires the file to exist on the path and have the right format. An example of such a custom call can be found hereflutter.csv:

config_uuid = rtisdev.set_recording_settings(callCustom="mycall.csv")

Note that when multiple recording configurations are loaded, the module will automatically load the settings as asked for by the configName argument to the RTIS device before performing a new measurement.

set_processing_settings

def set_processing_settings(configName: str, premade: str=None, jsonPath: str=None, customPath: str=None, microphoneLayout: str='eRTIS_v3D1', mode: int=1, directions: int=181, azimuthLowLimit: float=-90, azimuthHighLimit: float=90, elevationLowLimit: float=-90, elevationHighLimit: float=90, elevation2DAngle: float=0, minRange: float=0.5, maxRange: float=5, pdmEnable: bool=True, preFilterEnable: bool=False, matchedFilterEnable: bool=True, beamformingEnable: bool= True, postFilterEnable: bool=False, enveloppeEnable: bool=True, cleanEnable: bool=True, preloadToggle: bool=True, preFilter: np.ndarray =None, postFilter: np.ndarray=None, meanEnergyRangeMultiplier: float=2, maxEnergyRangeThresholdMultiplier: float=0.5) [source]

Set the processing settings. All parameters are optional and most have default values. Please read their decription carefully.

Examples

You can get the available premade settings with get_premade_recording_settings_list(). Create settings from a premade setup with all processing steps on:

rtisdev.set_processing_settings(premade="3D_5m_3000", pdmEnable=True, matchedFilterEnable=True, preFilterEnable=False, beamformingEnable=True, enveloppeEnable=True, postFilterEnable=False, cleanEnable=True, preloadToggle=True, configName=config_uuid)

You don't have to define all the processing steps, as they are all on by default:

rtisdev.set_processing_settings(directions=91, configName=config_uuid)

Create settings from a premade setup with only part of the processing steps enabled and no preloading. You can get the available premade settings with get_premade_recording_settings_list():

rtisdev.set_processing_settings(pdmEnable=True, preFilterEnable=False, matchedFilterEnable=True, beamformingEnable=False, enveloppeEnable=False, cleanEnable=False, configName=config_uuid)

Create settings from a json file with full processing settings on. This expects a json to be available with a premade a format such as seen below. Note that the json does not include support for pre- and post-filters. Any other setting not defined in the json example below should also still be set manually as argument if the default value is not desired. An example of such json files can be found here. Here we use auto-generated processing files:

{
    "microphoneLayout" : "eRTIS_v3D1",
    "minRange" : 0.5,
    "maxRange" : 5,
    "directions": 181,
    "azimuthLowLimit": -30,
    "azimuthHighLimit": 30,
    "2D": 1
}

rtisdev.set_processing_settings(jsonPath="./myprocessingsettings.json", configName=config_uuid)

Create settings from a json file with full processing settings on. This expects a json to be available with a format such as seen below. Here we use manually generated processing files. They have to be available on these paths and have the right format. Note that the json does not include support for pre- and post-filters. Any other setting not defined in the json example below should also still be set manually as argument if the default value is not desired. An example of such custom processing files can be found here as well:

{
    "microphoneLayout" : "eRTIS_v3D1",
    "directionsCustom": "./directions.csv",
    "delayMatrixCustom": ".premade/delaymatrix.csv",
    "rangesCustom": ".premade/ranges.csv"
}

rtisdev.set_processing_settings(jsonPath="./myprocessingsettings.json", configName=config_uuid)

Create full custom settings with the arguments. All arguments that aren't filled in will use default values:

rtisdev.set_processing_settings(mode = 0, directions = 1337, minRange = 0.5, configName=config_uuid)

Load in manually generated processing files. This requires 3 files to exist in the given path: delaymatrix.csv, directions.csv and ranges.csv. Don't forget to also perhaps set the microphoneLayout and microphoneSampleFrequency values correctly as these are absent in these csv files! Note that the custom paths does not include support for pre- and post-filters. Any other setting not should also still be set manually as argument if the default value is not desired. An example of such custom processing files can be found here:

rtisdev.set_processing_settings(customPath="mysettingsfolder", configName=config_uuid)

The pre-filter is an optional filter to be performed after PDM filtering and before matched filter. It should be created using a scipy firwin filter function as in the example below:

pref = scipy.signal.firwin(513, 20000 / (450000 / 2), pass_zero=False).astype(np.float32)
rtisdev.set_processing_settings(postFilter=pref, preFilterEnable=True, configName=config_uuid)

Similarly, The post-beamforming filter is an optional filter to be performed after beamforming. It should be created using a scipy firwin filter function as in the example below:

postf = scipy.signal.firwin(512, [40000 / (450000 / 2), 50000 / (450000 / 2)], pass_zero=False).astype(np.float32)
rtisdev.set_processing_settings(postFilter=postf, postFilterEnable=True, configName=config_uuid)

get_current_settings_config_name_list

def get_current_settings_config_name_list() [source]

Get a list of names of all the currently loaded configurations.

get_current_settings

def get_current_settings(configName: str='') [source]

Returns all(dict) or a single RTISSettings object of the current settings for processing and recording.

clear_current_settings

def clear_current_settings(configName: str='') [source]

Clear all or the current applied RTISSettings configuration depending on setting the configName parameter.

get_settings

def get_settings(recordingPremade: str=None, recordingJsonPath: str=None, recordingCallCustom: str=None, processingPremade: str=None, processingJsonPath: str=None, processingCustomPath: str=None, microphoneSamples: int=163840, microphoneSampleFrequency: int=4500000, callSampleFrequency: int=450000, callDuration: float=2.5, callMinimumFrequency: int=25000, callMaximumFrequency: int=50000, callEmissions: int=1, microphoneLayout: str='eRTIS_v3D1', mode: int=1, directions: int=181, azimuthLowLimit: float=-90, azimuthHighLimit: float=90, elevationLowLimit: float=-90, elevationHighLimit: float=90, elevation2DAngle: float=0, minRange: float=0.5, maxRange: float=5, pdmEnable: bool=True, preFilterEnable: bool=False, matchedFilterEnable: bool=True, beamformingEnable: bool=True, postFilterEnable: bool=False, enveloppeEnable: bool=True, cleanEnable: bool=True, preloadToggle: bool =True, preFilter: np.ndarray=None, postFilter: np.ndarray=None, meanEnergyRangeMultiplier: float=2, maxEnergyRangeThresholdMultiplier: float=0.5, configName: str='') [source]

Returns an RTISSettings object with all chosen recording and processing settings based on the given arguments. It will not set these settings to the RTIS Device or activate processing. It only creates the settings object. For examples of what some of these settings do and how to use them, please see the set_recording_settings() and set_processing_settings() examples.

set_settings_from_class

def set_settings_from_class(settings: RTISSettings, applyToDevice: bool=True ) [source]

Set the wanted settings from an RTISSettings object. These can be created with the get_settings() or get_current_settings() methods.

get_premade_processing_settings_list

def get_premade_processing_settings_list() [source]

Get a list of names of all the available premade settings for processing.

get_premade_recording_settings_list

def get_premade_recording_settings_list() [source]

Get a list of names of all the available premade settings for recording.

get_microphone_layout_list

def get_microphone_layout_list() [source]

Get a list of names of all the available microphone layouts that are available for recording.

prepare_processing

def prepare_processing(configName: str='') [source]

Start the CUDA workers for looped measurements with processing enabled. It is not required to run this method for doing processing, but it will speed up the workflow significantly if doing many processed measurements at a high frequency. Furthermore, if using the default settings for processing this is enabled already. If no config name parameter is provided it will assume only one settings configuration is available and will prepare that one.

unload_processing

def unload_processing(configName: str='') [source]

Stop all CUDA workers of all workers or of one specified if the configuration name is provided. Only required if actually using preloading of CUDA workers. CUDA workers are also automatically stopped when your script ends.

get_raw_measurement

def get_raw_measurement(behaviour: bool=False, configName: str='' ) [source]

Start an RTIS sonar measurement and return the raw data in an RTISMeasurement object. This means that it will only record and not perform any processing.

Examples

Create a connection, set recording settings and make a raw measurement with passive behaviour:

import rtisdev
rtisdev.open_connection()
config_uuid = rtisdev.set_recording_settings(callMinimumFrequency=25000, callMaximumFrequency=50000)
measurement = rtisdev.get_raw_measurement(True, configName=config_uuid)

Note that when multiple recording configurations are loaded, the module will automatically load the settings as asked for by the configName argument to the RTIS device before performing a new measurement.

get_signal_measurement

def get_signal_measurement(behaviour: bool=False, configName: str='' ) [source]

Start an RTIS sonar measurement and process it with only PDM filtering and subsampling enabled to get the microphone signals returned in an RTISMeasurement object. This means it will overwrite the enabled and disabled processing steps that the user might have set. But will still use the other chosen recording and processing settings.

Examples

Create a connection, set recording and processing settings and make a signal measurement with active behaviour:

import rtisdev
rtisdev.open_connection()
config_uuid = rtisdev.set_recording_settings(callMinimumFrequency=25000, callMaximumFrequency=50000)
signal_measurement = rtisdev.get_signal_measurement(True, configName=config_uuid)

Note that when multiple recording configurations are loaded, the module will automatically load the settings as asked for by the configName argument to the RTIS device before performing a new measurement.

get_processed_measurement

def get_processed_measurement(behaviour: bool=False, configName: str='' ) [source]

Start an RTIS sonar measurement and process it and return the raw and processed data in an RTISMeasurement object. This will use the chosen recording and processing settings.

Examples

Create a connection, set recording and processing settings and make a processed measurement with active behaviour:

import rtisdev
rtisdev.open_connection()
config_uuid = rtisdev.set_recording_settings(callMinimumFrequency=25000, callMaximumFrequency=50000)
rtisdev.set_processing_settings(directions=91, configName=config_uuid)
processed_measurement = rtisdev.get_processed_measurement(True, configName=config_uuid)

Note that when multiple recording configurations are loaded, the module will automatically load the settings as asked for by the configName argument to the RTIS device before performing a new measurement.

process_measurement

def process_measurement(measurement: RTISMeasurement, configName: str='' ) [source]

Process a previously recorded raw RTIS sonar measurement from a RTISMeasurement object and return same measurement with processed data in a new RTISMeasurement object.

Examples

Create a connection, set recording and processing settings and make a raw measurement with active behaviour. Then afterward process it:

import rtisdev
rtisdev.open_connection()
config_uuid = rtisdev.set_recording_settings(callMinimumFrequency=25000, callMaximumFrequency=50000)
rtisdev.set_processing_settings(directions=91, configName=config_uuid)
measurement = rtisdev.get_raw_measurement(True, configName=config_uuid)
processed_measurement = rtisdev.process_measurement(measurement, configName=config_uuid)

set_counter

def set_counter(newCount: int=0) [source]

Set the internal measurement counter of the sonar hardware.

set_behaviour

def set_behaviour(mode: bool) [source]

Set the behaviour of the sonar hardware to passive or active. This is only necessary if using external measurement triggers. As using the normal RTIS Dev functions of get_raw_measurement(behaviour), get_signal_measurement(behaviour) and get_processed_measurement(behaviour) will use the given function argument to define the sonar behaviour.

get_firmware_version

def get_firmware_version() [source]

Get the firmware version of the internal RTIS firmware used on the device.

create_measure_external_trigger_queue

def create_measure_external_trigger_queue(dataQueue: Queue, configName: str='' ) [source]

This will create and return a Multiprocessing Process that will be waiting for an external trigger to measure from the RTIS Device and afterward put this measurement on a data queue. This method will return a MeasureExternalTriggerQueueThread.

Examples

Create a queue to save the measurement to and assign it to the process:

from multiprocessing import Manager
import rtisdev
rtisdev.open_connection()
config_uuid = rtisdev.set_recording_settings(callMinimumFrequency=25000, callMaximumFrequency=50000)
rtisdev.set_processing_settings(directions=91,configName=config_uuid)
manager = Manager()
dataQueue = manager.Queue()
measure_thread = rtisdev.create_measure_external_trigger_queue(dataQueue, config_uuid)
measure_thread.start()
measure_thread.join()

create_measure_external_trigger_callback

def create_measure_external_trigger_callback(callback: Callable[[ RTISMeasurement], any], configName: str='' ) [source]

This will create and return a Multiprocessing Process that will be waiting for an external trigger to measure from the RTIS Device and afterward put this measurement on a data queue. This method will return a MeasureExternalTriggerCallbackThread.

Examples

Create a callback to save the measurement to disk:

import rtisdev
rtisdev.open_connection()
config_uuid = rtisdev.set_recording_settings(callMinimumFrequency=25000, callMaximumFrequency=50000)
rtisdev.set_processing_settings(directions=91, configName=config_uuid)
index = 0
def save_callback(measurement=None):
    if measurement is not None :
        if measurement.rawData is not None:
            data_sonar = measurement.rawData.tobytes()
            file_handle_data = open(str(index) + ".bin","wb")
            file_handle_data.write(data_sonar)
            file_handle_data.close()
            index = index + 1
measure_thread = rtisdev.create_measure_external_trigger_callback(save_callback, config_uuid)
measure_thread.start()
measure_thread.join()

create_processing_workers

def create_processing_workers(workerCount: int, inputQueue: Queue, outputQueue: Queue, configName: str='') [source]

This will create and return a Multiprocessing Pool that will generate a chosen amount of processing workers to handle incoming RTISMeasurement objects on the input Multiprocessing Queue and after processing place them on the output Multiprocessing Queue.

Please set the preloadToggle argument to False when using set_processing_settings() and/or make sure to not use set_processing_settings() before this point as it will cause an error or result in a potential crash of RTIS Dev!

Examples

Create the data queues, set up the worker pool with 4 workers, generate some measurements and afterward parse all these measurements by getting them from the output queue. Once the work is done, terminate the workers gracefully:

from multiprocessing import Manager
import rtisdev
rtisdev.open_connection()
config_uuid = rtisdev.set_recording_settings(callMinimumFrequency=25000, callMaximumFrequency=50000)
rtisdev.set_processing_settings(directions=91, configName=config_uuid, preloadToggle=False)
manager = Manager()
inputQueue = manager.Queue()
outputQueue = manager.Queue()
workersPool = rtisdev.create_processing_workers(4, inputQueue, outputQueue, config_uuid)
for measurement_index in range(0, 30):
    measurement = rtisdev.get_raw_measurement(configName=config_uuid)
    inputQueue.put(measurement)
for measurement_index in range(0, 30):
    measurement = outputQueue.get()
workersPool.terminate()

toggle_amplifier

def toggle_amplifier(mode: bool) [source]

Enable/disable the high voltage amplifier's step up controller. It is enabled by default so has to be manually disabled if wanted. This will save on power usage and heat production.

toggle_external_triggers

def toggle_external_triggers(mode: bool, pin: int=1) [source]

Enable/disable external triggers being able to start a measurement on the RTIS device. They are disabled by default so have to be manually enabled. You can also set the input pin (1 or 2).

custom_command

def custom_command(command: str) [source]

Send a custom command to the RTIS device to execute over serial. This is usually in the format of !c,i,j,k. With c being the command ID character and i,j and k being the three comma-seperated command values.

reset_device

def reset_device(stm32pin: int=7) [source]

The function to reset the RTIS device hardware.

set_log_mode

def set_log_mode(mode: int) [source]

The function to set the logging level of the RTIS Dev module.

set_custom_logger

def set_custom_logger(customLogger: logging.Logger) [source]

The function to set a custom logger to be used by RTIS Dev.