19
\$\begingroup\$

I wrote an application a while back to plan and manage drones races. (Well, it could be anything with identification numbers going through checkpoints.) At that time, communication between the gates (checkpoints) and the device running this application was done through XBee radios. When a gate detected a drone, it sent an XBee dataframe containing the gate identification letter and the number ID of the detected drone.

The application was responsible of reading and interpretting these XBee messages. It was done through the xbee module. I also included an stdin reader for debug/testing purposes.

For a new event, it was decided to switch from XBee communication to WiFi. So I was asked to handle messages through UDP datagrams. I would like to put the data readers and their selection at startup to review. The whole code is being accessible on GitHub and some other parts might end up as new questions in a near future.

droneracer.py

import os

from argparse import ArgumentParser
import drone_racer


# Be sure to be at the right place for relative path of images in Gtk
os.chdir(os.path.dirname(os.path.abspath(__file__)))

parser = ArgumentParser(description='Interface graphique "Drone Racer"')

# GUI args
parser.add_argument('--fancy-title', dest='fancy', action='store_true',
                    help='Utilise une barre de titre un peu plus Gtk3')

# XBee args
parser.add_argument('--serial-port', dest='serial', metavar='FILE',
                    default=None, help='Spécifie le port série à utiliser '
                    'pour récupérer les informations provenant du XBee')
parser.add_argument('--zigbee', dest='zigbee', action='store_true',
                    help='Spécifie si le module XBee est un ZigBee')
parser.add_argument('--baudrate', dest='baudrate', metavar='BPS',
                    type=int, default=9600, help='Débit du port série '
                    'utilisé pour la connexion avec le module XBee')

# UDP args
parser.add_argument('--use-udp', dest='udp', action='store_true',
                    help='Spécifie si la communication doit se faire '
                    'par datagrames UDP.')
parser.add_argument('--port', dest='port', metavar='NUM', type=int,
                    default=4387, help='Port à utiliser pour l’écoute UDP')

# Choose the appropriate reader
args = parser.parse_args()
if args.serial is not None:
    reader = drone_racer.XBeeReader(
            args.serial, args.baudrate, zigbee=args.zigbee)
elif args.udp:
    reader = drone_racer.UDPReader(args.port)
else:
    reader = drone_racer.StdInReader

# Launch the GUI (which will, in turn, start the reader)
app = drone_racer.Application(reader, args.fancy)
app.run()

At some point in drone_racer.Application, reader will be instanciated with a callback function: reader(self.console.compute_data).

drone_racer/__init__.py

from .ui import DroneRacer as Application
from .threads import StdInReader, XBeeReader, UDPReader

drone_racer/threads.py

import os
import sys
import socket
from threading import Thread
from select import select
try:
    from serial import Serial
    from xbee import XBee, ZigBee
except ImportError:
    XBee = None


class BaseReader(Thread):
    """Base class for custom data readers."""

    def __init__(self, update_function):
        """Spawn a thread that continuously read data for drones statuses.

        Parameter:
          - update_function: the function that will be called each time a
            valid data is read.
        """
        super().__init__(name="reader")
        self._update_data = update_function
        self._should_continue = True
        self.start()

    def run(self):
        """The main action of the thread.

        Wait for data, read them and send them to the rest of the application
        for further computation.
        """
        while self._should_continue:
            try:
                gate, drone = self.read_new_value()
            except TypeError:
                pass
            else:
                self._process_value(gate, drone)

    def stop(self):
        """Signal that the thread has to stop reading its inputs."""
        self._should_continue = False

    def read_new_value(self):
        """Read input data and return them as a tuple (gate identifier, drone
        number). Subclasses must implement this method.
        """
        raise NotImplementedError("Subclasses must implement this method")

    def _process_value(self, gate, drone):
        """Send input data to the rest of the application.

        Parameters:
          - gate: the gate identification letter(s)
          - drone: the drone identification number (0-based)
        """
        if drone < 0:
            return
        self._update_data(gate, drone)


