Pull 'n' Push
Project description
Pull 'n' Push
Pulls data from sources and pushes it to sinks.
1. Installation
2. Getting started
3. Runner
4. Building Blocks
4.1. Pull
4.2. Push
4.3. Selector
4.4. Dependencies
4.5. Envelope (>= 0.7.1)
4.6. Engines (>= 0.10.0)
4.6.1. pnp.engines.sequential.SequentialEngine
4.6.2. pnp.engines.thread.ThreadEngine
4.6.3. pnp.engines.process.ProcessEngine
5. Useful hints
5.1. Configuration checking
5.2. Logging (>= 0.11.0)
5.3. dictmentor (>= 0.11.0)
5.4. Advanced selector expressions (>= 0.12.0)
5.5. Docker images
6. Plugins
6.1. pnp.plugins.pull.fs.FileSystemWatcher
6.2. pnp.plugins.pull.gpio.Watcher
6.3. pnp.plugins.pull.http.Server
6.4. pnp.plugins.pull.monitor.Stats
6.5. pnp.plugins.pull.mqtt.Subscribe
6.6. pnp.plugins.pull.sensor.DHT
6.7. pnp.plugins.pull.sensor.OpenWeather
6.8. pnp.plugins.pull.simple.Count
6.9. pnp.plugins.pull.simple.Repeat
6.10. pnp.plugins.pull.zway.ZwayPoll
6.11. pnp.plugins.pull.zway.ZwayReceiver
6.12. pnp.plugins.push.fs.FileDump
6.13. pnp.plugins.push.http.Call
6.14. pnp.plugins.push.ml.FaceR
6.15. pnp.plugins.push.mqtt.Publish
6.16. pnp.plugins.push.notify.Pushbullet
6.17. pnp.plugins.push.simple.Echo
6.18. pnp.plugins.push.simple.Execute
6.19. pnp.plugins.push.storage.Dropbox
6.20. pnp.plugins.push.timedb.InfluxPush
7. Changelog
1. Installation
pip install pnp
Optional extras
- dht: Enables
pnp.plugins.pull.sensor.DHT
(temperature and humidity sensor). Only works on ARM-based systems (like raspberry, arduino) - fswatcher: Enables
pnp.plugins.pull.fs.FileSystemWatcher
(Watch file system for created, modified, deleted, moved files) - faceR: Enables
pnp.plugins.push.ml.FaceR
(Screen image files for known faces)
Installation with extras:
pip install pnp[fswatcher,faceR]
# In case of extra 'dht' you have to enable the option --process-dependency-links ...
# ... cause the required adafruit package is not on pypi.
pip install --process-dependency-links pnp[dht]
2. Getting started
Define pulls
to suck/pull data from source systems.
Define one push
or multiple pushes
per pull to transfer the pulled data anywhere else (you only need a plugin that
knows how to handle the target). You can define your configurations in yaml
or json
.
It is up to you. I prefer yaml...
- name: hello-world
pull:
plugin: pnp.plugins.pull.simple.Repeat
args:
wait: 1
repeat: "Hello World"
push:
- plugin: pnp.plugins.push.simple.Echo
Copy this configuration and create the file helloworld.yaml
. Run it:
pnp helloworld.yaml
This example yields the string 'Hello World' every second.
Tip: You can validate your config without actually executing it with
pnp --check helloworld.yaml
3. Runner
> pnp --help
Pull 'n' Push
Usage:
pnp [(-c | --check)] [(-v | --verbose)] [--log=<log_conf>] <configuration>
pnp (-h | --help)
pnp --version
Options:
-c --check Only check configuration and do not run it.
-v --verbose Switches log level to debug.
--log=<log_conf> Specify logging configuration to load.
-h --help Show this screen.
--version Show version.
4. Building Blocks
Below the basic building blocks of pull 'n' push are explained in more detail
4.1. Pull
As stated before pulls fetch data from various source systems and/or apis. Please see the section plugins for already implemented pulls. To instantiate a pull by configuration file you only have to provide it's fully qualified name and the argument that should be passed.
- name: example
pull:
plugin: pnp.plugins.pull.mqtt.Subscribe
args:
host: localhost
port: 1883
topic: test/#
The above snippet will create a pull that listens on the topic test/# on a mqtt broker. The output of the pull is a dictionary that contains the topic, levels and the actual payload.
# When the message 'Here i am' arrives on the topic 'test/foo/bar' then the output will look like that:
{'topic': 'test/foo/bar', 'levels': ['test', 'foo', 'bar'], 'payload': 'Here i am'}
4.2. Push
A pull passes its data to multiple pushes to transfer/transform the data. For example a push might save sensor data to influx or dump a file to the file system.
- name: example
pull:
plugin: pnp.plugins.pull.mqtt.Subscribe
args:
host: localhost
port: 1883
topic: test/#
push:
- plugin: pnp.plugins.push.fs.FileDump
args:
directory: "/tmp"
binary_mode: False
- plugin: pnp.plugins.push.simple.Echo
The above snippet adds two pushes to the already known pull. The first push takes the incoming data and dumps it into the specified directory as a textfile. The second push just prints out the incoming data.
4.3. Selector
Sometimes the output of a pull needs to be transformed before the specified push can handle it. Selectors
to the
rescue. Given our input we decide to just dump the payload and print out the first level of the topic.
- name: example
pull:
plugin: pnp.plugins.pull.mqtt.Subscribe
args:
host: localhost
port: 1883
topic: test/#
push:
- plugin: pnp.plugins.push.fs.FileDump
selector: data.payload
args:
directory: "/tmp"
binary_mode: False
- plugin: pnp.plugins.push.simple.Echo
selector: data.levels[0]
Easy as that. We can reference our incoming data via data
or payload
.
4.4. Dependencies
By default any pushes will execute in parallel (not completly true) when new incoming data is available. But now it would be nice if we could chain pushes together. So that the output if one push becomes the input of the next push. The good thing is: Yes we can.
Back to our example let's assume we want to print out the path to the created file dump after the dump is created.
- name: example
pull:
plugin: pnp.plugins.pull.mqtt.Subscribe
args:
host: localhost
port: 1883
topic: test/#
push:
- plugin: pnp.plugins.push.fs.FileDump
selector: data.payload
args:
directory: "/tmp"
binary_mode: False
deps:
- plugin: pnp.plugins.push.simple.Echo
- plugin: pnp.plugins.push.simple.Echo
selector: data.levels[0]
As you can see we just add a dependant push to the previous one.
4.5. Envelope (>= 0.7.1)
Using envelopes it is possible to change the behaviour of pushes
during runtime.
Best examples are the pnp.plugins.push.fs.FileDump
and pnp.plugins.push.mqtt.MQTTPush
plugins, where
you can override / set the actual file_name
and extension
of the file to dump
resp. the topic
where the message should be published.
Given the example ...
- name: envelope
pull:
plugin: pnp.plugins.pull.simple.Count
args:
wait: 1
push:
plugin: pnp.plugins.push.fs.FileDump
selector: '{"file_name": str(data), "extension": ".cnt", "data": data}'
args:
directory: "/tmp/counter"
file_name: "counter" # Overridden by envelope
extension: ".txt" # Overridden by envelope
binary_mode: False # text mode
... this push dumps multiple files (0.cnt, 1.cnt, 2.cnt, ...) for each pulled counter value, instead of dumping one file 'couter.txt' which is overridden each time a new counter is emitted.
How does this work: If the emitted or transformed payload (via selector) contains the key data
or
payload
it is assumed that the actual payload is the data stored in this key and all other keys
represent the so called envelope
.
Remark: This feature might actually break your existing configurations if you use the plugin
pnp.plugins.pull.mqtt.MQTTPull
which will now emit an enveloped payload.
This snippet echoed a dictionary with the keys 'topic', 'levels' and 'payload' previously to version 0.7.2. It will now differentiate between the actual 'payload' (key 'payload' resp. 'data') and the envelope (other keys).
- name: subscriber
pull:
plugin: pnp.plugins.pull.mqtt.MQTTPull
args:
host: localhost
topic: test/counter
push:
plugin: pnp.plugins.push.simple.Echo
If you want to "restore" the previous behaviour, you only have to wrap the whole payload into a dictionary inside the 'payload' or 'data' key via selector.
- name: subscriber
pull:
plugin: pnp.plugins.pull.mqtt.MQTTPull
args:
host: localhost
topic: test/counter
push:
plugin: pnp.plugins.push.simple.Echo
selector: "{'data': data}"
4.6. Engines (>= 0.10.0)
If you do not specify any engine the ThreadEngine
is chosen by default accompanied by the AdvancedRetryHandler
.
This keeps maximum backwards compatibility.
4.6.1. pnp.engines.sequential.SequentialEngine
By using the Sequential
engine you can run your configs as scripts. Given the example below, the "script" will
end when it has finished counting to 3. Make sure to use the NoRetryHandler
to actually end the runner when
the pull has finished instead of retrying the "failed" pull
. You cn only run a single task not multiple.
When you want to run multiple task in a concurrent manner you have to use the ThreadEngine
or the ProcessEngine
.
#### Simple sequential handler
#### Counts from 1 to 3 and then terminates
engine: !engine
type: pnp.engines.sequential.SequentialEngine
retry_handler: !retry
type: pnp.engines.NoRetryHandler # Is the key to termination after counting has finished
tasks:
- name: sequential
pull:
plugin: pnp.plugins.pull.simple.Count
args:
wait: 1
from_cnt: 1
to_cnt: 4
push:
- plugin: pnp.plugins.push.simple.Echo
4.6.2. pnp.engines.thread.ThreadEngine
#### Will use threads to accomplish concurrency
#### Drawback: If a plugin does not stop gracefully the termination will hang...
engine: !engine
type: pnp.engines.thread.ThreadEngine
queue_worker: 1
retry_handler: !retry
type: pnp.engines.SimpleRetryHandler
tasks:
- name: threading
pull:
plugin: pnp.plugins.pull.simple.Repeat
args:
wait: 1
repeat: "Hello World"
push:
- plugin: pnp.plugins.push.simple.Echo
4.6.3. pnp.engines.process.ProcessEngine
#### Will use multiprocessing to accomplish concurrency
#### Drawback: Some plugins might not work or need to be aware of
engine: !engine
type: pnp.engines.process.ProcessEngine
queue_worker: 1
retry_handler: !retry
type: pnp.engines.SimpleRetryHandler
tasks:
- name: process
pull:
plugin: pnp.plugins.pull.simple.Repeat
args:
wait: 1
repeat: "Hello World"
push:
- plugin: pnp.plugins.push.simple.Echo
5. Useful hints
5.1. Configuration checking
You can check your pnp configuration file by starting pnp with the -c | --check
flag set. This will only run
the initializer but not execute the configuration.
pnp --check <pnp_configuration>
5.2. Logging (>= 0.11.0)
You can use different logging configurations in two ways:
# Specify when starting pnp
pnp --log=<logging_configuration> <pnp_configuration>
# Specify by environment variable
export PNP_LOG_CONF=<logging_configuration>
pnp <pnp_configuration>
A simple logging configuration that will log severe errors to a separate rotating log file looks like this:
version: 1
disable_existing_loggers: False
formatters:
simple:
format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
handlers:
console:
class: logging.StreamHandler
level: DEBUG
formatter: simple
stream: ext://sys.stdout
error_file_handler:
class: logging.handlers.RotatingFileHandler
level: ERROR
formatter: simple
filename: errors.log
maxBytes: 10485760 # 10MB
backupCount: 20
encoding: utf8
root:
level: INFO
handlers: [console, error_file_handler]
5.3. dictmentor (>= 0.11.0)
You can augment the configuration by extensions from the dictmentor
package.
Please see DictMentor for further reference.
The DictMentor
instance will be instantiated with the following code and thus the following extensions:
from dictmentor import DictMentor, ext
return DictMentor(
ext.Environment(fail_on_unset=True),
ext.ExternalResource(base_path=os.path.dirname(config_path)),
ext.ExternalYamlResource(base_path=os.path.dirname(config_path))
)
Example:
# Uses the dictmentor package to augment the configuration by dictmentor extensions.
# Make sure to export the environment variable to echo:
# export MESSAGE="Hello World"
- name: dictmentor
pull:
plugin: pnp.plugins.pull.simple.Repeat
args:
wait: 1
repeat: "{{env::MESSAGE}}"
push:
- external: echo.pull
- external: nop.pull
# Contents of echo.pull
plugin: pnp.plugins.push.simple.Echo
# Contents of nop.pull
plugin: pnp.plugins.push.simple.Nop
5.4. Advanced selector expressions (>= 0.12.0)
Instead of string-only selector expressions, you may now use complex dictionary and/or list constructs in your yaml
to define a selector expression. The configuration below will repeat {'hello': 'Hello', 'words': ['World', 'Moon', 'Mars']}
.
- name: selector
pull:
plugin: pnp.plugins.pull.simple.Repeat
args:
wait: 1
repeat: "Hello World Moon Mars"
push:
- plugin: pnp.plugins.push.simple.Echo
selector:
hello: payload.split(' ')[0]
words:
- payload.split(' ')[1]
- payload.split(' ')[2]
- payload.split(' ')[3]
Before the advanced selector feature your epxressions would have probably looked similiar to this:
dict(hello=payload.split(' ')[0], words=[payload.split(' ')[1], payload.split(' ')[2], payload.split(' ')[3]])
.
The first one is more readable, isn't it?
5.5. Docker images
make docker # OR: make docker-arm for raspberry image
# Mount the task and logging configuration when starting up the container
docker run -it --rm \
-v /path/to/pnp/config/01_hello_world.yaml:/config/config.yaml \
-v /path/to/logging/config/file.logging:/config/logging.yaml \
pnp:0.10.0
6. Plugins
6.1. pnp.plugins.pull.fs.FileSystemWatcher
Watches the given directory for changes like created, moved, modified and deleted files. If ignore_directories
is
set to False, then directories will be reported as well.
Per default will recursively report any file that is touched, changed or deleted in the given path. The
directory itself or subdirectories will be object to reporting too, if ignore_directories
is set to False.
Requires extra fswatcher
.
Arguments
path (str): The path to track for file / directory changes.
recursive (bool, optional): If set to True, any subfolders of the given path will be tracked too.
Default is True.
patterns (str or list, optional): Any file pattern (e.g. .txt or [.txt, *.md].
If set to None no filter is applied. Default is None.
ignore_patterns (str or list, optional): Any patterns to ignore (specify like argument patterns
).
If set to None, nothing will be ignored. Default is None.
ignore_directories (str, optional): If set to True will send events for directories when file change.
Default is False.
case_sensitive (bool, optional): If set to True, any pattern is case_sensitive, otherwise it is case insensitive.
Default is False.
events (str or list, optional): The events to track. One or multiple of 'moved', 'deleted', 'created'
and/or 'modified'. If set to None all events will be reported. Default is None.
load_file (bool, optional): If set to True the file contents will be loaded into the result. Default is False.
mode (str, optional): Open mode of the file (only necessary when load_file is True). Can be text, binary or auto
(guessing). Default is auto.
base64 (bool, optional): If set to True the loaded file contents will be converted to base64 (only applicable when
load_file is True). Argument mode
will be automatically set to 'binary'. Default is False.
defer_modified (float, optional): If set greater than 0, it will defer the sending of modified events for that
amount of time (seconds). There might be multiple flushes of a file before it is written completely to disk.
Without defer_modified each flush will raise a modified event. Default is 0.5.
Result
Example of an emitted message
{
'operation': 'modified',
'source': '/tmp/abc.txt',
'is_directory': False,
'destination': None, # Only non-None when operation = 'moved'
'file': { # Only present when load_file is True
'file_name': 'abc.txt',
'content': 'foo and bar',
'read_mode': 'text',
'base64': False
}
}
Examples
- name: file_watcher
pull:
plugin: pnp.plugins.pull.fs.FileSystemWatcher
args:
path: "/tmp"
ignore_directories: True
events: [created, deleted, modified]
load_file: False
push:
plugin: pnp.plugins.push.simple.Echo
6.2. pnp.plugins.pull.gpio.Watcher
Listens for low/high state changes on the configured gpio pins.
In more detail the plugin can raise events when one of the following situations occur:
- rising (high) of a gpio pin - multiple events may occur in a short period of time
- falling (low) of a gpio pin - multiple events may occur in a short period of time
- switch of gpio pin - will suppress multiple events a defined period of time (bounce time)
- motion of gpio pin - will raise the event
motion_on
if the pin rises and set a timer with a configurable amount of time. Any other gpio rising events will reset the timer. When the timer expires themotion_off
event is raised.
Requires extra gpio
.
Arguments
pins (list): The gpio pins to observe for state changes. Please see the examples section on how to configure it.
default (on of [rising, falling, switch, motion]: The default edge that is applied when not configured. Please see the examples section for further details.
Result
{
"gpio_pin": 17 # The gpio pin which state has changed
"event": rising # One of [rising, falling, switch, motion_on, motion_off]
}
Examples
- name: gpio
pull:
plugin: pnp.plugins.pull.gpio.Watcher
args:
default: rising
pins:
- 2 # No mode specified: Default mode (in this case 'rising')
- 2 # Duplicates will get ignored
- 3:rising # Equal to '3' (without explicit mode)
- 3:falling # Get the falling event for gpio pin 3 as well
- 4:switch # Uses some debouncing magic and emits only one rising event
- 5:switch(1000) # Specify debounce in millseconds (default is 500ms)
- 5:switch(500) # Duplicates - even when they have other arguments - will get ignored
- 7:motion # Uses some delay magic to emit only one motion on and one motion off event
- 9:motion(1m) # Specify delay (default is 30 seconds)
push:
- plugin: pnp.plugins.push.simple.Echo
6.3. pnp.plugins.pull.http.Server
Listens on the specified port
for requests to any endpoint.
Any data passed to the endpoint will be tried to be parsed to a dictionary (json). If this is not possible
the data will be passed as is. See sections Result
for specific payload and examples.
Remark: You will not able to make requests to the endpoint DELETE /_shutdown
because it is used internally.
Requires extra http-server
.
Arguments
port (int, optional): The port the rest server should listen to for requests. Default is 5000.
allowed_methods (str or list, optional): List of http methods that are allowed. Default is 'GET'.
server_impl (str, optional): Choose the implementation of the WSGI-Server (wraps the flask-app).
Possible values are: [flask, gevent]. flask
uses the internal Flask Development server. Not recommended for
production use. gevent
uses gevent. Default is gevent
.
Result
curl -X GET 'http://localhost:5000/resource/endpoint?foo=bar&bar=baz' --data '{"baz": "bar"}'
{
'endpoint': 'resource/endpoint,
'method': 'GET',
'query': {'foo': 'bar', 'bar': 'baz'},
'data': {'baz': 'bar'},
'is_json': True
}
curl -X GET 'http://localhost:5000/resource/endpoint' --data 'no json obviously'
{
'endpoint': 'resource/endpoint,
'method': 'GET',
'query': {},
'data': b'no json obviously',
'is_json': False
}
Examples
- name: rest
pull:
plugin: pnp.plugins.pull.http.Server
args:
port: 5000
allowed_methods: [GET, POST]
push:
plugin: pnp.plugins.push.simple.Echo
6.4. pnp.plugins.pull.monitor.Stats
Emits every interval
various metrics / statistics about the host system. Please see the 'Result' section for available metrics.
Result
{
'cpu_count': 4,
'cpu_freq': 700, # in Mhz
'cpu_use': 6.6, # in %
'cpu_temp': 52.6, # in °C (might not be available on all systems, e.g. MacOS)
'memory_use': 56.0, # in %
'swap_use': 23.2, # in %
'disk_use': 69.8, # in % (of your root)
'load_1m': 1.81591796875, # CPU queue length last minute
'load_5m': 2.06689453125, # CPU queue length last 5 minutes
'load_15m': 2.15478515625 # CPU queue length last 15 minutes
}
Examples
- name: stats
pull:
plugin: pnp.plugins.pull.monitor.Stats
args:
interval: 10s
instant_run: True
push:
plugin: pnp.plugins.push.simple.Echo
6.5. pnp.plugins.pull.mqtt.Subscribe
Pulls messages from the specified topic from the given mosquitto mqtt broker (identified by host and port).
Arguments
host (str): Host where the mosquitto broker is running.
port (int): Port where the mosquitto broker is listening.
topic (str): Topic to pull messages from.
You can listen to multiple topics by using the #-wildcard (e.g. test/#
will listen to all topics underneath test).
All arguments can be automatically injected via environment variables with MQTT
prefix (e.g. MQTT_HOST).
Result
The emitted message will look like this:
{
'topic': 'test/device/device1',
'levels': ['test', 'device', 'device1']
'payload': 'The actual event message'
}
Examples
- name: mqtt
pull:
plugin: pnp.plugins.pull.mqtt.Subscribe
args:
host: localhost
port: 1883
topic: test/#
push:
plugin: pnp.plugins.push.simple.Echo
6.6. pnp.plugins.pull.sensor.DHT
Periodically polls a dht11 or dht22 (aka am2302) for temperature and humidity readings.
Polling interval is controlled by interval
.
Requires extra dht
.
Arguments
device (str, optional): The device to poll (one of dht22, dht11, am2302). Default is 'dht22'.
data_gpio (int, optional): The data gpio port where the device operates on. Default is 17.
humidity_offset (float, optional): Positive/Negative offset for humidity. Default is 0.0.
temp_offset (float, optional): Positive/Negative offset for temperature. Default is 0.0.
Result
{
"humidity": 65.4 # in %
"temperature": 23.7 # in celsius
}
Examples
- name: dht
pull:
plugin: pnp.plugins.pull.sensor.DHT
args:
device: dht22 # Connect to a dht22
data_gpio: 17 # DHT is connected to gpio port 17
interval: 5m # Polls the readings every 5 minutes
humidity_offset: -5.0 # Subtracts 5% from the humidity reading
temp_offset: 1.0 # Adds 1 °C to the temperature reading
push:
- plugin: pnp.plugins.push.simple.Echo
selector: payload.temperature # Temperature reading
- plugin: pnp.plugins.push.simple.Echo
selector: payload.humidity # Humidity reading
6.7. pnp.plugins.pull.sensor.OpenWeather
Periodically polls weather data from the OpenWeatherMap
api.
Arguments
api_key (str): The api_key you got from OpenWeatherMap after registration.
lat (float): Latitude. If you pass lat
, you have to pass lon
as well.
lon (float): Longitude. If you pass lon
, you have to pass lat
as well.
city_name (str): The name of your city. To minimize ambiguity use lat/lon or your country as a suffix,
e.g. London,GB.
units (str on of (metric, imperial, kelvin)): Specify units for temperature and speed.
imperial = fahrenheit + miles/hour, metric = celsius + m/secs, kelvin = kelvin + m/secs. Default is metric.
tz (str, optional): Time zone to use for current time and last updated time. Default is your local timezone.
Remark: You have to pass whether city_name
or lat/lon
.
Result
{
"temperature": 13.03,
"pressure": 1021,
"humidity": 62,
"cloudiness": 40,
"wind": {
"speed": 9.3,
"deg": 300
},
"poll_dts": "2018-10-03T15:41:32.156930+02:00",
"last_updated_dts": "2018-10-03T15:20:00+02:00",
"raw": {
"coord": {
"lon": 10,
"lat": 53.55
},
"weather": [{
"id": 521,
"main": "Rain",
"description": "shower rain",
"icon": "09d"
}],
"base": "stations",
"main": {
"temp": 13.03,
"pressure": 1021,
"humidity": 62,
"temp_min": 12,
"temp_max": 14
},
"visibility": 10000,
"wind": {
"speed": 9.3,
"deg": 300
},
"clouds": {
"all": 40
},
"dt": 1538572800,
"sys": {
"type": 1,
"id": 4883,
"message": 0.0202,
"country": "DE",
"sunrise": 1538544356,
"sunset": 1538585449
},
"id": 2911298,
"name": "Hamburg",
"cod": 200
}
}
You can consult the specification https://openweathermap.org/current#parameter to checkout the documentation about the meaning of individual fields.
Examples
### Make sure you export your api key with: `export OPENWEATHER_API_KEY=<your_api_key>`
- name: openweather
pull:
plugin: pnp.plugins.pull.sensor.OpenWeather
args:
city_name: "Hamburg,DE" # Alternative: pass lat and lon
# lon: 10
# lat: 53.55
units: metric # imperial (fahrenheit + miles/hour), metric (celsius + m/secs), kelvin (kelvin + m/secs)
instant_run: True
# tz: GMT
push:
plugin: pnp.plugins.push.simple.Echo
6.8. pnp.plugins.pull.simple.Count
Emits every wait
seconds a counting value which runs from from_cnt
to to_cnt
.
If to_cnt
is None the counter will count to infinity.
Arguments
wait (int): Wait the amount of seconds before emitting the next counter.
from_cnt (int): Starting value of the counter.
to_cnt (int, optional): End value of the counter. If not passed set to "infinity" (precise: int.max).
Result
Counter value (int).
Examples
- name: count
pull:
plugin: pnp.plugins.pull.simple.Count
args:
wait: 1
from_cnt: 1
to_cnt: 10
push:
plugin: pnp.plugins.push.simple.Echo
6.9. pnp.plugins.pull.simple.Repeat
Emits every wait
seconds the same repeat
.
Arguments
wait (int): Wait the amount of seconds before emitting the next repeat.
repeat (any): The object to emit.
Result
Emits the repeat
-object as it is.
Examples
- name: repeat
pull:
plugin: pnp.plugins.pull.simple.Repeat
args:
repeat: "Hello World" # Repeats 'Hello World'
wait: 1 # Every second
push:
plugin: pnp.plugins.push.simple.Echo
6.10. pnp.plugins.pull.zway.ZwayPoll
Pulls the specified json content from the zway rest api. The content is specified by the url, e.g.
http://<host>:8083/ZWaveAPI/Run/devices
will pull all devices and serve the result as a json.
Specify the polling interval by setting the argument interval
. User / password combination is required when
your api is protected against guest access (by default it is).
Use multiple pushes and the related selectors to extract the required content like temperature readings (see the examples section for guidance).
Arguments
url (str): The url to poll periodically.
user (str): Authentication user name.
password (str): Authentication password.
interval (polling literal, optional): Polling interval (default: 1m).
All arguments (url
, user
and password
) can be automatically injected via environment variables.
- ZWAY_URL
- ZWAY_USER
- ZWAY_PASSWORD
Result
Emits the content of the fetched url as it is.
Examples
### Please make sure to adjust url and device ids
### Username and Password are injected from environment variables:
### export ZWAY_USER=admin
### export ZWAY_PASSWORD=secret_one
- name: zway
pull:
plugin: pnp.plugins.pull.zway.ZwayPoll
args:
url: "http://smarthome:8083/ZWaveAPI/Run/devices"
interval: 5s
push:
- plugin: pnp.plugins.push.simple.Echo
# Temperature of fibaro motion sensor
# You can access the returned json like you would inquire the zway-api
selector: payload[19].instances[0].commandClasses[49].data[1].val.value
- plugin: pnp.plugins.push.simple.Echo
# Luminiscence of fibaro motion sensor
selector: payload[19].instances[0].commandClasses[49].data[3].val.value
Appendix
Below are some common selector examples to fetch various metrics from various devices
Fibaro Motion Sensor
- Temperature
payload[deviceid].instances[0].commandClasses[49].data[1].val.value
- Luminescence
payload[deviceid].instances[0].commandClasses[49].data[3].val.value
fibaro Wallplug
- Meter
payload[deviceid].instances[0].commandClasses[50].data[0].val.value
Thermostat (Danfoss / other should work as well)
- Setpoint
payload[deviceid].instances[0].commandClasses[67].data[1].val.value
Battery operated devices
- Battery level
payload[deviceid].instances[0].commandClasses[128].data.last.value
6.11. pnp.plugins.pull.zway.ZwayReceiver
Setups a http server to process incoming GET-requests from the Zway-App HttpGet
.
Arguments
url_format (str): The url_format that is configured in your HttpGet App. If you configured
http://<ip>:<port>/set?device=%DEVICE%&state=%VALUE%
(default of the App), you basically have to copy the path
component set?device=%DEVICE%&state=%VALUE%
to be your url_format
.
mode ([mapping, auto, both]): If set to mapping
(default) you should provide the device_mapping
to manually map your virtual devices.
If set to auto
the plugin will try to determine the device_id, command class, mode and the type on it's own. If set to both
the plugin
will first try the device_mapping
and then perform the auto-magic.
device_mapping (Or(Dict[Str, Str], Dict[Str, Dict]), optional): A mapping to map the somewhat cryptic virtual device names to
human readable ones. Default is None, which means that no mapping will be performed. Two ways possible:
- Ordinary mapping from virtual device name -> alias.
- Enhanced mapping from virtual device name to dictionary with additional properties. One property has to be alias.
ignore_unknown_devices (bool, optional): If set to True all incoming requests that are associated with an device that is not part of the mapping or - when mode = [auto, both] - cannot be auto mapped will be ignored. Default is False.
Additionally the component will accept any arguments that pnp.plugins.pull.http.Server
would accept.
Result
Given the url_format set?%DEVICE%&value=%VALUE%
, the url http://<ip>:<port>/set?vdevice1&value=5.5
and
the device_mapping vdevice1 -> alias of vdevice1
the emitted message will look like this:
{
'device_name': 'alias of vdevice1',
'raw_device': 'vdevice1'
'value': '5.5',
'props': {}
}
When mode
is auto
or both
the plugin will try to determine the device id and the type of the virtual device on it's
own. Given the virtual device name ZWayVDev_zway_7-0-48-1
and the value of on
will produce the following:
{
'device_name': '7',
'raw_device': 'ZWayVDev_zway_7-0-48-1',
'value': 'on'
'props': {
'command_class': '48',
'mode': '1',
'type': 'motion'
}
}
Examples
- name: zway_receiver
pull:
plugin: pnp.plugins.pull.zway.ZwayReceiver
args:
port: 5000
mode: mapping # mapping, auto or both
device_mapping:
vdevice1: # Props = {type: motion}
alias: dev1
type: motion
vdevice2: # Props = {type: switch, other_prop: foo}
alias: dev2
type: switch
other_prop: foo
vdevice3: dev3 # props == {}
url_format: "%DEVICE%?value=%VALUE%"
ignore_unknown_devices: False
push:
- plugin: pnp.plugins.push.simple.Echo
selector: "'Got value {} from device {} ({}) with props {}'.format(data.value, data.device_name, data.raw_device, data.props)"
6.12. pnp.plugins.push.fs.FileDump
This push dumps the given payload
to a file to the specified directory
.
If argument file_name
is None, a name will be generated based on the current datetime (%Y%m%d-%H%M%S).
If file_name
is not passed (or None) you should pass extension
to specify the extension of the generated
file name.
Argument binary_mode
controls whether the dump is binary (mode=wb) or text (mode=w).
Arguments
directory (str, optional): The target directory to store the dumps. Default is '.' (current directory).
file_name (str, optional): The name of the file to dump. If set to None a file name will be automatically
generated. You can specify the file_name via the envelope, too. Envelope will override init file name.
Default is None.
extension (str, optional): The extension to use when the file name is automatically generated. Can be overridden by
envelope. Default is '.dump'.
binary_mode (bool, optional): If set to True the file will be written in binary mode ('wb');
otherwise in text mode ('w'). Default is False.
Result
Will return an absolute path to the file created.
Examples
- name: file_dump
pull:
plugin: pnp.plugins.pull.simple.Repeat
args:
repeat: "Hello World"
push:
plugin: pnp.plugins.push.fs.FileDump
args:
directory: "/tmp"
file_name: null # Auto-generated file (timestamp)
extension: ".txt" # Extension of auto-generated file
binary_mode: False # text mode
deps:
- plugin: pnp.plugins.push.simple.Echo
- name: file_dump
pull:
plugin: pnp.plugins.pull.simple.Repeat
args:
repeat: "Hello World"
push:
plugin: pnp.plugins.push.fs.FileDump
# Override `file_name` and `extension` via envelope.
# Instead of an auto generated file, the file '/tmp/hello-world.hello' will be dumped.
selector: '{"payload": payload, "file_name": "hello-world", "extension": ".hello"}'
args:
directory: "/tmp"
file_name: null # Auto-generated file (timestamp)
extension: ".txt" # Extension of auto-generated file
binary_mode: False # text mode
deps:
- plugin: pnp.plugins.push.simple.Echo
6.13. pnp.plugins.push.http.Call
Makes a request to a http resource.
Arguments
url (str): Request url. Can be overridden via envelope.
method (str, optional): The http method to use for the request. Must be a valid http method (GET, POST, ...).
Default is 'GET'. Can be overridden via envelope.
fail_on_error (bool, optional): If True the push will fail on a http status code <> 2xx. This leads to an error
message recorded into the logs and no further execution of any dependencies. Default is False. Can be overridden
by the envelope.
provide_response (bool, optional): If True the push will not return the payload as it is, but instead provide the
response status_code, fetched url content and a flag if the url content is a json response. This is useful for
other push instances in the dependency chain. Default is False.
Result
Will return the payload as it is for easy chaining of dependencies.
If provide_response
is True the push will return a dictionary that looks like this:
{
"status_code": 200,
"data": "fetched url content",
"is_json": False
}
Please note that this structure will be interpreted as an envelope with the keys status_code
and is_json
along with
the payload 'fetched url content' by other push instances in the dependency chain.
Examples
### Simple example calling the built-in rest server
### Oscillates between http method GET and POST. Depending on the fact if the counter is even or not.
- name: http_call
pull:
plugin: pnp.plugins.pull.simple.Count
args:
wait: 5
push:
plugin: pnp.plugins.push.http.Call
selector: "dict(data=dict(counter=payload), method='POST' if int(payload) % 2 == 0 else 'GET')"
args:
url: http://localhost:5000/
- name: rest_server
pull:
plugin: pnp.plugins.pull.http.Server
args:
port: 5000
allowed_methods:
- GET
- POST
push:
plugin: pnp.plugins.push.simple.Echo
### Demonstrates the use of `provide_response` set to True.
### Call will return a response object to dependent push instances.
- name: http_call
pull:
plugin: pnp.plugins.pull.simple.Count
args:
wait: 5
push:
plugin: pnp.plugins.push.http.Call
args:
url: http://localhost:5000/
provide_response: True
deps:
plugin: pnp.plugins.push.simple.Echo
- name: rest_server
pull:
plugin: pnp.plugins.pull.http.Server
args:
port: 5000
allowed_methods:
- GET
push:
plugin: pnp.plugins.push.simple.Nop
6.14. pnp.plugins.push.ml.FaceR
FaceR (short one for face recognition) tags known faces in images. Output is the image with all faces tagged whether
with the known name or an unknown_label
. Default for unknown ones is 'Unknown'.
Known faces can be ingested either by a directory of known faces (known_faces_dir
) or by mapping of known_faces
(dictionary: name -> [list of face files]).
The payload
passed to the push
method is expected to be a valid byte array that represents an image in memory.
Arguments
known_faces (dict<str, file_path as str>, optional): Mapping of a person's name to a list of images that contain
the person's face. Default is None.
known_faces_dir (str, optional): A directory containing images with known persons (file_name -> person's name).
Default is None.
unknown_label (str, optional): Tag label of unknown faces. Default is 'Unknown'.
You have to specify either known_faces
or known_faces_dir
. If both are unsupplied the push will fail.
Result
Will return a dictionary that contains the bytes of the tagged image (key tagged_image
) and metadata (no_of_faces
,
known_faces
)
{
'tagged_image': <bytes of tagged image>
'no_of_faces': 2
'known_faces': ['obama']
}
Examples
- name: faceR
pull:
plugin: pnp.plugins.pull.fs.FileSystemWatcher
args:
path: "/tmp/camera"
recursive: True
patterns: "*.jpg"
ignore_directories: True
case_sensitive: False
events: [created]
load_file: True
mode: binary
base64: False
push:
plugin: pnp.plugins.push.ml.FaceR
args:
known_faces_dir: "/tmp/faces"
unknown_label: "don't know him"
6.15. pnp.plugins.push.mqtt.Publish
Will push the given payload
to a mqtt broker (in this case mosquitto).
The broker is specified by host
and port
. In addition a topic needs to be specified were the payload
is pushed to (e.g. home/living/thermostat).
The payload
will be pushed as it is. No transformation is applied. If you need to some transformations, use the
selector.
Arguments
host (str): The host where the mosquitto broker is running.
port (int, optional): The port where the mosquitto broker is listening. Default is 1883.
topic (str, optional): The topic to subscribe to. If set to None the envelope of the
payload has to contain a 'topic' key or the push will fail (default is None). If both exists
the topic from the envelope will overrule the init one.
retain (bool, optional): If set to True will mark the message as retained. Default is False.
See the mosquitto man page for further guidance
https://mosquitto.org/man/mqtt-7.html.
multi (bool, optional): If set to True the payload is expected to be a dictionary. Each item of that dictionary will
be sent individually to the broker. The key of the item will be appended to the configured topic. The value of the item
is the actual payload. Default is False.
Result
For chaining of pushes the payload is simply returned as is.
Examples
- name: mqtt
pull:
plugin: pnp.plugins.pull.simple.Count
push:
# Will push the counter to the 'home/counter/state' topic
plugin: pnp.plugins.push.mqtt.Publish
args:
host: localhost
topic: home/counter/state
port: 1883
retain: True
- name: mqtt
pull:
plugin: pnp.plugins.pull.simple.Count
push:
plugin: pnp.plugins.push.mqtt.Publish
# Lets override the topic via envelope mechanism
# Will publish even counts on topic 'even' and uneven counts on 'uneven'
selector: "{'data': data, 'topic': 'even' if int(data) % 2 == 0 else 'uneven'}"
args:
host: localhost
port: 1883
- name: mqtt
pull:
# Periodically gets metrics about your system
plugin: pnp.plugins.pull.monitor.Stats
args:
instant_run: True
interval: 10s
push:
# Push them to the mqtt
plugin: pnp.plugins.push.mqtt.Publish
args:
host: localhost
topic: devices/localhost/
port: 1883
retain: True
# Each item of the payload-dict (cpu_count, cpu_usage, ...) will be pushed to the broker as multiple items.
# The key of the item will be appended to the topic, e.g. `devices/localhost/cpu_count`.
# The value of the item is the actual payload.
multi: True
6.16. pnp.plugins.push.notify.Pushbullet
Sends a message to the Pushbullet service. The type of the message will guessed:
push_link
for a single http linkpush_file
if the link is directed to a file (mimetype will be guessed)push_note
for everything else (converted tostr
)
Requires extra pushbullet
.
Arguments
api_key (str): The api key to your pushbullet account.
title (str, optional): The title to use for your messages. Defaults to pnp
Result
Will return the payload as it is for easy chaining of dependencies.
Examples
### Make sure that you provided PUSHBULETT_API_KEY as an environment variable
- name: pushbullet
pull:
plugin: pnp.plugins.pull.fs.FileSystemWatcher
args:
path: "/tmp"
ignore_directories: True
events:
- created
load_file: False
push:
plugin: pnp.plugins.push.notify.Pushbullet
args:
title: "Watcher"
selector: "'New file: {}'.format(data.source)"
6.17. pnp.plugins.push.simple.Echo
Simply log the passed payload to the default logging instance.
Arguments
None.
Result
Will return the payload as it is for easy chaining of dependencies.
Examples
- name: count
pull:
plugin: pnp.plugins.pull.simple.Count
args:
wait: 1
from_cnt: 1
to_cnt: 10
push:
plugin: pnp.plugins.push.simple.Echo
6.18. pnp.plugins.push.simple.Execute
Executes a command with given arguments in a shell of the operating system.
Will return the exit code of the command and optionally the output from stdout and stderr.
Arguments
command (str): The command to execute.
args (str or iterable, optional): The arguments to pass to the command. Default is no arguments.
cwd (str, optional): Specifies where to execute the command (working directory). Default is current working directory.
timeout (duration literal, optional): Specifies how long the worker should wait for the command to finish.
capture (bool, optional): If True stdout and stderr output is captured, otherwise not.
Result
Returns a dictionary that contains the return_code
and optionally the output from stdout
and stderr
whether
capture
is set or not. The output is a list of lines.
{
'return_code': 0
'stdout': ["hello", "dude!"]
'stderr': []
}
Examples
- name: execute
pull:
plugin: pnp.plugins.pull.simple.Count
args:
wait: 1
from_cnt: 1
to_cnt: 10
push:
plugin: pnp.plugins.push.simple.Execute
args:
command: date # The command to execute
args: # Argument passed to the command
- "-v"
- "-1d"
- "+%Y-%m-%d"
timeout: 2s
cwd: # None -> current directory
capture: True # Capture stdout and stderr
deps:
- plugin: pnp.plugins.push.simple.Echo
6.19. pnp.plugins.push.storage.Dropbox
Uploads provided file to the specified dropbox account.
Arguments
api_key (str): The api key to your dropbox account/app.
target_file_name (str, optional): The file path on the server where to upload the file to.
If not specified you have to specify this argument during push time by setting it in the envelope.
create_shared_link (bool, optional): If set to True, the push will create a publicly available link to your uploaded file. Default is True
.
Requires extra dropbox
.
Result
Returns a dictionary that contains metadata information about your uploaded file. If you uploaded a file named 42.txt
,
your result will be similiar to the one below:
{
"name": "42.txt",
"id": "HkdashdasdOOOOOadss",
"content_hash": "aljdhfjdahfafuhu489",
"size": 42,
"path": "/42.txt",
"shared_link": "http://someserver/tosomestuff/asdasd?dl=1",
"raw_link": "http://someserver/tosomestuff/asdasd?raw=1"
}
shared_link
is the one that is publicly available (if you know the link). Same for raw_link
, but this link will return
the raw file (without the dropbox overhead). Both are None
if create_shared_link
is set to False
.
Examples
### Make sure that you provided DROPBOX_API_KEY as an environment variable
- name: dropbox
pull:
plugin: pnp.plugins.pull.fs.FileSystemWatcher
args:
path: "/tmp"
ignore_directories: True
events:
- created
- modified
load_file: False
push:
- plugin: pnp.plugins.push.storage.Dropbox
args:
create_shared_link: True # Create a publicly available link
selector:
data: data.source # Absolute path to file
target_file_name: basename(data.source) # File name only
6.20. pnp.plugins.push.timedb.InfluxPush
Pushes the given payload
to an influx database using the line protocol
.
You have to specify host
, port
, user
, password
and the database
.
The protocol
is basically a string that will be augmented at push-time with data from the payload.
E.g. {payload.metric},room={payload.location} value={payload.value} assumes that payload contains metric, location
and value.
See https://docs.influxdata.com/influxdb/v1.5/write_protocols/line_protocol_tutorial/
Arguments
host (str): The host where the influxdb is running.
port (int): The port where the influxdb service is listening on.
user (str): Username to use for authentication.
password (str): Related password.
database (str): The database to write to.
protocol (str): Line protocol template (augmented with payload-data).
All arguments can be automatically injected via environment variables with INFLUX
prefix (e.g. INFLUX_HOST
).
Result
For the ability to chain multiple pushes together the payload is simply returned as is.
Examples
- name: mqtt_pull
pull:
plugin: pnp.plugins.pull.mqtt.MQTTPull
args:
host: mqtt
topic: home/#
push:
plugin: pnp.plugins.push.timedb.InfluxPush
selector: "{'data': payload}"
args:
host: influxdb
port: 8086
user: root
password: secret
database: home
protocol: "{payload.levels[2]},room={payload.levels[1]} {payload.levels[3]}={payload.payload}"
7. Changelog
We cannot ensure not to introduce any breaking changes to interfaces / behaviour. This might occur every commit whether it is intended or by accident. Nevertheless we try to list breaking changes in the changelog that we are aware of. You are encouraged to specify explicitly the version in your dependency tools, e.g.:
pip install pnp==0.10.0
0.12.0
- Adds additional argument
multi
(default False) topush.mqtt.MQTTPush
to send multiple messages to the broker if the payload is a dictionary (see plugin docs for reference) - Adds plugin
pull.monitor.Stats
to periodically emit stats about the host system - Adds plugin
push.notify.Pushbullet
to send message via thepushbullet
service - Adds plugin
push.storage.Dropbox
to upload files to adropbox
account/app - Adds feature to use complex lists and/or dictionary constructs in selector expressions
- Adds plugin
pull.gpio.Watcher
(extragpio
) to watch gpio pins for state changes. Only works on raspberry - Adds plugin
push.simple.Execute
to run commands in a shell - Adds extra
http-server
to optionally installflask
andgevent
when needed - Adds utility method to check for installed extras
- Adds
-v | --verbose
flag to pnp runner to switch logging level toDEBUG
. No matter what...
0.11.3
- Adds auto-mapping magic to the
pull.zway.ZwayReceiver
. - Adds humidity and temperature offset to dht
0.11.2
- Fixes error catching of
run_pending
inPolling
base class
0.11.1
- Fixes resolution of logging configuration on startup
0.11.0
- Introduces the pull.zway.ZwayReceiver and pull.sensor.OpenWeather component
- Introduces logging configurations. Integrates dictmentor package to augment configuration
0.10.0
- Introduces engines. You are not enforced to explicitly use one and backward compatibility with legacy configs is given (actually the example configs work as they did before the change). So there shouldn't be any breaking change.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.