Built-in Apps

Built-in applications

class switchio.apps.AppManager(pool, ppfuncargs=None, **kwargs)

Manage apps over a cluster/slavepool.


Iterable over all unique contained subapps

load_app(app, app_id=None, ppkwargs=None, with_measurers=())

Load and activate an app for use across all slaves in the cluster.

load_multi_app(apps_iter, app_id=None, **kwargs)

Load a “composed” app (multiple apps using a single app name/id) by providing an iterable of (app, prepost_kwargs) tuples. Whenever the app is triggered from and event loop all callbacks from all apps will be invoked in the order then were loaded here.

switchio.apps.app(*args, **kwargs)

Decorator to register switchio application classes. Example usage:

class CoolAppController(object):

# This will register the class as a switchio app.
# The name of the app defaults to `class.__name__`.
# The help for the app is taken from `class.__doc__`.

# You can also provide an alternative name via a
# decorator argument:

class CoolAppController(object):

# or with a keyword arg:

class CoolAppController(object):

Get a registered app by name or None if one isn’t registered.


Return an iterable which delivers tuples (<modulename>, <apps_subiter>)


Iterable over all registered apps.

switchio.apps.load(packages=(), imp_excs=('pandas', ))

Load by importing all built-in apps along with any apps found in the provided packages list.

Parameters:packages (str | module) – package (names or actual modules)
Return type:dict[str, types.ModuleType]
switchio.apps.register(cls, name=None)

Register an app in the global registry

Load testing

Call generator app for load testing

class switchio.apps.call_gen.Originator(slavepool, debug=False, auto_duration=True, app_id=None, **kwargs)

An auto-dialer built for stress testing.


Compare current state to ident


Hangup all calls for all slaves, period, even if they weren’t originated by this instance and stop the burst loop.


Send the ‘hupall’ command to hangup all active calls.


Indicate whether the call burst thread is up

load_app(app, app_id=None, ppkwargs={}, weight=1, with_metrics=True)

Load a call control app for use across the entire slave cluster.

If app is an instance then it’s state will be shared by all slaves. If it is a class then new instances will be instantiated for each Client-Observer pair and thus state will have per slave scope.


The maximum rate value which can be set. Setting rate any higher will simply clip to this value.


Originate str used for making calls


Apply load test settings on the slave server


Shutdown this originator instance and hanging up all active calls and triggering the burst loop to exit.


Start the originate burst loop by starting and/or notifying a worker thread to begin. Changes state INITIAL | STOPPED -> ORIGINATING.


The current operating state as a string


Stop originate loop if currently originating sessions. Change state ORIGINATING -> STOPPED


Return bool indicating if in the stopped state.

waitforstate(state, **kwargs)

Block until the internal state matches state.

waitwhile(predicate=None, **kwargs)

Block until predicate evaluates to False.

The default predicate waits for all calls to end and for activation of the “STOPPED” state.

See switchio.utils.waitwhile for more details on predicate usage.

class switchio.apps.call_gen.State(state=0)

Enumeration to represent the originator state machine

class switchio.apps.call_gen.WeightedIterator(counter=None)

Pseudo weighted round robin iterator. Delivers items interleaved in weighted order.


Endlessly iterates the most up to date keys in counts. Allows for real-time weight updating from another thread.

switchio.apps.call_gen.get_originator(contacts, *args, **kwargs)

Originator factory


Yield slave pairs up until a slave has reached a number of calls less then or equal to it’s predefined capacity limit

Measurement Collection

CDR app for collecting signalling latency and performance stats.

class switchio.apps.measure.cdr.CDR

Collect call detail record info including call oriented event time stamps and and active sessions data which can be used for per call metrics computations.

log_stats(sess, job)

Append measurement data only once per call


Store total (cluster) session count at channel create time


Default call measurements computed from data retrieved by the CDR app.

Rudimentary system stats collection using psutil.

class switchio.apps.measure.sys.SysStats(psutil, rpyc=None)

A switchio app for capturing system performance stats during load test using the psutil module.

An instance of this app should be loaded if rate limited data gathering is to be shared across multiple slaves (threads).


Reindex on the call index to allign with call metrics data and interpolate.

Media testing

Common testing call flows

class switchio.apps.players.MutedPlayRec

On stop either swap the mute state of the channels if more iterations are required or hangup the call. If the current call is being recorded schedule the recordings to stop and expect downstream callbacks to schedule call teardown.

  1. Mute the peer session
  2. Trigger endless clip playback on both sessions
class switchio.apps.players.PlayRec

Play a recording to the callee and record it onto the local file system

This app can be used in tandem with MOS scoring to verify audio quality. The filename provided must exist in the FreeSWITCH sounds directory such that ${FS_CONFIG_ROOT}/${sound_prefix}/<category>/<filename> points to a valid wave file.


On stop either trigger a new playing of the signal if more iterations are required or hangup the call. If the current call is being recorded schedule the recordings to stop and expect downstream callbacks to schedule call teardown.


Trigger clip playback on the given session by doing the following: - Start playing a silence stream on the peer session - This will in turn trigger a speech playback on this session in the “PLAYBACK_START” callback

class switchio.apps.players.RecInfo(host, caller, callee)

Alias for field number 2


Alias for field number 1


Alias for field number 0

class switchio.apps.players.TonePlay

Play a ‘milli-watt’ tone on the outbound leg and echo it back on the inbound

Dtmf tools

class switchio.apps.dtmf.DtmfChecker

Play dtmf tones as defined by the iterable attr sequence with tone duration. Verify the rx sequence matches what was transmitted.

For each session which is answered start a sequence check. For any session that fails digit matching store it locally in the failed attribute.

Bert testing

class switchio.apps.bert.Bert

Call application which runs the bert test application on both legs of a call

See the docs for mod_bert and discussion by the author here.


Toggle whether to hangup calls when a bert test fails


Increment counters on synchronization failure

The following stats can be retrieved using the latest version of mod_bert:

sync_lost_percent - Error percentage within the analysis window sync_lost_count - How many times sync has been lost cng_count - Counter of comfort noise packets err_samples - Number of samples that did not match the sequence

Knows how to get us riled up


Mark session as bert time out


Toggle whether to run the bert_test application on all sessions of the call. Leaving this False means all other legs will simply run the echo application.