import logging import json import inspect from lib.config import Config from lib.videomix import CompositeModes from lib.response import NotifyResponse, OkResponse def decodeName(items, name_or_id): try: name_or_id = int(name_or_id) if name_or_id < 0 or name_or_id >= len(items): raise IndexError("unknown index %d" % name_or_id) return name_or_id except ValueError as e: try: return items.index(name_or_id) except ValueError as e: raise IndexError("unknown name %s" % name_or_id) def decodeEnumName(enum, name_or_id): try: name_or_id = int(name_or_id) if name_or_id < 0 or name_or_id >= len(enum): raise IndexError("unknown index %d" % name_or_id) return name_or_id except ValueError as e: try: return enum[name_or_id] except KeyError as e: raise IndexError("unknown name %s" % name_or_id) def encodeName(items, id): try: return items[id] except IndexError as e: raise IndexError("unknown index %d" % id) def encodeEnumName(enum, id): try: return enum(id).name except ValueError as e: raise IndexError("unknown index %d" % id) class ControlServerCommands(object): def __init__(self, pipeline): self.log = logging.getLogger('ControlServerCommands') self.pipeline = pipeline self.sources = Config.getlist('mix', 'sources') self.blankerSources = Config.getlist('stream-blanker', 'sources') # Commands are defined below. Errors are sent to the clients by throwing # exceptions, they will be turned into messages outside. def message(self, *args): """sends a message through the control-server, which can be received by user-defined scripts. does not change the state of the voctocore.""" return NotifyResponse('message', *args) def help(self): helplines = [] helplines.append("Commands:") for name, func in ControlServerCommands.__dict__.items(): if name[0] == '_': continue if not func.__code__: continue params = inspect.signature(func).parameters params = [str(info) for name, info in params.items()] params = ', '.join(params[1:]) command_sig = '\t' + name if params: command_sig += ': '+params if func.__doc__: command_sig += '\n'+'\n'.join( ['\t\t'+line.strip() for line in func.__doc__.splitlines()])+'\n' helplines.append(command_sig) helplines.append('\t'+'quit / exit') helplines.append("\n") helplines.append("Source-Names:") for source in self.sources: helplines.append("\t"+source) helplines.append("\n") helplines.append("Stream-Blanker Sources-Names:") for source in self.blankerSources: helplines.append("\t"+source) helplines.append("\n") helplines.append("Composition-Modes:") for mode in CompositeModes: helplines.append("\t"+mode.name) return OkResponse("\n".join(helplines)) def _get_video_status(self): a = encodeName( self.sources, self.pipeline.vmix.getVideoSourceA() ) b = encodeName( self.sources, self.pipeline.vmix.getVideoSourceB() ) return [a, b] def get_video(self): """gets the current video-status, consisting of the name of video-source A and video-source B""" status = self._get_video_status() return OkResponse('video_status', *status) def set_video_a(self, src_name_or_id): """sets the video-source A to the supplied source-name or source-id, swapping A and B if the supplied source is currently used as video-source B""" src_id = decodeName(self.sources, src_name_or_id) self.pipeline.vmix.setVideoSourceA(src_id) status = self._get_video_status() return NotifyResponse('video_status', *status) def set_video_b(self, src_name_or_id): """sets the video-source B to the supplied source-name or source-id, swapping A and B if the supplied source is currently used as video-source A""" src_id = decodeName(self.sources, src_name_or_id) self.pipeline.vmix.setVideoSourceB(src_id) status = self._get_video_status() return NotifyResponse('video_status', *status) def _get_audio_status(self): src_id = self.pipeline.amix.getAudioSource() return encodeName(self.sources, src_id) def get_audio(self): """gets the name of the current audio-source""" status = self._get_audio_status() return OkResponse('audio_status', status) def set_audio(self, src_name_or_id): """sets the audio-source to the supplied source-name or source-id""" src_id = decodeName(self.sources, src_name_or_id) self.pipeline.amix.setAudioSource(src_id) status = self._get_audio_status() return NotifyResponse('audio_status', status) def _get_composite_status(self): mode = self.pipeline.vmix.getCompositeMode() return encodeEnumName(CompositeModes, mode) def get_composite_mode(self): """gets the name of the current composite-mode""" status = self._get_composite_status() return OkResponse('composite_mode', status) def set_composite_mode(self, mode_name_or_id): """sets the name of the id of the composite-mode""" mode = decodeEnumName(CompositeModes, mode_name_or_id) self.pipeline.vmix.setCompositeMode(mode) composite_status = self._get_composite_status() video_status = self._get_video_status() return [ NotifyResponse('composite_mode', composite_status), NotifyResponse('video_status', *video_status) ] def set_videos_and_composite(self, src_a_name_or_id, src_b_name_or_id, mode_name_or_id): """sets the A- and the B-source synchronously with the composition-mode all parametets can be set to "*" which will leave them unchanged.""" if src_a_name_or_id != '*': src_a_id = decodeName(self.sources, src_a_name_or_id) self.pipeline.vmix.setVideoSourceA(src_a_id) if src_b_name_or_id != '*': src_b_id = decodeName(self.sources, src_b_name_or_id) self.pipeline.vmix.setVideoSourceB(src_b_id) if mode_name_or_id != '*': mode = decodeEnumName(CompositeModes, mode_name_or_id) self.pipeline.vmix.setCompositeMode(mode) composite_status = self._get_composite_status() video_status = self._get_video_status() return [ NotifyResponse('composite_mode', composite_status), NotifyResponse('video_status', *video_status) ] def _get_stream_status(self): blankSource = self.pipeline.streamblanker.blankSource if blankSource is None: return ('live',) return 'blank', encodeName(self.blankerSources, blankSource) def get_stream_status(self): """gets the current streamblanker-status""" status = self._get_stream_status() return OkResponse('stream_status', *status) def set_stream_blank(self, source_name_or_id): """sets the streamblanker-status to blank with the specified blanker-source-name or -id""" src_id = decodeName(self.blankerSources, source_name_or_id) self.pipeline.streamblanker.setBlankSource(src_id) status = self._get_stream_status() return NotifyResponse('stream_status', *status) def set_stream_live(self): """sets the streamblanker-status to live""" self.pipeline.streamblanker.setBlankSource(None) status = self._get_stream_status() return NotifyResponse('stream_status', *status) def get_config(self): """returns the parsed server-config""" confdict = {header: dict(section) for header, section in dict(Config).items()} return OkResponse('server_config', json.dumps(confdict))