class StdInReader(BaseReader):
    """Read data from stdin. Primarily used for tests and debug."""

    def read_new_value(self):
        """Read input data and return them as a tuple (gate identifier,
        drone number).

        Convert data such as "0 1" to the tuple ('A', 1).
        """
        raw = input('[@] ').split()
        try:
            gate, drone = raw
            return chr(int(gate) + ord('A')), int(drone)
        except ValueError:
            pass


class _UDPReader(BaseReader):
    """Read data from UDP datagrams. Used when communicating via WiFi
    with the gates.
    """

    def __init__(self, iface, port, update_function):
        """Spawn a thread that continuously read data for drones statuses.

        Parameter:
          - iface: the address of the interface to listen on.
          - port: the socket port to listen on.
          - update_function: the function that will be called each time a
            valid data is read.
        """
        com = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        com.bind((iface, port))
        self._socket = [com]
        super().__init__(update_function)

    def read_new_value(self):
        """Read input data and return them as a tuple (gate identifier,
        drone number).
        Decode an UDP datagram containing b"C:3" to the tuple ('C', 2).
        """
        # Non-blocking read so this thread will shut down with the application
        ready, _, _ = select(self._socket, [], [], 0)
        for socket in ready:
            msg = socket.recv(128) # Way too much for messages like <A:1>
            try:
                gate, drone = msg.split(b':')
                gate = gate.decode()
                # Compensate for the drone numbering vs. its indexing
                drone = int(drone) - 1
            except (UnicodeError, ValueError) as e:
                print('Le message', msg, 'a été reçu mais n’est pas'
                      'compris par l’application.', file=sys.stderr)
                print(e, file=sys.stderr)
            else:
                return gate, drone


class UDPReader:
    """Factory of _UDPReaders."""

    def __init__(self, port):
        """Save parameters for future use.

        Parameter:
          - port: the socket port to listen on.
        """
        self._port = port

    def __call__(self, callback):
        """Generate the appropriate class to read data.

        Parameter:
          - callback: the function that will be called each
            time a valid data is read.
        """
        return _UDPReader(socket.gethostname(), self._port, callback)


if XBee is None:
    class _BeeReader(BaseReader):
        """Read data from a serial port bound to an XBee.
        Dummy implementation because xbee module could not be loaded.
        """

        def read_new_value(self):
            """Cancel this thread to avoid burning resources."""
            self._should_continue = False

    def XBeeReader(*args, **kwargs):
        """Wrapper around the xbee module to integrate our _BeeReaderMixin
        into the appropriate base class.
        Dummy implementation because xbee module could not be loaded.
        """
        print('Le module XBee est instrouvable. Aucune donnée ne pourra',
              'être lue', file=sys.stderr)
        return _BeeReader
else:
    class _BeeReaderMixin:
        """Read data from a serial port bound to an XBee."""

        def __init__(self, serial, callback):
            """Initialize the XBee reader thanks to the mro.

            Parameters:
              - serial: the serial port object to read data from
              - callback: the function that will be called each
                time a valid data is read.
            """
            self._update_data = callback
            super().__init__(serial, callback=self._process_value)

        def _process_value(self, response_dict):
            """Convert a raw data received in a frame by the XBee
            into suitable data for the application.

            Should be called each time a frame is read by the XBee.
            """
            try:
                gate, drone = response_dict['rf_data'].split(b':')
                gate = gate.decode()
                # Compensate for the drone numbering vs. its indexing
                drone = int(drone) - 1
            except (UnicodeError, ValueError) as e:
                print('Le message', response_dict['rf_data'],
                      'a été reçu mais n’est pas compris par l’application.',
                      file=sys.stderr)
                print(e, file=sys.stderr)
            except KeyError as e:
                print('Un message ne contenant pas de données a été reçu.',
                      file=sys.stderr)
                print(e, file=sys.stderr)
            else:
                self._update_data(gate, drone)

        def stop(self):
            """Halt the thread from reading its input and close the
            underlying serial port.
            """
            self.halt()
            self.serial.close()

    class XBeeReader:
        """Wrapper around the xbee module to integrate our _BeeReaderMixin
        into the appropriate base class.
        """

        def __init__(self, *args, **kwargs):
            """Save parameters for future use.
            Everything is used to initialize a serial.Serial object
            except for the named attribute 'zigbee' which define the
            base class to use.

            Parameter:
              - zigbee: whether to use the xbee.ZigBee base class or
                the xbee.XBee one
            """
            zigbee = kwargs.pop('zigbee', False)
            self._args = args
            self._kwargs = kwargs
            self._base_cls = ZigBee if zigbee else XBee

        def __call__(self, callback):
            """Generate the appropriate class to read data.

            Parameter:
              - callback: the function that will be called each
                time a valid data is read.
            """
            serial = Serial(*self._args, **self._kwargs)
            self._args = None
            self._kwargs = None
            return type('XBeeReader', (_BeeReaderMixin, self._base_cls), {})(
                    serial, callback)

