helpers (Helpers)

Author:

Description:

This module supports functions and classes which don’t have an own Module (yet).

Submodules:

Unittest:

See also the unittest documentation.

Module Documentation:

class helpers.continues_statistic(mean=None, min_val=None, max_val=None, quantifier=0)

This class stores a statistic for a stream of values. The statistic includes: mean, min, max, quantifier.

Note

You can use the mathematic operators “+, +=” to add a value to the statistic.

Parameters:

  • mean (numeric) – The mean start value, the start value or None.

  • min_val (numeric) – The min start value or None

  • max_val (numeric) – The max start value or None

  • quantifier (int) – The quantifier start value or 0

Note

You are able to initialise this class in the following ways:

  • Without arguments to get an empty instance

  • With one single value to get an starting instance

  • With mean, min_val, max_val, quantifier to initialise an existing statistic

Example:

import sys
sys.path.append('../..')

import helpers


cs = helpers.continues_statistic()
for val in (1, 3, 5, 4):
    cs += helpers.continues_statistic(val)
    print(cs)

Will result to the following output:

mean=1, min=1, max=1, quantifier=1
mean=2.0, min=1, max=3, quantifier=2
mean=3.0, min=1, max=5, quantifier=3
mean=3.25, min=1, max=5, quantifier=4
property max

The current max value.

property mean

The current mean value.

property min

The current min value.

pop()

This pops out the current statistic data and returns these as an instance of helpers.continues_statistic.

Returns:

The current statistic or None, if no value is available.

Return type:

helpers.continues_statistic or None

property quantifier

The current quantifier (number of values for continues statistic).

class helpers.continues_statistic_multivalue(*args, **kwargs)

This class stores multiple statistics of stream values. The statistic of each value is stored as helpers.continues_statistic.

Note

You can use the mathematic operators “+, +=” to add instances.

Note

You are able to initialise this class in the following ways:

Example:

import sys
sys.path.append('../..')

import helpers


csm = helpers.continues_statistic_multivalue()
print(csm, '\n--------------------------------------------------------------')
csm += helpers.continues_statistic_multivalue(a={'mean': 3.1, 'min_val': 1, 'max_val': 3.5, 'quantifier': 305})
print(csm, '\n--------------------------------------------------------------')
csm += helpers.continues_statistic_multivalue(a=4.31, b={'mean': 17.1, 'min_val': 13.1, 'max_val': 19.3, 'quantifier': 27})
print(csm, '\n--------------------------------------------------------------')
print(helpers.continues_statistic_multivalue(dict(csm)), '\n--------------------------------------------------------------')
print('pop out b:', csm.pop('b'), '\n--------------------------------------------------------------')
print('pop out b:', csm.pop('b'), '\n--------------------------------------------------------------')
print(csm, '\n--------------------------------------------------------------')
print('pop out all:\n%s' % csm.pop(), '\n--------------------------------------------------------------')
print('pop out all:\n%s' % csm.pop(), '\n--------------------------------------------------------------')
print(csm, '\n--------------------------------------------------------------')

Will result to the following output:

- 
--------------------------------------------------------------
a: mean=3.1, min=1, max=3.5, quantifier=305 
--------------------------------------------------------------
a: mean=3.103954248366013, min=1, max=4.31, quantifier=306
b: mean=17.1, min=13.1, max=19.3, quantifier=27 
--------------------------------------------------------------
a: mean=3.103954248366013, min=1, max=4.31, quantifier=306
b: mean=17.1, min=13.1, max=19.3, quantifier=27 
--------------------------------------------------------------
pop out b: mean=17.1, min=13.1, max=19.3, quantifier=27 
--------------------------------------------------------------
pop out b: mean=None, min=None, max=None, quantifier=0 
--------------------------------------------------------------
a: mean=3.103954248366013, min=1, max=4.31, quantifier=306
b: mean=None, min=None, max=None, quantifier=0 
--------------------------------------------------------------
pop out all:
a: mean=3.103954248366013, min=1, max=4.31, quantifier=306
b: mean=None, min=None, max=None, quantifier=0 
--------------------------------------------------------------
pop out all:
- 
--------------------------------------------------------------
- 
--------------------------------------------------------------
pop(key=None)

