Hello there,
This may be off-topic, but I am willing to read data logged with the tlm-data-logger into a TCP port (eg. in this case 9999). I am scaffolding the script from this other script which reads any messages in the port 9999, but I don’t get any of them.
Do you know how could I read the data with Python?
Thanks in advance!
The source code for tlm-data-logger is freely available in Parrot Telemetry’s public github repository: https://github.com/Parrot-Developers/telemetry/blob/master/tools/gndctrl_datalogger.cpp.
In the same repository, you can also find a Telemetry consumer example that can be adapted to your needs. You need a workspace to build this example. It is recommended to use the
parrot-groundsdk (https://developer.parrot.com/docs/olympe/installation.html#clone-the-parrot-groundsdk-repo-workspace) workspace for that purpose. The parrot-groundsdk workspace does not contain the telemetry library but you can clone it inside the packages directory along with its dependencies:
git clone https://github.com/Parrot-Developers/libshdata.git packages/common/libshdata
git clone https://github.com/Parrot-Developers/telemetry.git packages/common/telemetry
Then, inside your workspace, you can build tlm-consumer:
./build.sh -p olympe -A tlm-consumer
And execute it while Parrot-Sphinx is running:
. out/olympe-linux/staging/native-wrapper.sh
tlm-consumer 0
You will find below an example that shows how to display the drone’s current position:
/**
* Copyright (C) 2019 Parrot Drones SAS
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in
* the documentation and/or other materials provided with the
* distribution.
* * Neither the name of the Parrot Company nor the names
* of its contributors may be used to endorse or promote products
* derived from this software without specific prior written
* permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
* FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
* PARROT COMPANY BE LIABLE FOR ANY DIRECT, INDIRECT,
* INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
* BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS
* OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
* AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
* OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
* OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
* SUCH DAMAGE.
*/
#include <signal.h>
#include <unistd.h>
#include "libtelemetry.hpp"
static volatile bool running = true;
static void printDouble(const char *name, const double value)
{
fprintf(stdout, "%s", name);
fprintf(stdout, " %f", value);
fprintf(stdout, "\n");
}
/**
*/
static void doTestOmniscient(uint32_t rate)
{
struct WorldPosition {
double x;
double y;
double z;
} worldPosition;
memset(&worldPosition, 0, sizeof(worldPosition));
telemetry::Consumer *consumer = telemetry::Consumer::create("/dev/shm/sphinx-11345");
consumer->reg(worldPosition.x, "omniscient_anafi4k.worldPosition.x");
consumer->reg(worldPosition.y, "omniscient_anafi4k.worldPosition.y");
consumer->reg(worldPosition.z, "omniscient_anafi4k.worldPosition.z");
consumer->regComplete();
while (running) {
usleep(rate);
consumer->getSample(NULL, TLM_LATEST);
printDouble("worldPosition.x", worldPosition.x);
printDouble("worldPosition.y", worldPosition.y);
printDouble("worldPosition.z", worldPosition.z);
}
telemetry::Consumer::release(consumer);
}
/**
*/
int main(int argc, char *argv[])
{
signal(SIGINT, [](int signo){running = false;});
signal(SIGTERM, [](int signo){running = false;});
if (argc < 2) {
fprintf(stderr, "%s <rate(us)>\n", argv[0]);
exit(1);
}
uint32_t rate = atoi(argv[1]);
if (rate == 0)
rate = 500 * 1000;
doTestOmniscient(rate);
return 0;
}
The tkgndctrl.py tool is a Python program that consumes Telemetry data, it must be executed in the tools directory and is only compatible with Python 2:
cd packages/common/telemetry/tools
python2 ./tkgndctrl.py inet:127.0.0.1:9060 5000
You can easily adapt this script to your needs.
So as far as I understand,
should get the data published by telemetryd and tlm-data-logger in the specified dataport as specified in
tlm-data-logger [<options>] <ctrladdr> [<dataport>]
but so far I don’t understanding how it works. This is what I get when I execute both
tlm-data-logger -s 1000000 inet:127.0.0.1:9060
bash out/olympe-linux/staging/native-wrapper.sh tlm-consumer 0
Could you give a more intuitive explanation of its usage? Its help only gives tlm-consumer <rate(us)> [<type>]; nothing about dataports is explained there.
Thanks in advance!
Likewise, I try to run python2 ./tkgndctrl.py inet:127.0.0.1:9060 5000 but I get a
Exception in Tkinter callback
Traceback (most recent call last):
File "/usr/lib/python2.7/lib-tk/Tkinter.py", line 1540, in __call__
return self.func(*args)
File "/usr/lib/python2.7/lib-tk/Tkinter.py", line 590, in callit
func(*args)
File "./tkgndctrl.py", line 216, in _onIdle
pomp.looper.stepLoop()
File "../../libpomp/python/pomp/looper.py", line 106, in stepLoop
while _tls.loop.step(timeout=timeout) and count < maxMsg:
File "../../libpomp/python/pomp/looper.py", line 65, in step
handler.cb(req)
File "../../libpomp/python/pomp/context.py", line 168, in _onRequest
self._doConnAdd(conn)
File "../../libpomp/python/pomp/context.py", line 185, in _doConnAdd
self.eventHandler.onConnected(self, conn)
File "../../libpomp/python/pomp/context.py", line 57, in onConnected
raise NotImplementedError()
NotImplementedError
tlm-consumer is just an example that shows how to write a telemetry consumer. If you replace consumer.cpp with the example code above and recompile tlm-consumer it should work with Sphinx and display world positions.
tlm-consumer does not depend on tlm-data-logger. They both consume and display telemetry data.
tkgndctrl.py is missing a few lines. Here’s a patch:
diff --git a/tools/tkgndctrl.py b/tools/tkgndctrl.py
index 5771fe1..f4f5698 100755
--- a/tools/tkgndctrl.py
+++ b/tools/tkgndctrl.py
@@ -167,6 +167,10 @@ class GndCtrlItf(object):
class _DataEventHandler(pomp.EventHandler):
def __init__(self, itf):
self.itf = itf
+ def onConnected(self, ctx, conn):
+ pass
+ def onDisconnected(self, ctx, conn):
+ pass
def recvMessage(self, ctx, conn, msg):
self.itf.recvDataMsg(msg)@ocrave thanks a lot! Now it works 
NB: Have you thought about simplifying the
telemetrydPython API as a more Pythonic app? Reviewing the code from thelibpompandtelemetry/toolspackages felt a lot of C++ -ish (i.e. hard to quick-review it, lots of low-level base packages, no comments at all).It would be great to see something more simple
Hey there @ocrave!
I am trying to extend the telemetry consumer to integrate it as a Python background task in my app, i.e. to keep logging to a file while running and controlling the drone on Sphinx. As of now:
The thing is that in your tkgndctrl.py example, you use a Tkinter app with its mainloop to keep the app running. I have tried switching it to a simple while True: ... and it didn’t work. Likewise, directly invoking it using threading.Thread doesn’t seem to work neither (also setting the daemon flag to True).
What can I do so to run it as a background task?
EDIT: it seems to be related to
pomp.looper
EDIT 2: Which solution would be simpler,
threading,multiprocessingorsubprocess?
Thanks in advance! 
threading is probably the simplest solution. multprocessing and subprocess are only required if you need multiple processes.
Here is a threading template for pomp:
import threading
sys.path.append("../../libpomp/python")
import pomp
class MyPompThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
pomp.looper.prepareLoop()
# create pomp context and stuff...
def run(self):
self.running = True
while self.running:
pomp.looper.stepLoop(maxMsg=1, timeout=1)
def stop(self):
self.running = False
if __name__ == '__main__':
thread = MyPompThread()
thread.start()This is not working 
when I run pomp using that template, in which I also add a signal handler for CTRL+C so when it’s pressed, it runs MyPompThread.stop, this happens:
(I set to print all the telemetry data I was receiving)
{'coord': 'z', 'pid': '2', 'ts': 236, 'topic': 'omniscient_bebop2.worldAttitude.z', 'value': 0.004425906562459608, 'namespace': 'omniscient_bebop2'}
{'coord': 'worldPressure', 'pid': '2', 'ts': 236, 'topic': 'omniscient_bebop2.worldPressure', 'value': 101324.42076483695, 'namespace': 'omniscient_bebop2'}
{'coord': 'worldTemperature', 'pid': '2', 'ts': 236, 'topic': 'omniscient_bebop2.worldTemperature', 'value': 298.14967581859844, 'namespace': 'omniscient_bebop2'}
{'coord': 'lastCollisionTime', 'pid': '2', 'ts': 236, 'topic': 'omniscient_bebop2.lastCollisionTime', 'value': 236.835, 'namespace': 'omniscient_bebop2'}
^C
Exiting telemetry daemon loop!
^C
Exiting telemetry daemon loop!
Exception ignored in: <module 'threading' from '/usr/lib/python3.5/threading.py'>
Traceback (most recent call last):
File "/usr/lib/python3.5/threading.py", line 1288, in _shutdown
t.join()
File "/usr/lib/python3.5/threading.py", line 1054, in join
self._wait_for_tstate_lock()
File "/usr/lib/python3.5/threading.py", line 1070, in _wait_for_tstate_lock
elif lock.acquire(block, timeout):
File "/home/pachacho/Documents/anafi_tools/envdata/telemetry/telemetryd.py", line 72, in _signal_handler
self.stop()
File "/home/pachacho/Documents/anafi_tools/envdata/telemetry/telemetryd.py", line 130, in stop
self.ctrlCtx.stop()
File "/home/pachacho/Documents/parrot/groundsdk/packages/common/libpomp/python/pomp/context.py", line 115, in stop
self._doConnRemove(self.connections[0])
File "/home/pachacho/Documents/parrot/groundsdk/packages/common/libpomp/python/pomp/context.py", line 190, in _doConnRemove
self.eventHandler.onDisconnected(self, conn)
File "/home/pachacho/Documents/anafi_tools/envdata/telemetry/telemetryd.py", line 182, in onDisconnected
sys.exit(0)
SystemExit: 0
On the first CTRL+C press, the signal was caught and MyPompThread.stop() is ran. However, pomp.looper.exitLoop() never exits.
On the second CTRL+C the program exits (i.e. Python exits).
class TelemetryDaemon:
SAMPLE_RATE = 1000 * 1000 # Samples every 1000ms
MSG_RATE = 1000 * 1000 # Message every 1s
DEFAULT_CTRLADDR = "inet:127.0.0.1:9060"
DEFAULT_DATAPORT = "5000"
def __init__(self, on_sample, name="tkgndctrl",
ctrlAddr="inet:127.0.0.1:9060", dataPort=5000, rate=1000):
"""
Setups the telemetry daemon's default configuration. Rate defaults to
1000 ms (1s).
"""
self.name = name
self.sample_rate = int(rate) * 1000
self.ctrlAddr = ctrlAddr
self.dataPort = dataPort
self.on_sample = on_sample
self.ctrlCtx = pomp.Context(TelemetryDaemon._CtrlEventHandler(self))
self.dataCtx = pomp.Context(TelemetryDaemon._DataEventHandler(self))
self.looper = TelemetryDaemon.PompThread()
self.sections = {}
signal.signal(signal.SIGINT, self._signal_handler)
def _signal_handler(self, sig, frame):
print("\nExiting telemetry daemon loop!\n")
self.stop()
def stop(self):
"""Stops the connection with the telemetry service and the pomp loop"""
self.ctrlCtx.stop()
self.dataCtx.stop()
self.looper.stop()
# [...]
class PompThread(threading.Thread):
def __init__(self):
"""Setups the pomp service in a separate thread"""
threading.Thread.__init__(self)
pomp.looper.prepareLoop()
def start(self):
self.running = True
while self.running:
pomp.looper.stepLoop(maxMsg=1, timeout=1)
def stop(self):
# This part NEVER ends
self.running = False
pomp.looper.exitLoop()
Honestly, I don’t know what more to do. Hope that someone can help me out 
>>> import sys
>>> sys.path.append("/home/pachacho/Documents/anafi_tools/envdata/telemetry")
>>> from telemetryd import TelemetryDaemon
>>> def hello(x):
... pass
...
>>> damn = TelemetryDaemon(hello)
>>> damn.start()
# I pressed enter
# I pressed enter
# I pressed enter
So,
I have been able to execute the process in background by using multiprocessing.Process, but the thing is that this way it’s not possible to share memory, which probably is required for pomp to work. E.g.
# to_ABC just does a 'pass'
>>> damn = TelemetryDaemon(to_ABC)
>>> damn.start()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/home/pachacho/Documents/anafi_tools/envdata/telemetry/telemetryd.py", line 154, in start
self.ctrlCtx.connect(family, addr)
File "/home/pachacho/Documents/parrot/groundsdk/packages/common/libpomp/python/pomp/context.py", line 91, in connect
self._start(1, self._runClient)
File "/home/pachacho/Documents/parrot/groundsdk/packages/common/libpomp/python/pomp/context.py", line 152, in _start
self.mainHandler = looper.Handler(self._onRequest)
File "/home/pachacho/Documents/parrot/groundsdk/packages/common/libpomp/python/pomp/looper.py", line 78, in __init__
self._loop = _tls.loop
AttributeError: '_thread._local' object has no attribute 'loop'
However, running threading.Thread allows to share memory i.e. the looper starts, but the main program gets blocked. WHAT CAN I DO?
I am trying to run it in a separate script and then invoking it in the main app with a subprocess i.e.
EDIT: I am willing to use the telemetry logger in a
forloop that generates X number of simulations to get the drone/pedestrian data for each simulation.
cmd = "python [...]/telemetry/tools/tkgndcntrl.py inet... 5000"
# cmd = cmd if using shell=True
# cmd = shlex.split(cmd) if using shell=False
subprocess.Popen(cmd, shell=[True or False], [other opts])
This doesn’t seem to work neither, so how the holy can I run the telemetry logger without blocking the main app and, more importantly, how can I exit it without getting blocked on pomp.looper.exitLoop???
Going back to the tkgndcntrl.py example posted on Github,
Wich is the proper way to exit the telemetryd loop?
Final update: until no official solution is up there (hope is not the shitty one I’m proposing, please coders mind the docs and PEPs), what I am doing is to run os._exit(0) when a KeyboardInterrupt exception (CTRL+C) is caught.
Thanks @ocrave for clarifying some core concepts on telemetry logging 
This solution works:
import sys, logging
import optparse
import signal
import threading
sys.path.append("../../libpomp/python")
import pomp
from tlmb_parser import TlmbSection
GNDCTRL_PROTOCOL_VERSION = 1
GNDCTRL_MSG_CONN_REQ =1
GNDCTRL_MSG_CONN_RESP =2
GNDCTRL_MSG_SUBSCRIBE_REQ = 3
GNDCTRL_MSG_SUBSCRIBE_RESP = 4
GNDCTRL_MSG_UNSUBSCRIBE_REQ = 5
GNDCTRL_MSG_UNSUBSCRIBE_RESP = 6
GNDCTRL_MSG_SECTION_ADDED = 7
GNDCTRL_MSG_SECTION_REMOVED = 8
GNDCTRL_MSG_SECTION_CHANGED = 9
GNDCTRL_MSG_SECTION_SAMPLE = 10
SAMPLE_RATE = 200 * 1000 # Samples every 200ms
MSG_RATE = 1000 * 1000 # Message every 1s
#===============================================================================
#===============================================================================
_USAGE = (
"usage: %prog [<options>] <ctrladdr> <dataport>\n"
"Connect to a ishtar server\n"
"\n"
" <options>: see below\n"
" <ctrladdr> : control address\n"
" <dataport> : data port\n"
"\n"
"<ctrladdr> format:\n"
" inet:<addr>:<port>\n"
" inet6:<addr>:<port>\n"
" unix:<path>\n"
" unix:@<name>\n"
)
#===============================================================================
#===============================================================================
class GndCtrlItf(object):
def __init__(self, app, name, ctrlAddr, dataPort):
self.app = app
self.name = name
self.ctrlAddr = ctrlAddr
self.dataPort = dataPort
self.ctrlCtx = pomp.Context(GndCtrlItf._CtrlEventHandler(self))
self.dataCtx = pomp.Context(GndCtrlItf._DataEventHandler(self))
self.sections = {}
def start(self):
(family, addr) = pomp.parseAddr(self.ctrlAddr)
self.ctrlCtx.connect(family, addr)
(family, addr) = pomp.parseAddr("inet:0.0.0.0:%u" % self.dataPort)
self.dataCtx.bind(family, addr)
def stop(self):
self.ctrlCtx.stop()
self.dataCtx.stop()
def recvCtrlMsg(self, msg):
if msg.msgid == GNDCTRL_MSG_CONN_RESP:
dec = pomp.Decoder()
dec.init(msg)
status = dec.readU32()
count = dec.readU32()
logging.info("Connected: status=%d", status)
for _ in range(0, count):
key = dec.readStr()
val = dec.readStr()
logging.info("%s='%s'", key, val)
elif msg.msgid == GNDCTRL_MSG_SECTION_ADDED:
(sectionId, sectionName) = msg.read("%u%s")
section = TlmbSection(sectionId, sectionName)
self.sections[sectionId] = section
logging.info("Section added: %s(%d)", sectionName, sectionId)
elif msg.msgid == GNDCTRL_MSG_SECTION_REMOVED:
(sectionId, ) = msg.read("%u")
section = self.sections.get(sectionId, None)
if section is not None:
logging.info("Section removed: %s(%d)", section.sectionName, sectionId)
self.app.sectionRemoved(section.sectionName)
del self.sections[sectionId]
elif msg.msgid == GNDCTRL_MSG_SECTION_CHANGED:
(sectionId, buf) = msg.read("%u%p")
section = self.sections.get(sectionId, None)
if section is not None:
newSection = TlmbSection(sectionId, section.sectionName)
newSection.readHeader(buf)
logging.info("Section changed: %s(%d)", section.sectionName, sectionId)
self.sections[sectionId] = newSection
elif msg.msgid == GNDCTRL_MSG_SECTION_SAMPLE:
# Only if client is configured to receive samples on the control channel
(sectionId, sec, nsec, buf) = msg.read("%u%u%u%p")
self.recvSample(sectionId, (sec, nsec), buf)
def recvDataMsg(self, msg):
if msg.msgid == GNDCTRL_MSG_SECTION_SAMPLE:
(sectionId, sec, nsec, buf) = msg.read("%u%u%u%p")
self.recvSample(sectionId, (sec, nsec), buf)
def recvSample(self, sectionId, timestamp, buf):
section = self.sections.get(sectionId, None)
if section is None:
return
logging.debug("Sample: %s(%d) %d.%06d", section.sectionName, sectionId,
timestamp[0], timestamp[1] // 1000)
varOff = 0
for varId in range(0, len(section.varDescs)):
varDesc = section.varDescs[varId]
varLen = varDesc.getTotalSize()
if varOff + varLen > len(buf):
break
varBuf = buf[varOff:varOff+varLen]
self.app.sample(sectionId, timestamp, varId, varDesc, varBuf)
varOff += varLen
class _CtrlEventHandler(pomp.EventHandler):
def __init__(self, itf):
self.itf = itf
def onConnected(self, ctx, conn):
# Send connection request
conn.send(GNDCTRL_MSG_CONN_REQ, "%u%s%u%u%u",
GNDCTRL_PROTOCOL_VERSION, self.itf.name, self.itf.dataPort,
SAMPLE_RATE, MSG_RATE)
def onDisconnected(self, ctx, conn):
# Clear internal state
logging.info("Disconnected")
self.sections = {}
def recvMessage(self, ctx, conn, msg):
self.itf.recvCtrlMsg(msg)
class _DataEventHandler(pomp.EventHandler):
def __init__(self, itf):
self.itf = itf
def onConnected(self, ctx, conn):
pass
def onDisconnected(self, ctx, conn):
pass
def recvMessage(self, ctx, conn, msg):
self.itf.recvDataMsg(msg)
class App():
def __init__(self, args):
self.sock_family = None
self.sock_addr = None
self.running = False
self.thread = None
self.itf = GndCtrlItf(self, "example", args[0], int(args[1]))
signal.signal(signal.SIGINT,
lambda signal, frame: self._signal_handler())
signal.signal(signal.SIGTERM,
lambda signal, frame: self._signal_handler())
def _signal_handler(self):
self.running = False
def __del__(self):
if self.running:
self.stop()
def start(self):
self.running = True
self.thread = threading.Thread(target=self.worker)
self.thread.start()
def stop(self):
self.running = False
self.thread.join()
def worker(self):
# setup loop for main thread
pomp.looper.prepareLoop()
# create pomp context
self.itf.start()
# run main loop
while self.running:
pomp.looper.stepLoop(maxMsg=1, timeout=1)
# destroy pomp context
self.itf.stop()
def sectionRemoved(self, sectionName):
pass
def sample(self, sectionId, timestamp, varId, varDesc, buf):
print(sectionId, timestamp, varId, varDesc, buf)
#===============================================================================
#===============================================================================
def main():
(options, args) = parseArgs()
setupLog(options)
try:
app = App(args)
app.start()
except KeyboardInterrupt:
root.destroy()
sys.exit(0)
#===============================================================================
#===============================================================================
def parseArgs():
# Setup parser
parser = optparse.OptionParser(usage=_USAGE)
parser.add_option("-q", "--quiet",
dest="quiet",
action="store_true",
default=False,
help="be quiet")
parser.add_option("-v", "--verbose",
dest="verbose",
action="store_true",
default=False,
help="verbose output")
# Parse arguments
(options, args) = parser.parse_args()
if len(args) != 2:
parser.error("Bad number or arguments")
return (options, args)
#===============================================================================
#===============================================================================
def setupLog(options):
logging.basicConfig(
level=logging.INFO,
format="[%(levelname)s][%(asctime)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
stream=sys.stderr)
logging.addLevelName(logging.CRITICAL, "C")
logging.addLevelName(logging.ERROR, "E")
logging.addLevelName(logging.WARNING, "W")
logging.addLevelName(logging.INFO, "I")
logging.addLevelName(logging.DEBUG, "D")
# Setup log level
if options.quiet == True:
logging.getLogger().setLevel(logging.CRITICAL)
elif options.verbose:
logging.getLogger().setLevel(logging.DEBUG)
#===============================================================================
#===============================================================================
if __name__ == "__main__":
main()Cool! Both solutions provided work 
You’re a lifesaver @ocrave
However, with both codes I’m still getting Rx buffer full (4096 bytes). Does that affect the program execution or is it just a warning message?
You should get this warning only once at startup and can safely ignore it.