Expected invocation includes:

  • python droneracer.py --use-udp --port 6329
  • python droneracer.py --serial-port /dev/ttyUSB0 --baudrate 9600
  • python droneracer.py

I’m mostly concerned about these parts:

  • Is the arguments handling intuitive enough or can it be better managed?
  • Is the reader and associated factory a good design choice?
  • Is the newly implemented UDPReader following good practices for non-blocking sockets management (more timeout, maybe)?
\$\endgroup\$
  • \$\begingroup\$ Hey, could you please check the git repo, I'm unable to start the application due to various typo failures in the UDPReader classes. Especially AttributeError: '_UDPReader' object has no attribute '_watch' at line 110 which sounds really weird. \$\endgroup\$ – Cilyan Dec 28 '15 at 0:56
  • \$\begingroup\$ Ok, silly me, the source problem is self._socket.listen(5): Operation not supported, as UDP is stateless. You probably got your code from a TCP example. UDP is much simpler: wiki.python.org/moin/UdpCommunication#Receiving The error in the above comment is only produced while tearing down the app. \$\endgroup\$ – Cilyan Dec 28 '15 at 1:09
  • \$\begingroup\$ @Cilyan I started using only bind and recv. But then I saw an example using listen and accept as well and got confused. Well, time to learn the basics again I guess \$\endgroup\$ – 409_Conflict Dec 28 '15 at 7:15
  • \$\begingroup\$ @MathiasEttinger: I raised a PR on GitHub to start with \$\endgroup\$ – Cilyan Dec 28 '15 at 11:44
  • \$\begingroup\$ A follow-up of this question has been posted here. \$\endgroup\$ – 409_Conflict Jan 7 '16 at 22:12
3
+100
\$\begingroup\$

WRT intuitive CLI argument parsing, you might consider checking out the click library, if not for this project then for the next.

Your CLI code might look something like this:

import click

@click.command()
@click.option("--fancy-title", default=False, help="Utilise une barre de titre"
              " un peu plus Gtk3")
@click.option("--serial-port", type=type=click.Path(exists=True),
              help="Spécifie le port série à utiliser pour récupérer les "
              "informations provenant du XBee")
@click.option("--zigbee", default=False, help="Spécifie si le module XBee"
              " est un ZigBee")
@click.option("--baudrate", default=9600, help='Débit du port série '
              'utilisé pour la connexion avec le module XBee')
@click.option("--use-udp", default=False, help='Spécifie si la communication doit se faire '
              'par datagrames UDP.')
@click.option("--port", default=4387, help='ort à utiliser pour l’écoute UDP')
def cli(fancy_title, serial_port, zigbee, baudrate, use_udp, port):
    if serial_port is not None:
        reader = drone_racer.XBeeReader(serial_port, baudrate, zigbee=zigbee)
    elif use_udp:
        reader = drone_racer.UDPReader(port)
    else:
        reader = drone_racer.StdInReader

app = drone_racer.Application(reader, fancy_title)

(No, not an author, or even contributor, just a happy user)

Writing this, I noticed one other thing: the --use-X pattern is potentially inconsistent. Invoking

python droneracer.py --serial-port /dev/ttyUSB0 --baudrate 9600 --use-udp --port 1234

is possible, with half of the arguments ending up unused. The connection types might be better thought of as subcommands:

python droneracer.py serial_port /dev/ttyUSB0 9600
python droneracer.py udp 6329
python droneracer.py stdin

Which would look something more like:

import click

