Adds image capturer

Author: albertein

.gitignore

@@ -1,3 +1,6 @@
 node_modules/
 .DS_Store
 kafka_2.11-0.10.0.0/
+cv.py
+cv2.so
+tmp.jpg

README.md

@@ -8,4 +8,7 @@ ullala is a demo that showcase using kafka as a data pipeline and log aggregatio
 2. Run Kafka, Zookeeper, Elasticsearch and Logstash: `docker-compose up`
 3. Set kafka on your host file to point to the running kafka instance (docker-machine ip default on mac)
 3. Setup topics `scripts/setup-topics.sh`
-
+4. Install OpenCV for capturer, `brew install opencv` on mac.
+5. Symlink opencv into capturer, for mac you can run `scripts/symlink-opencv-bew.sh`
+6. Install capturer dependencies `pip install -r capturer/requirements.txt`
+6. Run Capturer: `KAFKA_HOSTS=KAFKA:9092 capturer/capturer.py

capturer/capture.py

@@ -0,0 +1,64 @@
+from __future__ import print_function
+
+import io
+import sys
+import logging
+import calendar
+from os import getenv
+from time import sleep
+from datetime import datetime
+
+import avro.io
+import avro.schema
+from SimpleCV import Camera
+from kafka import KafkaProducer
+
+def get_timestamp():
+    timestamp = datetime.now()
+    return '{0}:{1}:{2}'.format(timestamp.hour, timestamp.minute, timestamp.second)
+
+def run_capturer(kafka_hosts, fps=24):
+    producer = KafkaProducer(bootstrap_servers=kafka_hosts)
+    cam = Camera()
+    while True:
+        img = cam.getImage()
+        img.drawText(get_timestamp(), fontsize=160)
+        img.save('tmp.jpg')
+        with open('tmp.jpg', mode='rb') as file:
+            content = file.read()
+        producer.send('CAMERA_FEED', pack_image(content))
+        print('Got an image')
+        sleep(0.4)
+
+def pack_image(image):
+    schema_string = """
+    {
+        "namespace": "ullala",
+        "type": "record",
+        "name": "Image",
+        "fields": [
+            {"name": "image", "type": "bytes"},
+            {"name": "capture_timestamp",  "type": "int"}
+        ]
+    }
+    """
+    schema = avro.schema.parse(schema_string)
+    capture_timestamp = calendar.timegm(datetime.utcnow().utctimetuple())
+
+    writer = avro.io.DatumWriter(schema)
+
+    bytes_writer = io.BytesIO()
+    encoder = avro.io.BinaryEncoder(bytes_writer)
+    
+    writer.write({"image": image, "capture_timestamp": capture_timestamp}, encoder)
+
+    return bytes_writer.getvalue()
+
+
+if __name__ == '__main__':
+    logging.getLogger().setLevel(logging.DEBUG)
+    kafka_hosts = getenv('KAFKA_HOSTS')
+    if not kafka_hosts:
+        print('You need to set $KAFKA_HOSTS environment variable', file=sys.stderr)
+        exit(1)
+    run_capturer(kafka_hosts, fps=24)

capturer/requirements.txt

@@ -0,0 +1,5 @@
+scipy
+simplecv
+pygame
+kafka-python
+avro

scripts/symlink-opencv-brew.sh

@@ -0,0 +1,4 @@
+#! /bin/sh
+
+ln -s /usr/local/Cellar/opencv/2.4.9/lib/python2.7/site-packages/cv2.so `pwd`/capturer/cv2.so
+ln -s /usr/local/Cellar/opencv/2.4.9/lib/python2.7/site-packages/cv.py `pwd`/capturer/cv.py