# “Drone Racer” now shipping in your local area

This question is a follow-up of this one; where I present a software to plan and manage drone races. I might come with a new question about the SQL part in a few days as well. The code can still be browsed on GitHub.

Since then, on top of the changes in documentation that I made in my own answer, I included the gettext module and changed the CLI to include sub-commands.

On top of the previous ones, I'm a little bit concerned about the layout of my project since I basically put everything in a single directory:

+ droneracer.py
+ drone_racer
+ fr
| + LC_MESSAGES
|   + cli.po
|   + cli.mo
|   + utils.po
|   + utils.mo
+ __init__.py
+ console.py
+ i18n.py
+ rest.py
+ sql.py
+ ui.py


The program invocation now looks like one of the following:

• python droneracer.py --fancy-title xbee /dev/ttyUSB0
• python droneracer.py udp --port 3487
• python droneracer.py --fancy-title

The code became:

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.

to a RESTful API for the audience.
"""

import os

from argparse import ArgumentParser
from drone_racer.i18n import translations
import drone_racer

_, _N = translations('cli')
XBEE_NAMES = 'xbee', 'bee', 'serial'
UDP_NAMES = 'udp', 'wifi'

parser = ArgumentParser(description=_('"Drone Racer"\'s Graphical User Interface'))
'--fancy-title', dest='fancy', action='store_true',
help=_('Use a fancier (Gtk3 like) titlebar for the GUI'))
title='communication', dest='reader', description=_('List off all '
'communication channels to get data from the gates. If none is '
help=_('More options are available per channel'))

name, *aliases = XBEE_NAMES
name, aliases=aliases, help=_('Communication through XBee frames'))
'device', metavar='FILE', default=None,
help=_('Serial file mapped to the XBee pins'))
'--zigbee', dest='zigbee', action='store_true',
help=_('Switch indicating wether it is an XBee or a ZigBee'))
'--baudrate', dest='baudrate', metavar='BPS', type=int, default=9600,
help=_('Serial port communication speed'))

name, *aliases = UDP_NAMES
name, aliases=aliases, help=_('Communication through UDP datagrams'))
'--port', dest='port', metavar='NUM', type=int, default=4387,
help=_('Socket port to listen on'))

args = parser.parse_args()
args.serial, args.baudrate, zigbee=args.zigbee)
else:

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

# Launch the GUI (which will, in turn, start the reader)
app.run()


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

__all__ = [
'Application',
]


drone_racer/i18n.py

import os.path
import gettext

def translations(domain):
locales_dir = os.path.abspath(os.path.dirname(__file__))
translation = gettext.translation(domain, locales_dir)
return translation.gettext, translation.ngettext


"""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

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 select import select
try:
from serial import Serial
from xbee import XBee, ZigBee
except ImportError:
XBee = None

from .i18n import translations

_, _N = translations('utils')

"""Base class for custom data readers."""

def __init__(self):
statuses.
"""

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
"""
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:
except TypeError:
pass
else:
self._process_value(gate, drone)

def stop(self):
self._should_continue = False

"""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)

"""Read data from stdin. Primarily used for tests and debug."""

"""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

"""Read data from UDP datagrams. Used when communicating via
WiFi with the gates.
"""

def __init__(self, port):
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]

"""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)
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:
file=sys.stderr)
print(e, file=sys.stderr)
else:
return gate, drone

if XBee is None:
"""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
"""
super().__init__()
file=sys.stderr)

"""Cancel this thread to avoid burning resources."""
self._should_continue = False

else:
"""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:
response_dict['rf_data']), file=sys.stderr)
print(e, file=sys.stderr)
except KeyError as e:
print(e, file=sys.stderr)
else:
self._update_data(gate, drone)

def stop(self):
underlying serial port.
"""
self.halt()
self.serial.close()

"""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)

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)

• Are you actually importing _ as a callable? That seems terribly dodgy to me, why not give it a real name? – SuperBiasedMan Jan 8 '16 at 10:43
• @SuperBiasedMan Because it is kind of a convention when using gettext. It is also used like that in Django for instance. I contemplated using tr() instead of _(); but the simplicity of using tools to manage gettext-aware code had me choose the latter. – Mathias Ettinger Jan 8 '16 at 10:48
• Ah I see, the point is to be unnoticeable especially as it's called on so many strings. That makes sense. – SuperBiasedMan Jan 8 '16 at 10:57
• You have a typo in the docstring of drone_racer/__init__.py (pubilc -> public) – Dannnno Jan 10 '16 at 17:31

## i18n

This is just me, but I've always found the _ and _N aspect of gettext a little hard to read and somewhat too tightly coupled. I've always preferred something like this:

i18n.py

import os.path
import gettext

def _translations(domain):
locales_dir = os.path.abspath(os.path.dirname(__file__))
translation = gettext.translation(domain, locales_dir)
return translation.gettext, translation.ngettext

