elsampsa / valkka-multiprocess Goto Github PK

{
  "cam1": "path/to/cam1/frame1.jpg",
  "cam2": "path/to/cam1/frame1.jpg",
  ...
  ...
  "cam10": "path/to/cam10/frame1.jpg"
}

class LiveStream:
	def __init__(self, shmem_buffers, shmem_name, address, slot, width, height):
		self.shmem_buffers = shmem_buffers
		self.shmem_name = shmem_name
		self.address = f"rtsp://{USER}:{PASSWORD}@{address}:554"
		self.slot = slot

		# RBGShmem Filter
		self.shmem_filter = RGBShmemFrameFilter(self.shmem_name, shmem_buffers, width, height)
		# self.shmem_filter = BriefInfoFrameFilter(self.shmem_name)  # For Debugging

		# SWS Filter
		self.sws_filter = SwScaleFrameFilter(f"sws_{self.shmem_name}", width, height, self.shmem_filter)
		# self.interval_filter = TimeIntervalFrameFilter("interval_filter", 0, self.sws_filter)

		# decoding part
		self.avthread = AVThread("avthread", self.sws_filter)
		self.av_in_filter = self.avthread.getFrameFilter()

		# define connection to camera
		self.ctx = LiveConnectionContext(LiveConnectionType_rtsp, self.address, self.slot, self.av_in_filter)

		self.avthread.startCall()
		self.avthread.decodingOnCall()

	def close(self):
		self.avthread.decodingOffCall()
		self.avthread.stopCall()


def start_valkka(camera_indexes, shmem_buffers, width, height, shmem_names=[], prod=False):
	livestreams = {}

	# Defining livethread
	livethread = LiveThread("livethread")

	# Defining filter
	shmem_filters = {}
	sws_filters = {}
	for i, cam in enumerate(camera_indexes, start=1):
		livestreams[i] = LiveStream(
			shmem_buffers=shmem_buffers,
			shmem_name=f"{cam}",
			address=cam,
			slot=i,
			width=width,
			height=height
		)

	# Start livethread
	livethread.startCall()

	# Register context to livethread
	for livestream in livestreams.values():
		livethread.registerStreamCall(livestream.ctx)
		livethread.playStreamCall(livestream.ctx)
		shmem_names.append(livestream.shmem_name)

	if prod:
		while True:
			continue
	else:
		time.sleep(30)

	for livestream in livestreams.values():
		livestream.close()
	livethread.stopCall()
	print("Bye")
	return True

import threading
import json
import cv2
import redis
from termcolor import colored
from valkka.api2 import ShmemRGBClient

from config import get_configs

conf = get_configs()
FRAMES_SAVE_TO = conf.get("frames_path")
REDIS_HOST = conf.get("redis")["host"]
REDIS_PORT = conf.get("redis")["port"]
REDIS_DB = conf.get("redis")["db"]


def save_frame(shmem_name, shmem_buffers, width, height, mstimeout):
	client = ShmemRGBClient(
		name=shmem_name,
		n_ringbuffer=shmem_buffers,
		width=width,
		height=height,
		mstimeout=mstimeout,
		verbose=False
	)
	lock = threading.Lock()
	while True:
		index, meta = client.pullFrame()
		if index is None:
			print(colored("No frame...", "red"))
			continue
		data = client.shmem_list[index][0:meta.size]
		img = data.reshape((meta.height, meta.width, 3))
		img = cv2.cvtColor(img, cv2.COLOR_RGB2BGR)
		image = FRAMES_SAVE_TO + f"{shmem_name}/{meta.mstimestamp}.jpg"
		# lock.acquire()
		# push2redis(shmem_name, image)
		# lock.release()
		cv2.imwrite(image, img)

		# img = imutils.resize(img, 640, 360)
		# cv2.imshow(shmem_name, img)
		# cv2.waitKey(1)

import threading
import time
from config import get_configs
from client import save_frame

from server import start_valkka

# Load config
conf = get_configs()
camera_indexes = conf.get("cameras", [0])
width = conf.get("image_width", 1920)
height = conf.get("image_height", 1080)
shmem_buffers = conf.get("shared_memory_buffers", 10)
mstimeout = conf.get("mstimeout", 10)

shmem_names = []
client_threads = []

if __name__ == '__main__':
	# Start server thread
	print("Starting Camera...")
	server_thread = threading.Thread(target=start_valkka, args=(camera_indexes, shmem_buffers, width, height, shmem_names, False))
	server_thread.start()
	time.sleep(3)
	print(f"Total shmem: {len(shmem_names)}")

	# Start client thread
	for shmem_name in shmem_names:
		thread = threading.Thread(target=save_frame, args=(shmem_name, shmem_buffers, width, height, mstimeout))
		client_threads.append(thread)
		thread.start()

	time.sleep(60)

	server_thread.join()
	for thread in client_threads:
		thread.join()