@click.command()
@click.option("--fancy-title", default=False, help="Utilise une barre de titre"
              " un peu plus Gtk3")
@click.pass_context
def cli(ctx, fancy_title):
    """Interface graphique "Drone Racer" """
    pass

@cli.result_callback(reader)
def cli_callback(reader, fancy_title)
    app = drone_racer.Application(reader, fancy_title)

@cli.group()
@click.argument("device", type=type=click.Path(exists=True),
              help="Spécifie le port série à utiliser pour récupérer les "
              "informations provenant du XBee", required=True)
@click.argument("baudrate", default=9600, help='Débit du port série '
              'utilisé pour la connexion avec le module XBee', required=False)
@click.option("--zigbee", default=False, help="Spécifie si le module XBee"
              " est un ZigBee")
def serial_port(device, baudrate, zigbee):
    return drone_racer.XBeeReader(device, baudrate, zigbee=zigbee)

@cli.group()
@click.argument("port", default=4387, required=False)
def udp(port):
    """Spécifie si la communication doit se faire par datagrames UDP."""
    return drone_racer.UDPReader(port)

@cli.group()
def stdin():
    """Specify the use of a serial port"""
    return drone_racer.StdInReader

As a second alternative, you could devise some URL-like specification and parse it. Consider a CLI like:

python droneracer.py udp:4387
python droneracer.py /dev/USB0:9600

It's not pretty and would involve custom parsing, but at least it's unambiguous and avoids the possibility of self-contradiction.

I think I prefer the subcommand architecture; the last suggestion would be okay if there were standard URL formats for the endpoints in question, but there aren't.

\$\endgroup\$
  • \$\begingroup\$ I do like the subcommand way or the urlpattern. The issue with potentially using both --use-udp and --serial-port was bothering me. I might give a try to click, but not for this project, due to other constraints. It feels interesting but weird at the same time, though. \$\endgroup\$ – 409_Conflict Jan 4 '16 at 21:52
3
\$\begingroup\$

Improve consistency

Duck-typing can be useful in some situation, but letting users that may want to provide their own reader for their own devices figure out the internals of your particular duck-typing is not very friendly.

Look at how you define your reader in droneracer.py:

if args.serial is not None:
    reader = drone_racer.XBeeReader(
            args.serial, args.baudrate, zigbee=args.zigbee)
elif args.udp:
    reader = drone_racer.UDPReader(args.port)
else:
    reader = drone_racer.StdInReader

Why construct an (factory) object in the first two cases and use the class object in the third one? How are we supposed to make the difference between the two?

You could, instead, have BaseReader objects callable (instead of initializable) with the callback function and start the thread at this moment only. It gives you more flexibility on how to initialize your objects and has two advantages:

  • classes that acts as both the thread and its initializer can be merged (so you can remove the UDPReader factory);
  • classes that need an external initializer (such as xbee.XBee or xbee.ZigBee) can use a factory and provide the exact same interface than BaseReader derived classes.

Improve documentation

PEP8 recommend to limit docstring length to 72 characters.

You document classes and methods but do not provide a docstring to populate the modules __doc__ from.

You could use the __all__ variable in drone_racer/__init__.py to improve the module's help. Classes that you include in this list will have their documentation merged to the one of the module when using help(drone_racer). It will also limit the amount of objects imported when/if using from drone_racer import *.

Rewriting

drone_racer/threads.py

"""Collection of classes to crete threaded objects allowing to read
data from various sources.

Readers should be created with whatever parameter they require and
then allow to be called with a callback function. This call return
the threaded object reading data.

These threaded objects are started immediatly and monitor incomming
data to normalize them before feeding them into the callback function.
They can easily be halted using their `stop` method.
"""


import os
import sys
import socket
from threading import Thread
from select import select
try:
    from serial import Serial
    from xbee import XBee, ZigBee
except ImportError:
    XBee = None


