add async stoppers to gst pipelines

stateMachine.py

@@ -11,7 +11,7 @@ from listeners import KeypadListener, PipeListener
 from blinkenLights import Blinkenlights
 from soundLib import SoundLib
 from webInterface import initWebInterface
-from stream import initCamera, FileDump
+from stream import Camera, FileDump
 
 logger = logging.getLogger(__name__)
 
@@ -135,22 +135,22 @@ class StateMachine:
 		
 		self.LED = Blinkenlights(17)
 		
-		initCamera()
+		self.camera = Camera()
 		
 		def action():
 			if self.currentState == self.states.armed:
 				self.selectState(SIGNALS.TRIGGER)
 
-		fileDump = FileDump()
+		self.fileDump = FileDump()
 		sensitiveStates = (self.states.armed, self.states.armedCountdown, self.states.triggered)
 
 		def actionVideo(pin):
 			if self.currentState in sensitiveStates:
 				self.selectState(SIGNALS.TRIGGER)
-				fileDump.addInitiator(pin)
+				self.fileDump.addInitiator(pin)
 				while GPIO.input(pin) and self.currentState in sensitiveStates:
 					time.sleep(0.1)
-				fileDump.removeInitiator(pin)
+				self.fileDump.removeInitiator(pin)
 
 		setupMotionSensor(5, 'Nate\'s room', action)
 		setupMotionSensor(19, 'front door', action)
@@ -210,6 +210,12 @@ class StateMachine:
 		if hasattr(self, 'LED'):
 			self.LED.__del__()
 			
+		if hasattr(self, 'camera'):
+			self.camera.__del__()
+			
+		if hasattr(self, 'fileDump'):
+			self.fileDump.__del__()
+			
 		if hasattr(self, 'soundLib'):
 			self.soundLib.__del__()
 			

stream.py

@@ -14,7 +14,7 @@ From a logging an error handling standpoint, all 'errors' here are logged as
 
 import gi, time, os, logging
 from datetime import datetime
-from threading import Thread, Lock
+from threading import Lock, Event
 
 from auxilary import waitForPath, mkdirSafe
 from exceptionThreading import async
@@ -31,7 +31,7 @@ from gi.repository import Gst, GObject
 class GstException(Exception):
 	pass
 
-def gstPrintMsg(pName, frmt, *args, level=logging.DEBUG, sName=None):
+def _gstPrintMsg(pName, frmt, *args, level=logging.DEBUG, sName=None):
 	if sName:
 		logger.log(level, '[{}] [{}] '.format(pName, sName) + frmt.format(*args))
 	else:
@@ -39,15 +39,15 @@ def gstPrintMsg(pName, frmt, *args, level=logging.DEBUG, sName=None):
 	if level == logging.ERROR:
 		raise GstException
 	
-def processErrorMessage(pName, sName, msg):
+def _processErrorMessage(pName, sName, msg):
 	error, debug = msg.parse_error()
 	if debug:
