Record Dispatch in Python

In this post we are going to dispatch Z21 records in Python. In a previous post I showed how to connect and read records from a Roco Z21 DCC command station in Python. After that I gained new insights into the quality of code. Based on this, I refactored a function for extracting individual records from a Z21 packet. Now, we are going to dispatch these records.

Deep dive in code
Deep dive in code

Let’s go back to the original feature: process records from incoming Z21 packets. We’ll go deeper into the user story “Dispatch extracted records.” For now we won’t go into the actual handling of the different record types. We will only focus on looping through the records from the incoming packet (in a record array) and dispatching them depending on the record type.

The user story can be refined as something like this:

IN ORDER TO process all records from a packet

AS a Z21 client

I WANT to dispatch each record to the appropriate handler.

Note that I’ll simplify a few things in this post to concentrate on the coding decisions. I’ll leave the exact binary nitty-gritty to when I’ll actually put together all these functions.

Dispatching – preparation

In this example we’ll dispatch two types of records: record A and record B. In practice, the Z21 can send many more record types. We’ll keep that in mind when writing a dispatcher. In this example, the first character of the record signifies the record type.


For now, we’ll have two handler functions (for A and B) that will simply return a value pair containing the name of the function and the content of the provided record. This way we can easily test that the dispatcher works correctly. We’ll develop two dummy functions for now. In the future when we will add logic to them, we will require that they continue to return the same value pairs so that we will be able to test them later as well. These are the functions:

def handleA(record):
    return "handleA", record
def handleB(record):
    return "handleB", record

Now let’s write some scenarios so that we can create automated tests for the dispatcher later:

GIVEN a Z21 record

Scenario 1: record is of type A

WHEN the record is “Abc”

THEN the result shall be “handleA”, “Abc”

Scenario 2: record is of type B

WHEN the record is “Bbc”

THEN the result shall be “handleB”, “Bbc”

Scenario 3: record is empty

WHEN the record is “”

THEN the result shall be “”

Scenario 4: record is of unknown type

WHEN the record is “Cbc”

THEN the result shall be “error”

Now, let’s turn these scenarios into a test function, in the true TDD spirit:

def dispatchTest():
    return \
        dispatch("") == "" and \
        dispatch("Abcde") == ("handleA", "Abcde") and \
        dispatch("Bbcde") == ("handleB", "Bbcde") and \
        dispatch("Cbcde") == "error"

Dispatch – implementation

I’ll present a number of different implementations for the dispatch function. They are all tested using the test function above. The first alternative is how I would typically start: a switch/case like structure. In Python the closest equivalent is an if/elif/else chain. It could look like this:

def dispatch(record):
    if len(record) == 0:
        return ""
    elif record[0] == "A":
        return handleA(record)
    elif record[0] == "B":
        return handleB(record)
        return "error"

This style is imperative. I find the code is ugly, especially if I expand the list of types. We can do better by using a ternary operator. This is what it could look like:

def dispatch(record):
    return \
        "" if len(record) == 0 else \
        handleA(record) if record[0] == "A" else \
        handleB(record) if record[0] == "B" else \

This looks more like a functional style. The code is much more concise even if seemingly the same components as in the previous implementation are used. Also if the list of record types will get longer, the code will not feel messier. However, I found a third alternative that is based on a dispatch table. This is such a function:

def dispatch(record):
    dispatchTable = {
        "A": handleA,
        "B": handleB
    return \
        "" if len(record) == 0 else \
        "error" if not record[0] in dispatchTable else \

What we see here is that the record types and the handlers for these types are listed in a dictionary. Then based on the record type, the associated function is looked up from the dispatch table and called. This approach has a few advantages: a longer list seems to have no negative effect on the performance. Also if more handlers need to be added, one only has to add the mapping to the dispatch table. In other words, less maintenance.

This sounds very good, but I’m impartial to the fact that the data becomes the code. While this reduces maintenance it can become messy if the dispatch table would be provided as a parameter. As long as the list is hardcoded in the function, I think the overview of the code is okay and risk for abuse is minimal.

Multidispatch – preparation

Now that we have a few working alternatives for the Dispatch function, let’s see how we can create a simple loop that performs this dispatch for each record in the record array. While this sounds so simple, again we’ll start with a few scenarios:

GIVEN an array of Z21 records

Scenario 1: the array is empty

WHEN the array is []

THEN the result shall be []

Scenario 2: the array contains 1 record

WHEN the array is [“Abc”]

THEN the result shall be [(“handleA”, “Abc”)]

Scenario 3: the array contains 2 records

WHEN the array is [“Abc”, “Abc”]

THEN the result shall be [(“handleA”, “Abc”), (“handleA”, “Abc”)]

No more variations in terms of different record types need to be tested, assuming that the Dispatch function was tested by itself.


The function for testing the Multidispatch function (assuming that the dispatch function is implemented as above and correct):

def multiDispatchTest():
    return \
        multiDispatch([]) == [] and \
        multiDispatch(["Abc"]) == [("handleA", "Abc")] and \
        multiDispatch(["Abc", "Abc"]) == [("handleA", "Abc"), ("handleA", "Abc")]

Multidispatch – implementation

Also for Multidispatch I found a number of implementations. The traditional implementation would look like this:

def multiDispatch_Traditional(recordArray):
    result = []
    for record in recordArray:
    return result

Again, this is an imperative approach. The code would look rather harmless if I wouldn’t want to collect the results. Now things start to look messy. Also, continuously updating the array containing the results is a no-go.

Python has a concept called list comprehension. This can create new lists based on existing lists using a transformation function. This is an implementation of Multidispatch using list comprehension:

def multiDispatch(recordArray):
    return [dispatch(record) for record in recordArray]

This implementation is certainly much cleaner. It is more concise and the only mutated variable, record, is contained in the for-statement. This is considered ‘Pythonesque,’ which means that this seems to be the preferred implementation amongst Python developers.

Then, last but not least, a pure functional implementation. It is based on the map function that applies a function (the first parameter) on an iterable (the second parameter). The list function wraps the result from the map function into a new list. The implementation below uses the functional map function.

def multiDispatch(recordArray):
    return list(map(dispatch, recordArray))


In this post, together with the previous post, we created all functions needed for picking up a Z21 packet, breaking it down into an array of records and dispatch each record to the appropriate handler function. We specified the requirements in terms of features and user stories. We added testable scenarios and created test functions.

These automated test functions were a great help to research alternative implementations for the different steps. We evaluated traditional (at least for me), imperative implementations, more Pythonesque and more functional implementations. It became clear for me that the traditional approach was certainly far from the best implementation. I found it valuable to investigate a bit deeper if alternative approaches could be better.

The specifications and implementations were slightly simplified compared to the actual Z21 specification. In a next post I’ll put the best implementations together into a fully functional feature. I’m not sure which is the ‘best’ implementation for each step, though. I’d be happy to hear what your thoughts are about this. Maybe you have even better ideas?