class BaseReader(Thread):
    """Base class for custom data readers."""

    def __init__(self):
        """Spawn a thread that will continuously read data for drones
        statuses.
        """
        super().__init__(name="reader")

    def __call__(self, update_function):
        """Starts the thread with the given callback function to
        process data with.

        Parameter:
          - update_function: the function that will be called each time
            a valid data is read.
        """
        self._update_data = update_function
        self._should_continue = True
        self.start()
        # Return ourselves to allow for duck typing and other classes
        # to return other kind of objects (see XBeeReader).
        return self

    def run(self):
        """The main action of the thread.

        Wait for data, read them and send them to the rest of the
        application for further computation.
        """
        while self._should_continue:
            try:
                gate, drone = self.read_new_value()
            except TypeError:
                pass
            else:
                self._process_value(gate, drone)

    def stop(self):
        """Signal that the thread has to stop reading its inputs."""
        self._should_continue = False

    def read_new_value(self):
        """Read input data and return them as a tuple (gate identifier,
        drone number). Subclasses must implement this method.
        """
        raise NotImplementedError("Subclasses must implement this method")

    def _process_value(self, gate, drone):
        """Send input data to the rest of the application.

        Parameters:
          - gate: the gate identification letter(s)
          - drone: the drone identification number (0-based)
        """
        if drone < 0:
            return
        self._update_data(gate, drone)


class StdInReader(BaseReader):
    """Read data from stdin. Primarily used for tests and debug."""

    def read_new_value(self):
        """Read input data and return them as a tuple (gate identifier,
        drone number).

        Convert data such as "0 1" to the tuple ('A', 1).
        """
        raw = input('[@] ').split()
        try:
            gate, drone = raw
            return chr(int(gate) + ord('A')), int(drone)
        except ValueError:
            pass


class UDPReader(BaseReader):
    """Read data from UDP datagrams. Used when communicating via
    WiFi with the gates.
    """

    def __init__(self, port):
        """Spawn a thread that continuously read data for drones
        statuses.

        Parameter:
          - port: the socket port to listen on.
        """
        super().__init__()
        com = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        iface = socket.gethostname()
        com.bind((iface, port))
        self._socket = [com]

    def read_new_value(self):
        """Read input data and return them as a tuple (gate identifier,
        drone number).

        Decode an UDP datagram containing b"C:3" to the tuple ('C', 2).
        """
        # Non-blocking read so this thread will shut down with the application
        ready, _, _ = select(self._socket, [], [], 0.05)
        for socket in ready:
            msg = socket.recv(128) # Way too much for messages like <A:1>
            try:
                gate, drone = msg.split(b':')
                gate = gate.decode()
                # Compensate for the drone numbering vs. its indexing
                drone = int(drone) - 1
            except (UnicodeError, ValueError) as e:
                print('Le message', msg, 'a été reçu mais n’est pas'
                      'compris par l’application.', file=sys.stderr)
                print(e, file=sys.stderr)
            else:
                return gate, drone


if XBee is None:
    class XBeeReader(BaseReader):
        """Read data from a serial port bound to an XBee.
        Dummy implementation because xbee module could not be loaded.
        """

        def __init__(self, *args, **kwargs):
            """Accepts arguments to be compatible with the "real"
            XBeeReader but prints a warning and terminate gracefully
            instead.
            """
            super().__init__()
            print('Le module XBee est instrouvable. Aucune donnée ne pourra',
                  'être lue', file=sys.stderr)

        def read_new_value(self):
            """Cancel this thread to avoid burning resources."""
            self._should_continue = False

