Use Case: Ingesting sensors, images, voice and video from moving vehicles and running deep learning in the running vehicle. Transporting data and messages to remote data centers via Apache MiniFi and NiFi over Secure S2S HTTPS. Background: NVidia Jetson TX1 is a specialized developer kit for running a powerful GPU as an embedded device for robots, UAV and specialized platforms. I envision it's usage in field trucks for intermodal, utilities, telecommunications, delivery services, government and other industries with field vehicles. Installation and Setup You will need a workstation running Ubuntu 16 with enough disk space and network access. This will be to download all the software and push it over a network to your NVidia Jetson TX1. You can download Ubuntu here (https://www.ubuntu.com/download/desktop). Fortunately I had a MiniPC with 4GB of RAM that reformatted with Ubuntu to be the host PC to build my Jetson. You cannot run this from a Mac or Windows machine. You will need a monitor, mouse and keyboard for your host machine and a set for your NVIdia Jetson. First step boot to your NVidia Jetson and set up WiFi networking and make sure your monitor, keyboards and mouse work. Make sure you download the latest NVidia JetPack on your host Ubuntu machine https://developer.nvidia.com/embedded/jetpack The one I used was JetPack 3.1 and that included: 64-bit Ubuntu 16.04 cuDNN 6.0 TensorRT 2.1 CUDA 8.0. Initial login: ubuntu/ubuntu After installation it will be nvidia/nvidia. Please change that password, security is important and this GPU could do some serious bitcoin mining. sudo su apt update apt-get install git zip unzip autoconf automake libtool curl zlib1g-dev maven swig bzip2 apt-get purge libopencv4tegra-dev libopencv4tegra apt-get purge libopencv4tegra-repo apt-get update apt-get install build-essential apt-get install cmake git libgtk2.0-dev pkg-config libavcodec-dev libavformat-dev libswscale-dev apt-get install libgstreamer1.0-dev libgstreamer-plugins-base1.0-dev apt-get install python2.7-dev apt-get install python-dev python-numpy libtbb2 libtbb-dev libjpeg-dev libpng-dev libtiff-dev libjasper-dev libdc1394-22-dev apt-get install libgtkglext1 libgtkglext1-dev apt-get install qtbase5-dev apt-get install libv4l-dev v4l-utils qv4l2 v4l2ucp cd $HOME/NVIDIA-INSTALL ./installer.sh Downloaded and run NVidia Jetson TX1 JetPack from host Ubuntu computer ./JetPack-L4T-3.1-linux-x64.run. This will run on the host server for probably an hour and require networking connection between the two and a few reboots. I added a 64GB SD Card as the space on the Jetson is tiny. I would recommend adding a big SATA hard drive. umount /dev/sdb1 mount -o umask=000 -t vfat /dev/sdb1 /media/ Turn on the fan on the Jetson, echo 255 > /sys/kernel/debug/tegra_fan/target_pwm. Download MiniFi https://nifi.apache.org/minifi/download.html or https://hortonworks.com/downloads/#dataflow. You will need to install JDK 8. sudo add-apt-repository ppa:webupd8team/java sudo apt update sudo apt install oracle-java8-installer -y download minifi-0.2.0-bin.zip unzip *.zip bin/minifi.sh start In the next part, we will classify images. Part 2: Classifying Images with ImageNet Part 3: Detecting Faces in Images Part 4: Ingesting with MiniFi and NiFi Shell Call Example /media/nvidia/96ed93f9-7c40-4999-85ba-3eb24262d0a5/jetson-inference-master/build/aarch64/bin/detectnet-console pic-007.png outputface7.png facenet Source Code https://github.com/tspannhw/jetsontx1-TensorRT References http://elinux.org/Jetson/Computer_Vision_Performance#Hardware_Acceleration_of_OpenCV https://github.com/dusty-nv/jetson-inference#system-setup https://github.com/dusty-nv/jetson-inference/blob/master/docs/deep-learning.md https://github.com/NVIDIA/DIGITS/tree/master/examples/semantic-segmentation https://developer.nvidia.com/tensorrt https://community.hortonworks.com/articles/130814/sensors-and-image-capture-and-deep-learning-analys.html https://github.com/dusty-nv/jetson-inference/blob/master/detectnet-console/detectnet-console.cpp http://elinux.org/Jetson_TX1#Jetson_TX1_Module http://www.jetsonhacks.com/2016/12/30/tensorflow-nvidia-jetson-tx1-development-kit/ https://developer.nvidia.com/embedded/learn/tutorials#collapseOne http://www.jetsonhacks.com/2016/12/30/install-tensorflow-on-nvidia-jetson-tx1-development-kit/ http://www.nvidia.com/object/JetsonTX1DeveloperKitSE.html? http://www.nvidia.com/object/embedded-systems-dev-kits-modules.html https://github.com/jetsonhacks/installTensorFlowTX1/blob/master/scripts/installDependencies.sh https://github.com/dusty-nv/jetson-inference#system-setup http://docs.nvidia.com/jetpack-l4t/index.html#developertools/mobile/jetpack/l4t/3.0/jetpack_l4t_install.htm https://github.com/dusty-nv/jetson-inference/blob/master/data/networks/ilsvrc12_synset_words.txt https://github.com/tspannhw/rpi-rainbowhat/blob/master/minifi.py https://community.hortonworks.com/articles/130814/sensors-and-image-capture-and-deep-learning-analys.html https://nifi.apache.org/minifi/system-admin-guide.html ... View more

