Sometimes you want to trigger events with a click on a special touchpad or device mounted somewhere. This could be in a factory, on a door or at your desk. For me it's a small device on my desk that I can use to trigger events. I have this running every 15 seconds looking for gestures. I could put it in an infinite loop, but Python and RPI could leak some memory. We are very constrained, so I am trying to keep it a little more minimal. I keep my batch duration to 15 seconds and my run schedule to 15 seconds. Let's Build A Simple NiFi Flow to Receive The JSON Data and React To It In my RouteOnContent, I just look for the word "center". I have thought of many options for running sql, doing a backup, etc.. Build The MiniFi Flow in NiFi Downloaded minifi 0.2.0 and minifi 0.2.0 toolkit (you can use newer, but make sure you install the same version on the device you are going to move your config.yml to). minifi-toolkit-0.2.0/bin/config.sh transform minififlick.xml config.yml Then SCP that config.yml and the minifi-*.zip to your device. Unzip (or tar-cvf it). Then you can run. This requires Java 8 JDK installed and running on your machine. The Oracle version runs best on RPI. Let's Install and Run MiniFi cd /opt/demo/minifi-0.2.0
bin/minifi.sh install
bin/minifi.sh start Example Message {"flick": "center", "host": "herrflick", "ipaddress": "192.168.1.185", "ts": "2017-08-14 21:19:21", "cputemp": 47.0} The important data is flick which is the gesture made (click, tap, movement, double click, etc...) The other data is one's I always like to grab for devices (hostname, IP Address, timestamp and CPU temperature). Since flick = center, we send a Slack message We could do just about anything you want in the flow based on the trigger. Start backups, send system information, anything you want to trigger on demand. or Source Code: #!/usr/bin/env python
# -*- coding: <utf-8> -*-
# Based on
#Author: Callum Pritchard, Joachim Hummel
#Project Name: Flick 3D Gesture
#Project Description: Sending Flick 3D Gesture sensor data to mqtt
#Version Number: 0.1
#Date: 15/6/17
#Release State: Alpha testing
#Changes: Created
import time
import colorsys
import os
import json
import sys, socket
import subprocess
import time
import datetime
from time import sleep
from time import gmtime, strftime
import signal
import flicklib
import time
from curses import wrapper
some_value = 5000
flicktxt = ''
#### Initialization
# yyyy-mm-dd hh:mm:ss
currenttime= strftime("%Y-%m-%d %H:%M:%S",gmtime())
external_IP_and_port = ('198.41.0.4', 53) # a.root-servers.net
socket_family = socket.AF_INET
host = os.uname()[1]
def getCPUtemperature():
res = os.popen('vcgencmd measure_temp').readline()
return(res.replace("temp=","").replace("'C\n",""))
def IP_address():
try:
s = socket.socket(socket_family, socket.SOCK_DGRAM)
s.connect(external_IP_and_port)
answer = s.getsockname()
s.close()
return answer[0] if answer else None
except socket.error:
return None
def message(publisher, value):
print value
@flicklib.move()
def move(x, y, z):
global xyztxt
xyztxt = '{:5.3f} {:5.3f} {:5.3f}'.format(x,y,z)
@flicklib.flick()
def flick(start,finish):
global flicktxt
flicktxt = 'FLICK-' + start[0].upper() + finish[0].upper()
message('flick',flicktxt)
@flicklib.airwheel()
def spinny(delta):
global some_value
global airwheeltxt
global flicktxt
some_value += delta
if some_value < 0:
some_value = 0
if some_value > 10000:
some_value = 10000
airwheeltxt = str(some_value/100)
flicktxt = airwheeltxt
@flicklib.double_tap()
def doubletap(position):
global doubletaptxt
global flicktxt
doubletaptxt = position
flicktxt = doubletaptxt
@flicklib.tap()
def tap(position):
global taptxt
global flicktxt
taptxt = position
flicktxt = taptxt
@flicklib.touch()
def touch(position):
global touchtxt
global flicktxt
touchtxt = position
flicktxt = touchtxt
def main():
global xyztxt
global flicktxt
global airwheeltxt
global touchtxt
global taptxt
global doubletaptxt
flickcount = 0
airwheeltxt = ''
airwheelcount = 0
touchtxt = ''
touchcount = 0
taptxt = ''
tapcount = 0
doubletaptxt = ''
doubletapcount = 0
time.sleep(0.1)
while flickcount < 100:
if (flicktxt != "") :
flickcount += 100
cpuTemp=int(float(getCPUtemperature()))
ipaddress = IP_address()
row = { 'ts': currenttime, 'host': host, 'cputemp': round(cpuTemp,2), 'ipaddress': ipaddress, 'flick': flicktxt }
json_string = json.dumps(row)
print(json_string)
sys.exit()
time.sleep(0.1)
flickcount += 1
main() See: https://github.com/tspannhw/rpi-rainbowhat References:
https://github.com/PiSupply/Flick.git Apps https://www.pi-supply.com/make/flick-quick-start-faq/ https://github.com/unixweb/Flick https://github.com/unixweb/Flick/blob/master/Flick.py https://github.com/tspannhw/rpi-rainbowhat https://github.com/tspannhw/rpi-rainbowhat/blob/master/minifi.py https://github.com/tspannhw/rpi-sensehat-mqtt-nifi https://github.com/tspannhw/rpi-sensehat-minifi-python https://github.com/tspannhw/rpi-flickhat-minifi/tree/master https://nifi.apache.org/minifi/getting-started.html
... View more