This pops out the current statistic data for a given key and returns these as an instance of helpers.continues_statistic.

If no key is given it pops out the current statistic data and returns these as an instance of helpers.continues_statistic_multiple.

Parameters:

key (str or NoneType) – The key to get data for

Returns:

The current statistic or None, if no value is available.

Return type:

helpers.continues_statistic or None

class helpers.direct_socket_base(max_len=None, virtual_rate_bps=None)

This is the base class for other classes in this module.

disconnect()

Method to disconnect client and server.

init_channel_name(channel_name=None)

With this Method, the channel name for logging can be changed.

Parameters:

channel_name (str) – The name for the logging channel

is_connected()

With this Method the connection status can be identified.

Returns:

True, if a connection is established, otherwise False.

Return type:

bool

receive(timeout=1.0, num=None)

This method returns received data.

Parameters:

  • timeout (float) – The timeout for receiving data (at least after the timeout the method returns data or None).

  • num (int or None) – the number of bytes to receive (use None to get all available data).

Returns:

The received data.

Return type:

bytes

register_callback(callback)

This method stores the callback which is executed, if data is available. You need to execute receive() of this instance given as first argument.

Parameters:

callback – The callback which will be executed, when data is available.

register_connect_callback(callback)

This method stores the callback which is executed, if a connection is established.

Parameters:

callback – The callback which will be executed, when a connection is established.

register_disconnect_callback(callback)

This method stores the callback which is executed, after the connection is lost.

Parameters:

callback – The callback which will be executed, after the connection is lost.

send(data, timeout=1.0)

This method sends data via the initiated communication channel.

Parameters:

  • data (bytes) – The data to be send over the communication channel.

  • timeout (float) – The timeout for sending data (e.g. time to establish new connection).

Returns:

True if data had been sent, otherwise False.

Return type:

bool

class helpers.direct_socket_client(max_len=None, virtual_rate_bps=None)

Class to create a direct client socket. See also parent helpers.direct_socket_base.

Example:

import sys
sys.path.append('../..')

import report
import helpers
import time


def mirror_callback(sock):
    sock.send(sock.receive())


report.stdoutLoggingConfigure(log_name_lvl=[('root', 'DEBUG'), ])
s = helpers.direct_socket_server(virtual_rate_bps=5)
s.init_channel_name('example_server')
s.register_callback(mirror_callback)
c = helpers.direct_socket_client(max_len=2, virtual_rate_bps=5)
c.init_channel_name('example_client')
c.connect(s)
c.send(b'abc')
print('The Client received: %s' % repr(c.receive()))

Will result to the following output:

2024-12-09 00:00:42,473:    DEBUG - helpers.all_others - comm-server: Cleaning up receive-buffer
2024-12-09 00:00:42,473:     INFO - helpers.all_others - comm-server: Waiting for incomming connection
2024-12-09 00:00:42,474:    DEBUG - helpers.all_others - comm-client: Cleaning up receive-buffer
2024-12-09 00:00:42,474:     INFO - helpers.example_client - comm-client: Connection established...
2024-12-09 00:00:42,474:    DEBUG - helpers.example_client - comm-client: Cleaning up receive-buffer
2024-12-09 00:00:42,474:     INFO - helpers.example_server - comm-server: Connection established...
2024-12-09 00:00:42,474:    DEBUG - helpers.example_server - comm-server: Cleaning up receive-buffer
2024-12-09 00:00:42,474:     INFO - helpers.example_client - comm-client: TX -> (2): 61 62
2024-12-09 00:00:42,875:     INFO - helpers.example_server - comm-server: RX <- (2): 61 62
2024-12-09 00:00:42,875:     INFO - helpers.example_client - comm-client: TX -> (1): 63
2024-12-09 00:00:42,875:     INFO - helpers.example_server - comm-server: TX -> (2): 61 62
2024-12-09 00:00:43,076:     INFO - helpers.example_server - comm-server: RX <- (1): 63
2024-12-09 00:00:43,276:     INFO - helpers.example_client - comm-client: RX <- (2): 61 62
2024-12-09 00:00:43,276:     INFO - helpers.example_server - comm-server: TX -> (1): 63
The Client received: b'ab'
connect(remote_socket)