The MiniFi flow executes two scripts: one to call TensorFlow Python that captures an OpenCV Raspberry Pi Camera and runs Inception on it. That message is formatted as JSON and sent on. The second script reads GPS values from a USB GSP sensor and outputs JSON. Get File reads the Pi Camera image produced by the ClassifyImages process. Cleanup logs is a standalone timed script that cleans up old logs on the Raspberry Pi. Using InferredAvroSchema I created a schema for the GPS unit and stored it in the Hortonworks Schema Registry. This is the provenance event for a typical GPS message sent. You can see what shell script we ran and from what host. In Apache NiFi we process the message, routing to the correct place, setting a schema and querying it for a latitude. Then we convert the AVRO record to ORC to save as a Hive table. MiniFi requires we change the NiFi created template to a configuration file via the command-line MiniFi Toolkit. minifi-toolkit-0.2.0/bin/config.sh transform gpstensorflowpiminifi2.xml config.yml scp config.yml pi@192.168.1.167:/opt/demo/minifi/conf/ ./gpsrun.sh {"ipaddress": "192.168.1.167", "utc": "2017-08-21T20:00:06.000Z", "epx": "10.301", "epv": "50.6", "serialno": "000000002a1f1e34", "altitude": "38.393", "cputemp": 58.0, "eps": "37.16", "longitude": "-74.52923472", "ts": "2017-08-21 20:00:03", "public_ip": "71.168.184.247", "track": "236.6413", "host": "vid5", "mode": "3", "time": "2017-08-21T20:00:06.000Z", "latitude": "40.268194845", "climb": "-0.054", "speed": "0.513", "ept": "0.005"} 2017-08-21 16:20:33,199 INFO [Timer-Driven Process Thread-6] o.apache.nifi.remote.client.PeerSelector New Weighted Distribution of Nodes: PeerStatus[hostname=HW13125.local,port=8080,secure=false,flowFileCount=0] will receive 100.0% of data 2017-08-21 16:20:34,261 INFO [Timer-Driven Process Thread-6] o.a.nifi.remote.StandardRemoteGroupPort RemoteGroupPort[name=MiniFi TensorFlowImage,targets=http://hw13125.local:8080/nifi] Successfully sent [StandardFlowFileRecord[uuid=f84767ec-c627-4b63-9e88-bba1dfb4eb9b,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1503346615133-2, container=default, section=2], offset=2198, length=441],offset=0,name=3460526041973,size=441]] (441 bytes) to http://HW13125.local:8080/nifi-api in 117 milliseconds at a rate of 3.65 KB/sec {"ipaddress": "192.168.1.167", "utc": "2017-08-21T20:17:21.010Z", "epx": "10.301", "epv": "50.6", "serialno": "000000002a1f1e34", "altitude": "43.009", "cputemp": 52.0, "eps": "1.33", "longitude": "-74.529242206", "ts": "2017-08-21 20:16:55", "public_ip": "71.168.184.247", "track": "190.894", "host": "vid5", "mode": "3", "time": "2017-08-21T20:17:21.010Z", "latitude": "40.268159632", "climb": "0.022", "speed": "0.353", "ept": "0.005"} To collect our GPS information, below is my script called by MiniFi. Source: #! /usr/bin/python import os from gps import * from time import * import time import threading import json 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 time import urllib2 # Need sudo apt-get install gpsd gpsd-clients python-gps ntp # 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 # Based on # Written by Dan Mandle http://dan.mandle.me September 2012 # License: GPL 2.0 # Based on: https://hortonworks.com/tutorial/analyze-iot-weather-station-data-via-connected-data-architecture/section/3/ #### 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 # Get Raspberry Pi Serial Number def get_serial(): # Extract serial from cpuinfo file cpuserial = "0000000000000000" try: f = open('/proc/cpuinfo','r') for line in f: if line[0:6]=='Serial': cpuserial = line[10:26] f.close() except: cpuserial = "ERROR000000000" return cpuserial # Get Raspberry Pi Public IP via IPIFY Rest Call def get_public_ip(): ip = json.load(urllib2.urlopen('https://api.ipify.org/?format=json'))['ip'] return ip cpuTemp=int(float(getCPUtemperature())) ipaddress = IP_address() # Attempt to get Public IP public_ip = get_public_ip() # Attempt to get Raspberry Pi Serial Number serial = get_serial() gpsd = None class GpsPoller(threading.Thread): def __init__(self): threading.Thread.__init__(self) global gpsd #bring it in scope gpsd = gps(mode=WATCH_ENABLE) #starting the stream of info self.current_value = None self.running = True #setting the thread running to true def run(self): global gpsd while gpsp.running: gpsd.next() #this will continue to loop and grab EACH set of gpsd info to clear the buffer if __name__ == '__main__': gpsp = GpsPoller() # create the thread stopthis = False try: gpsp.start() # start it up while not stopthis: if gpsd.fix.latitude > 0: row = { 'latitude': str(gpsd.fix.latitude), 'longitude': str(gpsd.fix.longitude), 'utc': str(gpsd.utc), 'time': str(gpsd.fix.time), 'altitude': str(gpsd.fix.altitude), 'eps': str(gpsd.fix.eps), 'epx': str(gpsd.fix.epx), 'epv': str(gpsd.fix.epv), 'ept': str(gpsd.fix.ept), 'speed': str(gpsd.fix.speed), 'climb': str(gpsd.fix.climb), 'track': str(gpsd.fix.track), 'ts': currenttime, 'public_ip': public_ip, 'serialno': serial, 'host': host, 'cputemp': round(cpuTemp,2), 'ipaddress': ipaddress, 'mode': str(gpsd.fix.mode)} json_string = json.dumps(row) print json_string gpsp.running = False stopthis = True except (KeyboardInterrupt, SystemExit): #when you press ctrl+c gpsp.running = False gpsp.join() # wait for the thread to finish what it's doing Link https://github.com/tspannhw/dws2017sydney ... View more

Technology: CSV, AVRO, Hortonworks Schema Registry, Apache NiFi, Streaming Analytics Manager, Kafka, Hadoop HDFS This is a simple example of reading CSV data, using it's schema to convert it to AVRO, sending it via Kafka to SAM. SAM then reads it and stores it to HDFS. This is a simple flow, but a start to setting up any level of complex flow. A number of questions have come up on how to setup this basic flow. Gotchas: Must send AVRO, beware of nulls, set your schema, put your schema name somewhere, create your kafka topic and make sure you have permissions. Building Sample CSV Data curl "http://api.mockaroo.com/api/ANID?count=1000&key=AKEY" > "simple.csv" ****You need to sign up for mockaroo and create your own schema. You will then need to add the key and account id to the above link. Import this schema to your account <strong>{ "id": 76671, "num_rows": 1000, "file_format": "csv", "name": "simple", "include_header": true, "columns": [ { "name": "first", "type": "First Name (Male)", "formula": "" }, { "name": "second", "type": "Row Number", "formula": "" } ] }</strong> Create a Kafka Topic (simple) ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partition 1 --topic simple You can Test Consuming the Messages ./kafka-console-consumer.sh --zookeeper princeton10.field.hortonworks.com:2181 --topic simple Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].{metadata.broker.list=princeton10.field.hortonworks.com:6667, request.timeout.ms=30000, client.id=console-consumer-7447, security.protocol=PLAINTEXT}aabaabb Apache NiFi 1.x Flow For CSV Ingest Set the Schema Name via Update Properties in NiFi Publish Kafka Record Settings (need CSV Reader and AVRO Record Set Writer) CSV Reader (Make sure your use the same Schema Registry, Access Strategy and Schema Name) Avro Writer (Make sure you use the same Schema Registry, Write Strategy and Schema Name) Add Your Simple Schema to the HSR Simple Streaming Analytics Manager Store to HDFS SAM Kafka Source: SAM HDFS Sink: Set a directory and what fields you want to output Result File in HDFS hdfs dfs -cat /simple/38-HDFS-2-0-1502904631885.txt a,1 a,2 b,1 a,5 a,-1 b,-5 b,9 a,1 a,2 b,1 a,5 a,-1 b,-5 b,9 Confirmations via Data Provenance Source File wc -l simple.csv 1000 simple.csv PublishKafka msg.count 1000 hdfs dfs -cat /simple/38-HDFS-2-3-1502906678558.txt | wc -l 472 hdfs dfs -cat /simple/38-HDFS-2-2-1502906677976.txt | wc -l 296 hdfs dfs -cat /simple/38-HDFS-2-1-1502906676961.txt | wc -l 232 Reference: https://community.hortonworks.com/questions/122811/schema-registry-error-from-nifi-unrecognized-field.html?childToView=122075#answer-122075 https://hortonworks.com/open-source/streaming-analytics-manager/ https://hortonworks.com/tutorial/real-time-event-processing-in-nifi-sam-schema-registry-and-superset/ Templates (Load greg.xml into NiFi via templates. Change simplejson.txt to simple.json and load into SAM) greg.xml simplejson.txt Sample Data simplecsv.txt Create a New Schema in the Registry from this Schema { "name": "simple", "type": "record", "fields": [ { "name": "first", "type": "string" }, { "name": "second", "type": "int" } ] } ... View more

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