else:
    class _BeeReaderMixin:
        """Read data from a serial port bound to an XBee."""

        def __init__(self, serial, callback):
            """Initialize the XBee reader thanks to the mro.

            Parameters:
              - serial: the serial port object to read data from
              - callback: the function that will be called each
                time a valid data is read.
            """
            self._update_data = callback
            super().__init__(serial, callback=self._process_value)

        def _process_value(self, response_dict):
            """Convert a raw data received in a frame by the XBee
            into suitable data for the application.

            Should be called each time a frame is read by the XBee.
            """
            try:
                gate, drone = response_dict['rf_data'].split(b':')
                gate = gate.decode()
                # Compensate for the drone numbering vs. its indexing
                drone = int(drone) - 1
            except (UnicodeError, ValueError) as e:
                print('Le message', response_dict['rf_data'],
                      'a été reçu mais n’est pas compris par l’application.',
                      file=sys.stderr)
                print(e, file=sys.stderr)
            except KeyError as e:
                print('Un message ne contenant pas de données a été reçu.',
                      file=sys.stderr)
                print(e, file=sys.stderr)
            else:
                self._update_data(gate, drone)

        def stop(self):
            """Halt the thread from reading its input and close the
            underlying serial port.
            """
            self.halt()
            self.serial.close()


    class XBeeReader:
        """Wrapper around the xbee module to integrate our
        _BeeReaderMixin into the appropriate base class.
        """

        def __init__(self, *args, **kwargs):
            """Save parameters for future use.

            Every parameter is used to initialize a serial.Serial
            object except for the named attribute 'zigbee' which
            define the base class to use.

            Parameter:
              - zigbee: whether to use the xbee.ZigBee base class or
                the xbee.XBee one
            """
            zigbee = kwargs.pop('zigbee', False)
            base_cls = ZigBee if zigbee else XBee
            self._serial = Serial(*args, **kwargs)
            self._cls = type('XBeeReader', (_BaseReaderMixin, base_cls), {})

        def __call__(self, callback):
            """Generate the appropriate object to read data.

            Parameter:
              - callback: the function that will be called each
                time a valid data is read.
            """
            return self._cls(self._serial, callback)

drone_racer/__init__.py

"""Pubilc interface to the various components defined in this package.

Allows to construct the GUI responsible of the whole application
and to select a reader from the built-in ones.
"""


from .ui import DroneRacer as Application
from .threads import StdInReader, XBeeReader, UDPReader


__all__ = [
    'Application',
    'StdInReader',
    'XBeeReader',
    'UDPReader',
]

droneracer.py

"""Drone Racer is a project primarily developed for the DroneFest
organized as part of the FabLab Festival 2015. Its aim is to provide
an all-in-one interface for races organizers to:

  - create different events for drones competition;
  - register contestants and their associated drones;
  - classify drones into categories;
  - create several routes with their own set of rules for each event;
  - setup and monitor races on a designated route;
  - gather statistics on races for drivers, event or kind of route.

To reduce the overhead of having extraneous services for database
access, Drone Racer makes use of the python's built-in sqlite module.
It uses it to store informations on the contestants, the drones, the
different type of routes and the races leaderboards.

Additionally, setup, updates & leaderboard for each race can be sent
to a RESTful API for the audience.
"""


import os

from argparse import ArgumentParser
import drone_racer


# Be sure to be at the right place for relative path of images in Gtk
os.chdir(os.path.dirname(os.path.abspath(__file__)))

parser = ArgumentParser(description='Interface graphique "Drone Racer"')

# GUI args
parser.add_argument('--fancy-title', dest='fancy', action='store_true',
                    help='Utilise une barre de titre un peu plus Gtk3')

# XBee args
parser.add_argument('--serial-port', dest='serial', metavar='FILE',
                    default=None, help='Spécifie le port série à utiliser '
                    'pour récupérer les informations provenant du XBee')
parser.add_argument('--zigbee', dest='zigbee', action='store_true',
                    help='Spécifie si le module XBee est un ZigBee')
parser.add_argument('--baudrate', dest='baudrate', metavar='BPS',
                    type=int, default=9600, help='Débit du port série '
                    'utilisé pour la connexion avec le module XBee')

# UDP args
parser.add_argument('--use-udp', dest='udp', action='store_true',
                    help='Spécifie si la communication doit se faire '
                    'par datagrames UDP.')
parser.add_argument('--port', dest='port', metavar='NUM', type=int,
                    default=4387, help='Port à utiliser pour l’écoute UDP')

# Choose the appropriate reader
args = parser.parse_args()
if args.serial is not None:
    reader = drone_racer.XBeeReader(
            args.serial, args.baudrate, zigbee=args.zigbee)
elif args.udp:
    reader = drone_racer.UDPReader(args.port)
else:
    reader = drone_racer.StdInReader()

# Launch the GUI (which will, in turn, start the reader)
app = drone_racer.Application(reader, args.fancy)
app.run()

Further improvements

To improve readability and sharing, you may be interested in taking a look into the gettext module.

\$\endgroup\$

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.