This is code for a measurement setup that receives a steady stream of UDP data, finds the trigger in one channel and operates on the data in the other channel to enhance the signal and remove noise, and display the enhanced signal.

I can see this project growing, and therefor I would like a more robust design and clean control flow and code. There are a lot of global variables there, mostly because they need to be persistent. Also, initialisation needs to happen early on, like in case of the updateable matplotlib plot. Object-orientation could help with the percistency, but I don't feel confident about OO. What would the objects and methods be?

Are the variable names helpful? Which ones need improvement?

I consider to go multi-process, with parts like reading from the socket, graphics and data evaluation all in seperate processes, and passing data between them in queues. However that makes things complicated quickly. What other options do I have?

#!/usr/bin/env python3

import collections
import socket
from struct import calcsize, unpack_from
import numpy as np
import argparse
import matplotlib.pyplot as plt

# Todo: no global variables (?)
# Todo: show several old signals, in a faint, transparent way, in the signal graph
# Todo: show trigger graph and signal graph over the whole time window in subplots (to make sure the 

COUNT = '<H'
DATA = '<ddd'  # contains the data of x, y, and z axis

commandline_parser = argparse.ArgumentParser(description='Receive the data, which the Sensys MX3DUW sends.')
commandline_parser.add_argument('port_num', metavar='P', type=int, help='the port number to listen on for the data')
args = commandline_parser.parse_args()

previousTimestamp: int = 0
rawData = collections.defaultdict(list)

server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.bind(('', args.port_num))

# one udp frame holds 8 samples
# each sample holds one time stamp and five sensor readings
# each sensor has three axis

cycle_length = int(2000 / 8) * 8
data = np.zeros((4, cycle_length * 2, 3))
signal_len = 60  # in samples

signal = np.zeros(signal_len)
noise = np.zeros(signal_len)

# prepare plot
x = np.arange(signal_len)
fig = plt.figure()
ax = fig.add_subplot(111)
plt.ylim(-15, 15)
plt.title("signal minus noise over many cycles")
line1, = ax.plot(x, signal, 'b-')  # Returns a tuple of line objects, thus the comma

sensor_num = 4
axis_num = 3
data_line = 0

correction = 0
measurement_cnt = 0
#old_flank = 0

def elvaluate_data(data):
    # these variables could just as well be local, if they were persistent!
    global measurement_cnt, signal, noise

    rising_flanks = find_rising_flanks(data)

    for flank in rising_flanks:
        # make sure the rising flank is not too close to the edge
        if (flank > 200) and (flank < (cycle_length - 200)):
            useful_flank = flank
            measurement_cnt += 1
            signal = signal + data[2, (useful_flank + 60):(useful_flank + 60 + signal_len), 2]
            noise = data[2, useful_flank - signal_len - 20:useful_flank - 20, 2]
            signal = (signal - noise) / measurement_cnt


def update_plot(signal):
    #    line2.set_ydata(data[2, :, 2])
    plt.ylim(np.min(signal), np.max(signal))

def stabilize_measurement_window(useful_flank):
    global correction, cycle_length
    correction = int((1000 - useful_flank) / 2)
    cycle_length = 2000 - correction
    # print("current cycle_length:", cycle_length, "useful_flank", useful_flank, "correction:", correction,
    #       "accelleration: ", old_flank - useful_flank)
    #            old_flank = useful_flank

def find_rising_flanks(data):
    mask = (np.abs(data[1, :, 2]) > 11.0)
    window = 10
    # count the number of times the value is below thresh in the window
    below_thresh = np.sum([mask[i:len(mask) - window + i] for i in range(window)], axis=0)
    idx_mask = below_thresh == window
    rising_flanks = np.where(idx_mask[1:] & (~idx_mask[:-1]))[0] + window + 1
    return rising_flanks