-		gstPrintMsg(pName, '{} - Additional debug info: {}', error.message, debug,
+		_gstPrintMsg(pName, '{} - Additional debug info: {}', error.message, debug,
 			level=logging.ERROR, sName=sName)
 	else:
-		gstPrintMsg(pName, error.message, level=logging.ERROR, sName=sName)
+		_gstPrintMsg(pName, error.message, level=logging.ERROR, sName=sName)
 
-def linkElements(e1, e2, caps=None):
+def _linkElements(e1, e2, caps=None):
 	if caps:
 		if not e1.link_filtered(e2, caps):
 			logger.error('cannot link \"%s\" to \"%s\" with caps %s',
@@ -58,25 +58,46 @@ def linkElements(e1, e2, caps=None):
 			logger.error('cannot link \%s\" to \"%s\"', e1.get_name(), e2.get_name())
 			raise SystemExit
 
-def linkTee(tee, *args):
+class ThreadedPipeline:
-	i = 0
+	'''
-	for e in args:
+	Launches a Gst Pipeline in a separate thread. Note that the 'threaded'
-		teePad = tee.get_request_pad('src_{}'.format(i))
+	aspect is impimented via and async decorator around the mainLoop below
-		ePad  = e.get_static_pad('sink')
+	'''
-		teePad.link(ePad)
+	def __init__(self, pName):
-		i += 1
+		self._pipeline = Gst.Pipeline.new(pName)
+		self._stopper = Event()
 		
-# TODO: make it clearer which pipeline everything comes from
+	def start(self, play=True):
-def eventLoop(pipeline, block=True, doProgress=False, targetState=Gst.State.PLAYING):
+		self._startPipeline(play)
+	
+	# TODO: this might not all be necessary
+	def stop(self):
+		self._stopper.set()
+		self._pipeline.set_state(Gst.State.NULL)
+		logger.debug('Shut down gstreamer pipeline: %s', self._pipeline.get_name())
+		
+	def __del__(self):
+		self.stop()
+		
+	# TODO: make it clearer which pipeline everything comes from
+	def _eventLoop(self, block=True, doProgress=False, targetState=Gst.State.PLAYING):
+		'''
+		This is the main loop that processes information on the bus and decides
+		how the pipeline should react. Sometimes this entails spitting out
+		messages, others it involves changing state or some other manipulation.
+		'''
 		buffering = False
 		inProgress = False
 		prerolled = targetState != Gst.State.PAUSED
 		
-	pName = pipeline.get_name()
+		pName = self._pipeline.get_name()
-	bus = pipeline.get_bus()
+		bus = self._pipeline.get_bus()
 		
-	# TODO: add killswitch to exit this thread?
+		while self._stopper:
-	while 1:
+			# TODO: if we actually want to stop the pipeline asyncronously we
+			# need to post a message on the bus that tells it to stop. Otherwise
+			# it will wait here forever (or will be terminated as a daemon
+			# thread by its parent)
 			msg = bus.timed_pop(1e18 if block else 0)
 
 			if not msg:
@@ -92,49 +113,50 @@ def eventLoop(pipeline, block=True, doProgress=False, targetState=Gst.State.PLAY
 			if msgType == Gst.MessageType.REQUEST_STATE:
 				state = msg.parse_request_state()
 
-			logger.info('Setting state to %s as requested by %s', state.value_name, msgSrcName)
+				logger.info('Setting state to %s as requested by %s',
+					state.value_name, msgSrcName)
 				
-			pipeline.set_state(state)
+				self._pipeline.set_state(state)
 				
 			elif msgType == Gst.MessageType.CLOCK_LOST:
 				logger.debug('Clock lost. Getting new one.')
-			pipeline.set_state(Gst.State.PAUSED)
+				self._pipeline.set_state(Gst.State.PAUSED)
-			pipeline.set_state(Gst.State.PLAYING)
+				self._pipeline.set_state(Gst.State.PLAYING)
 				
 			elif msgType == Gst.MessageType.LATENCY:
-			gstPrintMsg(pName, 'Redistributing latency', sName=msgSrcName)
+				_gstPrintMsg(pName, 'Redistributing latency', sName=msgSrcName)
-			pipeline.recalculate_latency()
+				self._pipeline.recalculate_latency()
 			
 			# messages that do not require pipeline manipulation	
 			elif msgType == Gst.MessageType.BUFFERING:
-			gstPrintMsg(pName, 'Buffering: {}', msg.parse_buffering(), sName=msgSrcName)
+				_gstPrintMsg(pName, 'Buffering: {}', msg.parse_buffering(), sName=msgSrcName)
 				
 			elif msgType == Gst.MessageType.NEW_CLOCK:
 				clock = msg.parse_new_clock()
 				clock = clock.get_name() if clock else 'NULL'
-			gstPrintMsg(pName, 'New clock: {}', clock)
+				_gstPrintMsg(pName, 'New clock: {}', clock)
 				
 			elif msgType == Gst.MessageType.INFO:
 				error, debug = msg.parse_info()
 				
 				if debug:
-				gstPrintMsg(pName, debug, level=logging.INFO, sName=msgSrcName)
+					_gstPrintMsg(pName, debug, level=logging.INFO, sName=msgSrcName)
 					
 			elif msgType == Gst.MessageType.WARNING:
 				error, debug = msg.parse_warning()
 				
 				if debug:
-				gstPrintMsg(pName, '{} - Additional debug info: {}', error.message,
+					_gstPrintMsg(pName, '{} - Additional debug info: {}', error.message,
 						debug, level=logging.WARNING, sName=msgSrcName)
 				else:
-				gstPrintMsg(pName, error.message, level=logging.WARNING, sName=msgSrcName)
+					_gstPrintMsg(pName, error.message, level=logging.WARNING, sName=msgSrcName)
 				
 			elif msgType == Gst.MessageType.ERROR:
-			processErrorMessage(pName, msgSrcName, msg)
+				_processErrorMessage(pName, msgSrcName, msg)
 				
 			elif msgType == Gst.MessageType.STATE_CHANGED:
 				# we only care about pipeline level state changes
-			if msgSrc == pipeline:
+				if msgSrc == self._pipeline:
 					old, new, pending = msg.parse_state_changed()
 					
 					# we only care if we reach the final target state
@@ -142,12 +164,12 @@ def eventLoop(pipeline, block=True, doProgress=False, targetState=Gst.State.PLAY
 						prerolled = True
 						
 						if buffering:
-						gstPrintMsg(pName, 'Prerolled, waiting for buffering to finish',
+							_gstPrintMsg(pName, 'Prerolled, waiting for buffering to finish',
 								level=logging.INFO)
 							continue
 							
 						if inProgress:
-						gstPrintMsg(pName, 'Prerolled, waiting for progress to finish',
+							_gstPrintMsg(pName, 'Prerolled, waiting for progress to finish',
 								level=logging.INFO)
 							continue
 							
@@ -166,14 +188,14 @@ def eventLoop(pipeline, block=True, doProgress=False, targetState=Gst.State.PLAY
 				  progressType == Gst.ProgressType.ERROR):
 					inProgress = False
 				
-			gstPrintMsg(pName, 'Progress: ({}) {}', code, text, sName=msgSrcName)
+				_gstPrintMsg(pName, 'Progress: ({}) {}', code, text, sName=msgSrcName)
 				
 				if doProgress and not inProgress and not buffering and prerolled:
 					return					
 
 			elif msgType == Gst.MessageType.HAVE_CONTEXT:
 				context = msg.parse_have_context()
-			gstPrintMsg(
+				_gstPrintMsg(
 					pName,
 					'Got context: {}={}',
 					context.get_context_type(),
@@ -196,18 +218,19 @@ def eventLoop(pipeline, block=True, doProgress=False, targetState=Gst.State.PLAY
 					else: 
 						valStr = Gst.value_serialize(val)
 					
-			gstPrintMsg(pName, '{}: {} = {}', obj.get_name(), propName, valStr, sName=msgSrcName)
+				_gstPrintMsg(pName, '{}: {} = {}', obj.get_name(), propName,
+					valStr, sName=msgSrcName)
 				
 			# these are things I might not need...
 			elif msgType == Gst.MessageType.STREAM_START:
-			if msgSrc == pipeline:
+				if msgSrc == self._pipeline:
-				gstPrintMsg(pName, 'Started stream', level=logging.INFO)
+					_gstPrintMsg(pName, 'Started stream', level=logging.INFO)
 
 			elif msgType == Gst.MessageType.QOS:
 				frmt, processed, dropped = msg.parse_qos_stats()
 				jitter, proportion, quality = msg.parse_qos_values()
 
-			gstPrintMsg(
+				_gstPrintMsg(
 					pName,
 					'QOS stats: jitter={} dropped={}',
 					jitter,
@@ -216,61 +239,73 @@ def eventLoop(pipeline, block=True, doProgress=False, targetState=Gst.State.PLAY
 				)
 					
 			elif msgType == Gst.MessageType.ELEMENT:
-			gstPrintMsg(pName, 'Unknown message ELEMENT', sName=msgSrcName)
+				_gstPrintMsg(pName, 'Unknown message ELEMENT', sName=msgSrcName)
 
 			elif msgType == Gst.MessageType.UNKNOWN:
-			gstPrintMsg(pName, 'Unknown message', sName=msgSrcname)
+				_gstPrintMsg(pName, 'Unknown message', sName=msgSrcname)
 		
-@async(daemon=True)
+	@async(daemon=True)
-def mainLoop(pipeline):
+	def _mainLoop(self):
-	eventLoop(pipeline, block=True, doProgress=False, targetState=Gst.State.PLAYING)
+		self._eventLoop(block=True, doProgress=False, targetState=Gst.State.PLAYING)
 		
-def startPipeline(pipeline, play=True):
+	def _startPipeline(self, play):
-	pName = pipeline.get_name()
+		pName = self._pipeline.get_name()
-	stateChange = pipeline.set_state(Gst.State.PAUSED)
+		stateChange = self._pipeline.set_state(Gst.State.PAUSED)
-	gstPrintMsg(pName, 'Setting to PAUSED', level=logging.INFO)
+		_gstPrintMsg(pName, 'Setting to PAUSED', level=logging.INFO)
 		
 		if stateChange == Gst.StateChangeReturn.FAILURE:
-		gstPrintMsg(pName, 'Cannot set to PAUSE', level=logging.INFO)
+			_gstPrintMsg(pName, 'Cannot set to PAUSE', level=logging.INFO)
-		eventLoop(pipeline, block=False, doProgress=False, targetState=Gst.State.VOID_PENDING)
+			self._eventLoop(block=False, doProgress=False, targetState=Gst.State.VOID_PENDING)
 		# we should always end up here because live
 		elif stateChange == Gst.StateChangeReturn.NO_PREROLL:
-		gstPrintMsg(pName, 'Live and does not need preroll')
+			_gstPrintMsg(pName, 'Live and does not need preroll')
 		elif stateChange == Gst.StateChangeReturn.ASYNC:
-		gstPrintMsg(pName, 'Prerolling')
+			_gstPrintMsg(pName, 'Prerolling')
 			try:
-			eventLoop(pipeline, block=True, doProgress=True, targetState=Gst.State.PAUSED)
+				_eventLoop(block=True, doProgress=True, targetState=Gst.State.PAUSED)
 			except GstException:
-			gstPrintMsg(pName, 'Does not want to preroll', level=logging.ERROR)
+				_gstPrintMsg(pName, 'Does not want to preroll', level=logging.ERROR)
 				raise SystemExit
 		elif stateChange == Gst.StateChangeReturn.SUCCESS:
-		gstPrintMsg(pName, 'Is prerolled')
+			_gstPrintMsg(pName, 'Is prerolled')
 		
 		# this should always succeed...
 		try:
-		eventLoop(pipeline, block=False, doProgress=True, targetState=Gst.State.PLAYING)
+			self._eventLoop(block=False, doProgress=True, targetState=Gst.State.PLAYING)
 		except GstException:
-		gstPrintMsg(pName, 'Does not want to preroll', level=logging.ERROR)
+			_gstPrintMsg(pName, 'Does not want to preroll', level=logging.ERROR)
 			raise SystemExit
 		# ...and end up here
 		else:
 			if play:
-			gstPrintMsg(pName, 'Setting to PLAYING', level=logging.INFO)
+				_gstPrintMsg(pName, 'Setting to PLAYING', level=logging.INFO)
 			
 				# ...and since this will ALWAYS be successful...
-			if pipeline.set_state(Gst.State.PLAYING) == Gst.StateChangeReturn.FAILURE:
+				if self._pipeline.set_state(Gst.State.PLAYING) == Gst.StateChangeReturn.FAILURE:
-				gstPrintMsg(pName, 'Cannot set to PLAYING', level=logging.ERROR)
+					_gstPrintMsg(pName, 'Cannot set to PLAYING', level=logging.ERROR)
-				err = pipeline.get_bus().pop_filtered(Gst.MessageType.Error)
+					err = self._pipeline.get_bus().pop_filtered(Gst.MessageType.Error)
-				processErrorMessage(pName, msgSrcName, err)
+					_processErrorMessage(pName, msgSrcName, err)
 			
 			# ...we end up here and loop until Tool releases their next album
 			try:
-			mainLoop(pipeline)
+				self._mainLoop()
 			except:
 				raise GstException
 
-def initCamera(video=True, audio=True):
+class Camera(ThreadedPipeline):
-	pipeline = Gst.Pipeline.new("camera")
+	'''
+	Class for usb camera. The 'video' and 'audio' flags are meant for testing.
+	
+	Makes two independent stream that share the same pipeline, and thus share a
+	clock for syncronization. Video uses the hardware-accelarated OMX extensions
+	for H264 encoding. Audio has no hardware accelaration (and thus is likely
+	the most resource hungry thing in this program) and encodes using Opus. Both
+	send their stream to two UDP ports (900X for video, 800X for audio, where 
+	X = 1 is used by the Janus WebRTC interface and X = 2 is used by the
+	FileDump class below.
+	'''
+	def __init__(self, video=True, audio=True):
+		super().__init__('camera')
 		
 		vPath = '/dev/video0'
 		
@@ -290,14 +325,14 @@ def initCamera(video=True, audio=True):
 		
 			vCaps = Gst.Caps.from_string('video/x-raw,width=640,height=480,framerate=30/1')
 			
-		pipeline.add(vSource, vConvert, vScale, vClock, vEncode, vRTPPay, vRTPSink)
+			self._pipeline.add(vSource, vConvert, vScale, vClock, vEncode, vRTPPay, vRTPSink)
 			
-		linkElements(vSource, vConvert)
+			_linkElements(vSource, vConvert)
-		linkElements(vConvert, vScale)
+			_linkElements(vConvert, vScale)
-		linkElements(vScale, vClock, vCaps)
+			_linkElements(vScale, vClock, vCaps)
-		linkElements(vClock, vEncode)
+			_linkElements(vClock, vEncode)
-		linkElements(vEncode, vRTPPay)
+			_linkElements(vEncode, vRTPPay)
-		linkElements(vRTPPay, vRTPSink)
+			_linkElements(vRTPPay, vRTPSink)
 		
 		if audio:
 			aSource = Gst.ElementFactory.make("alsasrc", "audioSource")
@@ -312,19 +347,19 @@ def initCamera(video=True, audio=True):
 
 			aCaps = Gst.Caps.from_string('audio/x-raw,rate=48000,channels=1')
 
-		pipeline.add(aSource, aConvert, aScale, aEncode, aRTPPay, aRTPSink)
+			self._pipeline.add(aSource, aConvert, aScale, aEncode, aRTPPay, aRTPSink)
 
-		linkElements(aSource, aConvert)
+			_linkElements(aSource, aConvert)
-		linkElements(aConvert, aScale)
+			_linkElements(aConvert, aScale)
-		linkElements(aScale, aEncode, aCaps)
+			_linkElements(aScale, aEncode, aCaps)
-		linkElements(aEncode, aRTPPay)
+			_linkElements(aEncode, aRTPPay)
-		linkElements(aRTPPay, aRTPSink)
+			_linkElements(aRTPPay, aRTPSink)
 			
 		waitForPath(vPath) # video is on usb, so wait until it comes back after we hard reset the bus
 		
-	startPipeline(pipeline)
+		self.start()
 
-class FileDump:
+class FileDump(ThreadedPipeline):
 	'''
 	Pipeline that takes audio and input from two udp ports, muxes them, and
 	dumps the result to a file. Intended to work with the Camera above. Will
@@ -348,7 +383,7 @@ class FileDump:
 
 		mkdirSafe(self._savePath, logger)
 
-		self.pipeline = Gst.Pipeline.new('filedump')
+		super().__init__('filedump')
 		
 		aSource = Gst.ElementFactory.make('udpsrc', 'audioSource')
 		aJitBuf = Gst.ElementFactory.make('rtpjitterbuffer', 'audioJitterBuffer')
@@ -373,28 +408,27 @@ class FileDump:
 		aSource.set_property('port', 8002)
 		vSource.set_property('port', 9002)
 		
-		self.pipeline.add(aSource, aJitBuf, aDepay, aQueue,
+		self._pipeline.add(aSource, aJitBuf, aDepay, aQueue,
 			vSource, vJitBuf, vDepay, vParse, vQueue, mux, self.sink)
 	
+		_linkElements(aSource, aJitBuf, aCaps)
+		_linkElements(aJitBuf, aDepay)
+		_linkElements(aDepay, aQueue)
+		_linkElements(aQueue, mux)
 		
-		linkElements(aSource, aJitBuf, aCaps)
+		_linkElements(vSource, vJitBuf, vCaps)
-		linkElements(aJitBuf, aDepay)
+		_linkElements(vJitBuf, vDepay)
-		linkElements(aDepay, aQueue)
+		_linkElements(vDepay, vParse)
-		linkElements(aQueue, mux)
+		_linkElements(vParse, vQueue)
+		_linkElements(vQueue, mux)
 		
-		linkElements(vSource, vJitBuf, vCaps)
+		_linkElements(mux, self.sink)
-		linkElements(vJitBuf, vDepay)
-		linkElements(vDepay, vParse)
-		linkElements(vParse, vQueue)
-		linkElements(vQueue, mux)
-		
-		linkElements(mux, self.sink)
 		
 		# TODO: there is probably a better way to init than starting up to PAUSE
 		# and then dropping back down to NULL
-		startPipeline(self.pipeline, play=False)
+		self.start(play=False)
 		
-		self.pipeline.post_message(Gst.Message.new_request_state(self.pipeline, Gst.State.NULL))
+		self._pipeline.post_message(Gst.Message.new_request_state(self._pipeline, Gst.State.NULL))
 		
 	def addInitiator(self, identifier):
 		with self._lock:
@@ -403,11 +437,11 @@ class FileDump:
 			else:
 				self._initiators.append(identifier)
 				
-			if self.pipeline.get_state(Gst.CLOCK_TIME_NONE).state == Gst.State.NULL:
+			if self._pipeline.get_state(Gst.CLOCK_TIME_NONE).state == Gst.State.NULL:
 				filePath = os.path.join(self._savePath, '{}.mkv'.format(datetime.now()))
 				self.sink.set_property('location', filePath)
 				# TODO: cannot post messages from null to change state, is this bad?
-				self.pipeline.set_state(Gst.State.PLAYING)
+				self._pipeline.set_state(Gst.State.PLAYING)
 		
 	def removeInitiator(self, identifier):
 		with self._lock:
@@ -417,12 +451,6 @@ class FileDump:
 				logger.warn('Attempted to remove nonexistant identifier \'%s\'', identifier)
 				
 			if len(self._initiators) == 0:
-				self.pipeline.set_state(Gst.State.NULL)
+				self._pipeline.set_state(Gst.State.NULL)
 		
 Gst.init(None)
-
-# this works for file dump
-# gst-launch-1.0 -v udpsrc port=8001 ! application/x-rtp,encoding-name=OPUS,payload=96 ! 
-# rtpjitterbuffer ! rtpopusdepay ! queue ! matroskamux name=mux ! filesink location=testicle.mkv \
-# udpsrc port=9001 ! application/x-rtp,encoding-name=H264,payload=96 ! rtpjitterbuffer ! \
-# rtph264depay ! h264parse ! queue ! mux.