class _BaseTranslator(object):

@staticmethod
def _translations(domain):
locales_dir = os.path.abspath(os.path.dirname(__file__))
translation = gettext.translation(domain, locales_dir)
return translation.gettext, translation.ngettext

def __init__(self, domain):
self.domain = domain
self._, self._N = self._translations(domain)

class _CliTranslations(_BaseTranslator):

def __init__(self, domain='cli'):
super().__init__(domain)

@property
def description(self):
return self._('"Drone Racer"\'s Graphical User Interface')

# etc

util_translations = _UtilTranslations('util')
cli_translations = _CliTranslations('cli')


cli.py

from i18n import cli_translations as cli

parser = ArgumentParser(description=cli.description)


I understand that it isn't for everyone, but I've always preferred that the original strings be less embedded in the source code, and instead be in the translating engine. I actually prefer it even more if they aren't in the source code at all, but rather in some JSON or XML file (or whatever your preferred format is). By putting all of the strings into the translating engine instead, you make it much easier to change where and how the strings and their translations are stored and received, and you don't have to change the source of any of the other files.

If you don't like that they're instances, you can do some hackery for class-level properties - I've never found a satisfying way to implement that where the properties aren't read-only, but in this context you want them to be read-only (win-win).

### Subclassing Thread

No other methods (except for the constructor) should be overridden in a subclass. In other words, only override the __init__() and run() methods of this class.

Now, currently, I don't see anything in them that actually overrides anything in the Thread class, however I would be mindful of this in case a future Python version changes it (tbh I don't see that happening, but it's worth being considerate of).

It seems that you don't actually need your readers to subclass Thread - what about making each of them callable objects instead, and then set them as the targets of the threads? Then you'll avoid any of these potential issues, and avoid having to muck around with Thread too much.

### Signalling

Instead of using a boolean to determine if the reader should continue, use the appropriate concurrency primitive, in this case threading.Event.

### dummy_threading

I don't know what sort of machines you're running this on, but if there's any chance that _thread won't be available you should have something like

try:
except ImportError:


### Appropriate errors

_process_value seems like it should do something if the drone id is less than 0 - probably emit some warning for the user so they know that they, or someone else, has done something wrong. The only reason to just return is if you either don't think it'll ever happen, or if ignoring it is fine. It seems that it would probably be indicative of some larger problem though, one that the operator/officiator/someone should be aware of.

Likewise, in run you ignore TypeErrors - why? Again, unless they can be safely ignored and wouldn't be either unusual or unexpected you should at least give some sort of warning that something unusual might be going on.

### Typo?

Should this say _BeeReaderMixin? Or are you trying to incorporate BaseReader? Some combination thereof?

self._cls = type('XBeeReader', (_BaseReaderMixin, base_cls), {})


The whole XBeeReader seems like it could be incorporated into a metaclass and then used that way instead of how you have implemented it, but metaclasses aren't my forte.

• Thanks for the typos, will edit ASAP. Interesting thoughts overall, too. About the errors, I should be able to remove _process_value since the callback data already check for the proper range (and print messages there); I needed it because I used to return -1, '?' for invalid data in read_new_value. Now that I return None I need to catch TypeError for the unpacking instead. Will add a comment to highlight that. – Mathias Ettinger Jan 10 '16 at 19:10
• And for the threading part: both xbee.XBee and xbee.ZigBee are already subclasses of Thread. I built around that to get a consistent interface. Do you feel like there could be a better way? Also what are the advantages of Event (flag + wait) here over a simple boolean (flag)? I don't need to wait until the flag is set, I need to work (monitor communication) until the flag is set. – Mathias Ettinger Jan 10 '16 at 19:14
• @MathiasEttinger If they're already subclasses of Thread then I wouldn't change it. I'd say the biggest advantage of Event would be that you could more easily coordinate multiple readers, if you needed them. Additionally, I generally prefer things designed to work with concurrency (such as Event) than the language defaults (such as booleans), but that's a personal preference. If you set the flag before starting, then you don't wait until it's set, you just work as long as it is set. – Dannnno Jan 11 '16 at 17:20

Consider using the timeout facilities built into the socket module instead of using select.select. Basically, you can call settimeout(amount) on any socket created and each blocking operation on that socket will raise socket.timeout exception after the specified amount of seconds if the operation did not complete.

Your UDPReader can thus look like:

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

def __init__(self, port):
statuses.

Parameter:
- port: the socket port to listen on.
"""
super().__init__()
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
iface = socket.gethostname()
self.socket.bind((iface, port))
# Non-blocking read so this thread will shut down with the application
self.socket.settimeout(1)

"""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).
"""
try:
msg = self.socket.recv(128) # Way too much for messages like <A:1>
except socket.timeout:
return
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:
file=sys.stderr)
print(e, file=sys.stderr)
else:
return gate, drone