To understand the core concepts in ThingFlow, let us build a simple app with a
dummy sensor that generates random data and feeds it to a dummy LED. The final
code for this example is at
Input Things and Output Things¶
Each ThingFlow “thing” is either an output thing, which emits events and and puts the into the workflow, an input thing, which consumes events, accepting event streams from the workflow, or both.
An output thing may create multiple output event streams. Each output stream is associated with a named output port. Likewise, an input thing may accept input streams via named input ports. Input and output ports form the basis for interconnections in our data flows.
In general, we can connect an input port to an output port via an output thing’s
connect()method like this:output_thing.connect(input_thing, port_mapping=('output_port_name', 'input_port_name'))
There also exists a special default port, which is used when no port name
is specified on a connection. If you leave off the port mapping
parameter in the
connect() call, it maps the default port of the
output to the default port of the input:
Once connected through the
connect call, a output and input thing interact
through three methods on the input thing:
on_next, which passes the next event in the stream to the input thing.
on_error, which should be called at most once, if a fatal error occurs. The exception that caused the error is passed as the parameter.
on_completed, which signals the end of the stream and takes no parameters.
Note that each output port may have multiple connections. The functionality
thingflow.base.OutputThing base class handles dispatching the
events to all downstream consumers.
More terms for specialized things¶
We call things which have a default input port and a default output port filters.
Filters can be easily composed into pipelines. We talk more about filters
below. A number of filters are defined by ThingFlow under the
Some things interface to outside world, connecting ThingFlow to
transports and data stores like MQTT,
PostgreSQL, and flat CSV files. We call these things adapters. Several may be
thingflow.adapters. We call an output thing that emits events
coming from an outside source a reader. An input thing which accepts event
and conveys them to an outside system a writer.
Since ThingFlow is designed for Internet of Things applications, data capture
from sensors is an important part of most applications. To this end, ThingFlow
provides a sensor abstraction. A sensor is any python class that implements
sample() method and has a
sensor_id property. The
takes no arguments and returns the current value of the sensor. The
property is used to identify the sensor in downstream events. Optionally, a
sensor can indicate that there is no more data available by thowing a
To plug sensors into the world of input and output things, ThingFlow provides
SensorAsOutputThing class. This class wraps any sensor, creating an
output thing. When the thing is called by the scheduler, it calls the sensor’s
method, wraps the value in an event (either
SensorEvent or a custom
event type), and pushes it to any connected input things. We will see
SensorAsOutputThing in action below.
There are cases where this simple sensor abstraction is not sufficient to model
a real-life sensor or you are outputting events that are not coming directly
from a sensor (e.g. from a file or a message broker). In those situations,
you can just create your own output thing class, subclassing from the base
Implementing a Sensor¶
Now, we will implement a simple test sensor that generates random values.
There is no base sensor class in ThingFlow, we just need a class that
sensor_id property and a
sample() method. We’ll take
sensor_id value as an argument to
__init__(). The sample
value will be a random number generated with a Gaussian distribution,
random.gauss. Here is the code for a simple version of our
import random random.seed() class RandomSensor: def __init__(self, sensor_id, mean, stddev): """Generate a random value each time sample() is called, using the specified mean and standard deviation. """ self.sensor_id = sensor_id self.mean = mean self.stddev = stddev def sample(self): return random.gauss(self.mean, self.stddev) def __str__(self): return "RandomSensor(%s, %s, %s)" % \ (self.sensor_id, self.mean, self.stddev)
This sensor will generate a new random value each time it is called. If we
run it with a scheduler, it will run forever (at least until the program
is interrupted via Control-C). For testing, it would be helpful to stop
the program after a certain number of events. We can do that, by passing
an event limit to the constructor, counting down the events, and throwing
StopIteration exception when the limit has been reached. Here is
an improved version of our sensor that can signal a stop after the specified
number of events:
import random random.seed() import time from thingflow.base import SensorAsOutputThing class RandomSensor: def __init__(self, sensor_id, mean, stddev, stop_after): """This sensor will signal it is completed after the specified number of events have been sampled. """ self.sensor_id = sensor_id self.mean = mean self.stddev = stddev self.events_left = stop_after def sample(self): if self.events_left>0: data = random.gauss(self.mean, self.stddev) self.events_left -= 1 return data else: raise StopIteration def __str__(self): return "RandomSensor(%s, %s, %s)" % \ (self.sensor_id, self.mean, self.stddev)
Now, let’s instantiate our sensor:
from thingflow.base import SensorAsOutputThing MEAN = 100 STDDEV = 10 sensor = SensorAsOutputThing(RandomSensor(1, MEAN, STDDEV, stop_after=5))
Implementing an Input Thing¶
Now, let us define a simple intput thing – a dummy LED actuator. The LED will
inherit from the
thingflow.base.IntputThing class, which defines the
input thing interface for receiving events on the default port. Here is the code:
from thingflow.base import InputThing class LED(InputThing): def on_next(self, x): if x: print("On") else: print("Off") def on_error(self, e): print("Got an error: %s" % e) def on_completed(self): print("LED Completed") def __str__(self): return 'LED'
As you can see, the main logic is in
on_next – if the event looks like a
true value, we just print “On”, otherwise we print “Off”. We won’t do anything
special for the
on_completed callbacks. Now, we can
instantiate an LED:
led = LED()
A filter is a thing that as a single default input port and a single default
output port. There is a base class for filters,
which subclasses from both
Although you can instantiate
filter classes directly, ThingFlow makes use of some Python metaprogramming
to dynamically add convenience methods to the base
to create and return filtes. This allows filters can be easily chained
together, implementing multi-step query pipelines without any glue code.
Let us now create a series of filters that connect together our dummy light
sensor and our LED. Here is some code to look at each event and send
the LED if the value exceeds the mean (provided to the sensor) and
import thingflow.filters.map sensor.map(lambda evt: evt.val > MEAN).connect(led)
import statement loads the code for the
map filter. By loading
it, it is added as a method to the
OutputThing class. Since the sensor was
SensorAsOutputThing, which inherits from
gets this method as
well. Calling the method creates a filter which runs the supplied
anonymous function on each event. This
filter is automatically connected to the sensor’s default output port.
map call returns the filter, allowing it to be used
in chained method calls. In this case, we
led to the
filter’s event stream.
Inside the Map filter¶
It is important to note that the call to a filter method returns a filter
object and not an event. This call happens at initializaiton time.
To get a better understanding of what’s happening, let’s take a look
First, let us create a straightfoward implementation of our filter
by subclassing from the base
Filter class and then overridding
from thingflow.base import Filter, filtermethod class MapFilter(Filter): def __init__(self, previous_in_chain, mapfun): super().__init__(previous_in_chain) self.mapfun = mapfun def on_next(self, x): next = self.mapfun(x) if next is not None: self._dispatch_net(next) @filtermethod(OutputThing) def map(this, mapfun): return MapFilter(this, mapfun)
In this case, the
on_next method applies the provided
mapping function to each incoming event and, if the result is not
passes it on to the default output port via the method
(whose implementation is inherited from the base
__init__ method of our filter, we accept a
argument and pass it to the parent class’s constructor. As the name implies,
this argument should be the previous filter in the chain which is acting as
a source of events to this filter.
Filter.__init__ will perform a
previous_in_chain.connect(self) call to establish the connection.
We can now wrap our filter in the function
map, which takes the previous
filter in the chain and our mapping function as arguments, returning a new
MapFilter. The decorator
functionfilter is used to attach
this function to
OutputThing as a method. We can then make calls
The actual code for
map``in ThingFlow map be found in the module ``thingflow.filters.map.
It is written slightly differently, in a more functional style:
from thingflow.base import OutputThing, FunctionFilter, filtermethod @filtermethod(OutputThing, alias="select") def map(this, mapfun): def on_next(self, x): y = mapfun(x) if y is not None: self._dispatch_next(y) return FunctionFilter(this, on_next, name='map')
FunctionFilter class is a subclass of
Filter which takes its
on_completed method implementations as function parameters.
In this case, we define
on_next inside of our
map filter. This avoids the
need to even create a
ThingFlow provides a namedtuple called
serve as elements of our data stream. The first member of the tuple, called
sensor_id is the sensor id property of the sensor from which the event
originated. The second member of the event tuple,
ts, is a timestamp
of when the event was generated. The third member,
val, is the value
returned by the sensor’s
SensorAsOutputThing wrapper class creates
SensorEvent instances by default.
However, you can provide an optional
make_sensor_event callback to
SensorAsOutputThing to override this behavior and provide your own event types.
Sensor Output Example¶
Imagine that the sensor defined above outputs the following three events, separated by 10 seconds each:
SensorEvent(1, 2016-06-21T17:43:25, 95) SensorEvent(1, 2016-06-21T17:43:35, 101) SensorEvent(1, 2016-06-21T17:43:45, 98)
select filter would output the following:
False True False
The LED would print the following:
Off On Off
Some Debug Output¶
There are a number of approaches one can take to help understand the behavior of
an event dataflow. First, can add an
output thing to various points in the
output thing just prints each event that it see. It is another
filter that can be added to the base
OutputThing class by importing the
associated Python package. For example, here is how we add it as a connection to
our sensor, to print out every event the sensor emits:
import thingflow.filters.output sensor.output()
Note that this does not actually print anything yet, we have to run the scheduler to start up our dataflow and begin sampling events from the sensor.
Another useful debugging tool is the
print_downstream method on
OutputThing. It can be called on any subclass to see a representation
of the event tree rooted at the given thing. For example, here is what we
get when we call it on the
sensor at this point:
***** Dump of all paths from RandomSensor(1, 100, 10) ***** RandomSensor(1, 100, 10) => select => LED RandomSensor(1, 100, 10) => output ************************************
OutputThing class also provices a
It will instument (transitively) all downstream connections. When the scheduler
runs the thing, all events passing over these connections will be printed.
As you can see, it is easy to create these pipelines. However, this sequence of things will do nothing until we hook it into the main event loop. In particular, any output thing that source events into the system (e.g. sensors) must be made known to the scheduler. Here is an example where we take the dataflow rooted at the light sensor, tell the scheduler to sample it once every second, and then start up the event loop:
import asyncio from thingflow.base import Scheduler scheduler = Scheduler(asyncio.get_event_loop()) scheduler.schedule_periodic(sensor, 1.0) # sample once a second scheduler.run_forever() # will run until there are no more active sensors print("That's all folks!") # This will never get called in the current version
The output will look something like this:
Off SensorEvent(sensor_id=1, ts=1466554963.321487, val=91.80221483640152) On SensorEvent(sensor_id=1, ts=1466554964.325713, val=105.20052817504502) Off SensorEvent(sensor_id=1, ts=1466554965.330321, val=97.78633493089245) Off SensorEvent(sensor_id=1, ts=1466554966.333975, val=90.08049816341648) Off SensorEvent(sensor_id=1, ts=1466554967.338074, val=89.52641383841595) On SensorEvent(sensor_id=1, ts=1466554968.342416, val=101.35659321534875) ...
The scheduler calls the
_observe method of
SensorAsOutputThing once every second.
This method samples the sensor and calls
_dispatch_next to pass it to
any downstream things connected to the output port.
In the program output above,
we are seeing the On/Off output from the LED interleaved with the original
events printed by the
output element we connected directly to the sensor.
Note that this will keep running forever, until you use Control-C to stop the
Stopping the Scheduler¶
As you saw in the last example, the
run_forever method of the scheduler will
keep on calling things as long as any have been scheduled. If we are just
running a test, it would be nice to stop the program automatically
ather than having to Control-C
out of the running program. Our sensor class addresses this by including an
stop_after parameter on the constuctor. When we instantiate our
sensor, we can pass in this additional parameter:
sensor = SensorAsOutputThing(RandomSensor(1, MEAN, STDDEV, stop_after=5))
run_forever() method does not really run forever – it only
runs until there are no more schedulable actions. When our sensor throws the
StopIteration exception, it causes the wrapping
SensorAsOutputThing to deschedule
the sensor. At that point, there are no more publishers being managed by
the scheduler, so it exits the loop inside
When we run the example this time, the program stops after five samples:
Off SensorEvent(sensor_id=1, ts=1466570049.852193, val=87.42239337997071) On SensorEvent(sensor_id=1, ts=1466570050.856118, val=114.47614678277142) Off SensorEvent(sensor_id=1, ts=1466570051.860044, val=90.26934530230736) On SensorEvent(sensor_id=1, ts=1466570052.864378, val=102.70094730226809) On SensorEvent(sensor_id=1, ts=1466570053.868465, val=102.65381015942252) LED Completed Calling unschedule hook for RandomSensor(1, 100, 10) No more active schedules, will exit event loop That's all folks!
You have reached the end of the tutorial. To learn more, you might:
- Continue with this documentation. In the next section, we look at implementing output things.
- Take a look at the code under the
- You can also read through the code in the
thingflowproper – a goal of the project is to ensure that it is clearly commented.