def read_data(offset, message):
    for sensor_cnt in range(0, sensor_num):
        data[sensor_cnt][data_line] = unpack_from(DATA, message, offset)
        offset += calcsize(DATA)
    offset += calcsize(DATA)  # skip the last last sensor, it is not connected
    return offset

signal = np.zeros(signal_len)
    while True:
        message, (sender_ip, sender_port) = server_socket.recvfrom(65507)  # this is 65535-28 == 65507
        # first wait till data is coming, then stop recording once it stops coming in
        # 28 is size of IP + UDP header
        offset = 0
        [sample_num] = unpack_from(COUNT, message, offset)
        offset += calcsize(COUNT)
        for sample_cnt in range(0, sample_num):
            sensor_config, timestamp = unpack_from(HEADER, message, offset)
            offset += calcsize(HEADER)
            stepSize = timestamp - previousTimestamp
            if stepSize > 500:
                # detect if we dropped packages. if we did, we might want to flush the current cycle
                print("Stepsize: " + str(stepSize) + " at time " + str(timestamp))
            previousTimestamp = timestamp
            offset = read_data(offset, message)
            data_line += 1
            if data_line == cycle_length:
                data_line = 0

except socket.timeout:
    print('\n no more data from measurement system')


elvaluate -> evaluate


They're the enemy of testing and re-entrant modules. Everything from commandline_parser through line1, =, and everything for signal = onward, should be moved into functions. The variables starting at sensor_num onward should be capitalized and kept as global constants.

The two "easy" ways to transfer global state into a re-entrant place are either conversion to an object where the state exists as members, or convert everything into function parameters and return values, passing around as necessary.

Define more global constants for the magic numbers within these statements:

cycle_length = int(2000 / 8) * 8

    if (flank > 200) and (flank < (cycle_length - 200)):

correction = int((1000 - useful_flank) / 2)

cycle_length = 2000 - correction

mask = (np.abs(data[1, :, 2]) > 11.0)


        if stepSize > 500:

Resource context management

rather than

except socket.timeout:
    print('\n no more data from measurement system')

Your socket should be closed unconditionally, and in a finally. Better yet, put the

server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

in a with, for example

def some_upper_server_loop():
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server_socket:
        server_socket.bind(('', args.port_num))
        # ...

Unit tests

Add them. Add them while it's still easy and your project is small. You claim that

Unit tests work best with object oriented code

but - happily - this is not whatsoever the case. Often, well-decoupled, re-entrant procedural code is actually easier to unit test than OO code because the incoming state can be more narrow and well-defined as parameters than as class member variables.

Picking on stabilize_measurement_window for a moment, if it's repaired so that correction is a parameter and cycle_length is a return value:

def stabilize_measurement_window(useful_flank: float, correction: float) -> int:
    correction = int((1000 - useful_flank) / 2)
    cycle_length = 2000 - correction
    return cycle_length

# ...

assert stabilize_measurement_window(500, 1) == 1749
  • \$\begingroup\$ After reading your answer, i looked for the with and found stackoverflow.com/questions/55661915/… but it does not show how to close the socket like that, or how the recvfrom-loop would look. Could you add a resource that showcases that? \$\endgroup\$ Feb 2 at 14:58
  • \$\begingroup\$ Unit tests work best with object oriented code. But you didnt suggest to go OO. Do you have a link for doing tests for my kind of code? \$\endgroup\$ Feb 2 at 14:59
  • \$\begingroup\$ You dont touch on the persistence aspect of the variables. Can you please explain how to achive that without global variables? i googled and found either OO with instance variables or global variables, perhaps with a wrapper function to hide the variable itself. @reinderien \$\endgroup\$ Feb 2 at 15:05
  • \$\begingroup\$ I've made an edit that attempts to answer these three things. \$\endgroup\$
    – Reinderien
    Feb 2 at 15:47
  • \$\begingroup\$ For extended discussion let's switch to chat.stackexchange.com/rooms/119252/… \$\endgroup\$
    – Reinderien
    Feb 2 at 15:48

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.