Method to create a connection between this client and a helpers.direct_socket_server instance.

Parameters:

remote_socket (helpers.direct_socket_server) – The remote socket to connect to.

reconnect()

Method to do a reconnect.

Note

The remote_socket of the prefious connect() call will be used.

class helpers.direct_socket_server(*args, **kwargs)

Class to create a direct server socket. See also parent helpers.direct_socket_base.

Example:

import sys
sys.path.append('../..')

import report
import helpers
import time


def mirror_callback(sock):
    sock.send(sock.receive())


report.stdoutLoggingConfigure(log_name_lvl=[('root', 'DEBUG'), ])
s = helpers.direct_socket_server(virtual_rate_bps=5)
s.init_channel_name('example_server')
s.register_callback(mirror_callback)
c = helpers.direct_socket_client(max_len=2, virtual_rate_bps=5)
c.init_channel_name('example_client')
c.connect(s)
c.send(b'abc')
print('The Client received: %s' % repr(c.receive()))

Will result to the following output:

2024-12-09 00:00:42,473:    DEBUG - helpers.all_others - comm-server: Cleaning up receive-buffer
2024-12-09 00:00:42,473:     INFO - helpers.all_others - comm-server: Waiting for incomming connection
2024-12-09 00:00:42,474:    DEBUG - helpers.all_others - comm-client: Cleaning up receive-buffer
2024-12-09 00:00:42,474:     INFO - helpers.example_client - comm-client: Connection established...
2024-12-09 00:00:42,474:    DEBUG - helpers.example_client - comm-client: Cleaning up receive-buffer
2024-12-09 00:00:42,474:     INFO - helpers.example_server - comm-server: Connection established...
2024-12-09 00:00:42,474:    DEBUG - helpers.example_server - comm-server: Cleaning up receive-buffer
2024-12-09 00:00:42,474:     INFO - helpers.example_client - comm-client: TX -> (2): 61 62
2024-12-09 00:00:42,875:     INFO - helpers.example_server - comm-server: RX <- (2): 61 62
2024-12-09 00:00:42,875:     INFO - helpers.example_client - comm-client: TX -> (1): 63
2024-12-09 00:00:42,875:     INFO - helpers.example_server - comm-server: TX -> (2): 61 62
2024-12-09 00:00:43,076:     INFO - helpers.example_server - comm-server: RX <- (1): 63
2024-12-09 00:00:43,276:     INFO - helpers.example_client - comm-client: RX <- (2): 61 62
2024-12-09 00:00:43,276:     INFO - helpers.example_server - comm-server: TX -> (1): 63
The Client received: b'ab'
class helpers.direct_socket_stp_base(*args, **kwargs)
receive(timeout=1)

This method returns one received messages via the initiated communication channel.

Parameters:

timeout (float) – The timeout for receiving data (at least after the timeout the method returns data or None).

Returns:

The received data.

Return type:

bytes

class helpers.direct_socket_stp_client(*args, **kwargs)
connect(remote_socket)

Method to create a connection between this client and a helpers.direct_socket_server instance.

Parameters:

remote_socket (helpers.direct_socket_server) – The remote socket to connect to.

reconnect()

Method to do a reconnect.

Note

The remote_socket of the prefious connect() call will be used.

class helpers.direct_socket_stp_server(*args, **kwargs)
class helpers.ringbuffer(*args, **kwargs)

Class for a list with a limited number of elements.

Note

On adding Objects, the list will be reduced to the maximum length by deleting entries starting from index 0.

Example:

import sys
sys.path.append('../..')

import helpers


rb = helpers.ringbuffer(length=5)
print(rb, '\n--------------------------------------------------------------')
rb.append(0)
print(rb, '\n--------------------------------------------------------------')
rb.extend((1,2,3,4,5,6))
print(rb, '\n--------------------------------------------------------------')
rb.append(17)
print(rb, '\n--------------------------------------------------------------')

Will result to the following output:

[] 
--------------------------------------------------------------
[0] 
--------------------------------------------------------------
[2, 3, 4, 5, 6] 
--------------------------------------------------------------
[3, 4, 5, 6, 17] 
--------------------------------------------------------------
append(obj)

Append object to the end of the list.

extend(iterable)

Extend list by appending elements from the iterable.