now working with images(webcam streaming)

Author: nikborovets

.gitignore

@@ -1,3 +1,5 @@
 .DS_Store
 
-myenv
+myenv
+processed_frames
+videos

README.md

@@ -6,9 +6,9 @@
 
 Проект включает три основных компонента:
 
-1. **Producer**: Генерирует данные заказов и отправляет их в Kafka.
-2. **Consumer**: Читает данные заказов из Kafka и сохраняет их в PostgreSQL.
-3. **Analyzer**: Анализирует данные заказов, хранящиеся в PostgreSQL.
+1. **Producer**: Захватывает кадры с веб-камеры и отправляет их в Kafka.
+2. **Consumer**: Читает кадры из Kafka, обрабатывает их (преобразует в оттенки серого) и сохраняет в PostgreSQL.
+3. **Analyzer**: Анализирует кадры, хранящиеся в PostgreSQL, и может создавать видео из этих кадров.
 
 ## Установка
 
@@ -36,4 +36,5 @@ source myenv/bin/activate && python consumer.py
 
 ```sh
 source myenv/bin/activate && python analyzer.py
-```
+```
+

analyzer.py

@@ -1,4 +1,11 @@
 import psycopg2
+import base64
+from PIL import Image
+from io import BytesIO
+import cv2
+import numpy as np
+import os
+from datetime import datetime
 
 conn = psycopg2.connect(
     dbname='orders_db',
@@ -9,6 +16,96 @@
 )
 cursor = conn.cursor()
 
-cursor.execute("SELECT COUNT(*), SUM(amount) FROM orders")
-count, total_amount = cursor.fetchone()
-print(f"Total Orders: {count}, Total Amount: {total_amount}")
+def list_frames():
+    cursor.execute("SELECT frame_id, frame_data, timestamp FROM frames")
+    rows = cursor.fetchall()
+    
+    for row in rows:
+        new_row = []
+        for value in row:
+            if isinstance(value, str) and len(value) > 10:
+                new_row.append(value[-10:] + "...")
+            else:
+                new_row.append(value)
+        print(new_row)
+    
+    print("The number of frames: ", cursor.rowcount)
+    
+def display_frame(frame_id):
+    cursor.execute("SELECT frame_data FROM frames WHERE frame_id = %s", (frame_id,))
+    row = cursor.fetchone()
+    
+    if row is not None:
+        frame_data = row[0]
+        frame_bytes = base64.b64decode(frame_data)
+
+        nparr = np.frombuffer(frame_bytes, np.uint8)
+        img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+        
+        img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
+        img_pil = Image.fromarray(img_rgb)
+        
+        img_pil.show()
+    else:
+        print("No frame found with the given frame_id")
+
+def create_video_from_frames(output_path, fps=1):
+    cursor.execute("SELECT frame_data FROM frames ORDER BY timestamp ASC")
+    rows = cursor.fetchall()
+
+    if not rows:
+        print("No frames found in the database.")
+        return
+
+    frame_list = []
+    for row in rows:
+        frame_data = row[0]
+        frame_bytes = base64.b64decode(frame_data)
+        
+        nparr = np.frombuffer(frame_bytes, np.uint8)
+        img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+        
+        frame_list.append(img)
+
+    height, width = frame_list[0].shape[:2]
+    # добавить когда-то проверку на размеры фреймов. если вдруг разные разрешения кадров будут
+    # print(frame_list[1].shape)
+
+    out = cv2.VideoWriter(output_path, cv2.VideoWriter_fourcc(*'mp4v'), fps, (width, height))
+
+    for frame in frame_list:
+        out.write(frame)
+
+    out.release()
+    print(f"Video created at {output_path}")
+
+def main():
+    while True:
+        print("Options: ")
+        print("1. List frames")
+        print("2. Display a frame")
+        print("3. Create video from frames")
+        print("4. Exit")
+        
+        choice = input("Enter your choice: ")
+        
+        if choice == '1':
+            list_frames()
+        elif choice == '2':
+            frame_id = input("Enter frame_id: ")
+            display_frame(frame_id)
+        elif choice == '3':
+            if not os.path.isdir('videos'):
+                os.makedirs('videos')
+            
+            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+            video_path = os.path.join(os.getcwd(), 'videos', f'output_video_{timestamp}.mp4')
+            
+            create_video_from_frames(video_path)
+        elif choice == '4':
+            break
+        else:
+            print("Invalid choice. Try again.")
+
+if __name__ == '__main__':
+    main()

check.txt

@@ -7,6 +7,12 @@ docker-compose up -d
 
 docker exec -it kafka_postgres_learning-postgres-1 psql -U user -d orders_db
 
+docker exec -it kafka_postgres_learning-kafka-1 /bin/bash
+
+kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic webcam-frames --from-beginning
+
+
+
 CREATE TABLE orders (
     order_id SERIAL PRIMARY KEY,
     user_id INT NOT NULL,
@@ -26,6 +32,9 @@ docker-compose down
 docker-compose ps
 
 DROP TABLE orders;
+DROP TABLE frames;
+TRUNCATE frames;
+
 
 docker images -a -q
 docker images

consumer.py

@@ -1,13 +1,17 @@
 from kafka import KafkaConsumer
+import cv2
+import base64
 import json
+import numpy as np
 import psycopg2
+from io import BytesIO
 
 consumer = KafkaConsumer(
-    'orders',
+    'frames',
     bootstrap_servers='localhost:9092',
     auto_offset_reset='earliest',
     enable_auto_commit=True,
-    group_id='order-group',
+    group_id='frame-group',
     value_deserializer=lambda x: json.loads(x.decode('utf-8'))
 )
 
@@ -20,14 +24,25 @@
 )
 cursor = conn.cursor()
 
-def save_order(order):
+def save_frame(frame_data, timestamp):
     cursor.execute(
-        "INSERT INTO orders (user_id, amount, timestamp) VALUES (%s, %s, %s)",
-        (order['user_id'], order['amount'], order['timestamp'])
+        "INSERT INTO frames (frame_data, timestamp) VALUES (%s, %s)",
+        (frame_data, timestamp)
     )
     conn.commit()
 
 for message in consumer:
-    order = message.value
-    save_order(order)
-    print(f"Consumed: {order}")
+    frame_str = message.value['frame']
+    frame_bytes = base64.b64decode(frame_str)
+    
+    nparr = np.frombuffer(frame_bytes, np.uint8)
+    img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+    
+    gray_img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
+    
+    _, buffer = cv2.imencode('.jpg', gray_img)
+    frame_data = base64.b64encode(buffer).decode('utf-8')
+    timestamp = message.value['timestamp']
+    
+    save_frame(frame_data, timestamp)
+    print(f"Consumed and saved frame with timestamp: {timestamp}")

producer.py

@@ -1,23 +1,32 @@
 from kafka import KafkaProducer
-import json
+import cv2
+import base64
 import time
-import random
+import json
 
 producer = KafkaProducer(
     bootstrap_servers='localhost:9092',
     value_serializer=lambda v: json.dumps(v).encode('utf-8')
 )
 
-def generate_order():
-    order = {
-        'user_id': random.randint(1, 100),
-        'amount': round(random.uniform(10.0, 1000.0), 2),
-        'timestamp': int(time.time())
-    }
-    return order
+cap = cv2.VideoCapture(0)
 
-while True:
-    order = generate_order()
-    producer.send('orders', order)
-    print(f"Produced: {order}")
+if not cap.isOpened():
+    print("Error: Could not open video stream.")
+    exit()
+
+while cap.isOpened():
+    ret, frame = cap.read()
+    if not ret:
+        print("Error: Could not read frame.")
+        break
+    
+    _, buffer = cv2.imencode('.jpg', frame)
+    frame_bytes = base64.b64encode(buffer).decode('utf-8')
+    message = {'frame': frame_bytes, 'timestamp': int(time.time())}
+    
+    producer.send('frames', message)
+    print(f"Produced: {message}")
     time.sleep(1)
+
+cap.release()

requirements.txt

@@ -1,4 +1,7 @@
 kafka-python==2.0.2
+numpy==1.26.4
+opencv-python==4.10.0.82
+pillow==10.3.0
 psycopg2-binary==2.9.9
 six==1.16.0
 SQLAlchemy==2.0.30

setup.sh

@@ -22,14 +22,13 @@ done
 sleep 3
 
 docker exec -i kafka_postgres_learning-postgres-1 psql -U user -d orders_db << EOF
-CREATE TABLE IF NOT EXISTS orders (
-    order_id SERIAL PRIMARY KEY,
-    user_id INT NOT NULL,
-    amount DECIMAL NOT NULL,
+CREATE TABLE IF NOT EXISTS frames (
+    frame_id SERIAL PRIMARY KEY,
+    frame_data TEXT NOT NULL,
     timestamp BIGINT NOT NULL
 );
 EOF
 
 # python producer.py &
 # python consumer.py &
-# python analyzer.py &
+# python analyzer.py &