#! /usr/bin/env python3 # # Interactive gstreamer shell # Copyright (c) 2007 Olivier Aubert # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston MA 02110-1301, USA. # import logging logger = logging.getLogger(__name__) __version__ = "0.5" import os import re import sys import time import gi gi.require_version('Gst', '1.0') gi.require_version('Gdk', '3.0') gi.require_version('Gtk', '3.0') from gi.repository import GLib from gi.repository import Gdk from gi.repository import Gtk from gi.repository import Gst Gst.init(sys.argv) from subprocess import check_output from evaluator import Evaluator INSPECT_PROGRAM='gst-inspect-1.0' def find_program_in_path(p): for d in os.getenv('PATH').split(':'): fp=os.path.join(d, p) if os.path.exists(fp): return fp return None # 0.10 -> 1.0 porting guide: https://wiki.ubuntu.com/Novacut/GStreamer1.0 class GstEvaluator: def __init__(self): self.pipeline=None # Full path to the gst-inspect program. Will be initialized on demand. self.inspect_path=None self.widget=self.build_widget() self.initialize_completions() # Default pipeline definition self.set_pipeline('videotestsrc ! videoconvert ! autovideosink sync=false') def test_image_overlay(self): self.set_pipeline("videotestsrc ! queue ! videomixer name=videomixer ! queue ! videoconvert ! xvimagesink sync=false") mixer=self.pipeline.get_by_name("videomixer") self.imageoverlaybin = Gst.Bin("ImageOverlay") filesrc = Gst.ElementFactory.make("filesrc") filesrc.set_property("location", '/tmp/a.png') pngdec = Gst.ElementFactory.make("pngdec") alphacolor = Gst.ElementFactory.make("alphacolor") videoscale = Gst.ElementFactory.make("videoscale") queue = Gst.ElementFactory.make("queue") self.imageoverlaybin.add(filesrc, pngdec, alphacolor, videoscale, queue) els = (filesrc, pngdec, alphacolor, videoscale) for (source, sink) in zip(els, els[1:]): source.link(sink) structure = Gst.Structure("video/x-raw-yuv") #(width, height) = [int(dim) for dim in self.resolution.split("x")] # e.g. '640x480' #structure["width"] = width #structure["height"] = height caps = Gst.Caps(structure) videoscale.link(queue, caps) ghostpad = Gst.GhostPad.new("src", queue.get_static_pad("src")) self.imageoverlaybin.add_pad(ghostpad) self.pipeline.add(self.imageoverlaybin) self.imageoverlaybin.link(mixer) self.imageoverlaybin.set_state(Gst.State.PLAYING) def on_message (self, bus, message): if hasattr(message, 'structure'): s = message.structure if s.get_name() == 'level': # Level information # Note: s['rms'] returns the level in stereo (left, # right). We use only 1 value ATM self.ev.info(self.format_time(s['timestamp'] / Gst.MSECOND), self.format_time(s['endtime'] / Gst.MSECOND), s['rms'][0]) else: self.ev.info("MSG " + bus.get_name() + ": " + s.to_string()) #self.ev.info("NAME" + message.structure.get_name()) return True def set_pipeline(self, p, redisplay=True): logger.warning("set_pipeline %s", p) # Deactivate the previous pipeline if self.pipeline: self.pipeline.set_state(Gst.State.NULL) try: pipeline = Gst.parse_launch(p) except Exception as e: self.ev.clear_output() logger.error("Cannot parse new pipe", exc_info=True) self.ev.log("Cannot parse new pipe:\n" + str(e)) else: self.ev.clear_output() self.ev.log("New pipeline successfully parsed.") self.pipeline=pipeline # Handle bus messages bus = self.pipeline.get_bus() bus.add_signal_watch() bus.connect("message", self.on_message) self.pipeline.set_state(Gst.State.PLAYING) # Provide shortcuts to access the pipeline elements. elements = dict( (e.get_name(), e) for e in ( self.pipeline.get_child_by_index(i) for i in range(self.pipeline.get_children_count()) ) ) elements['pipe']=self.pipeline elements['self']=self self.ev.locals_=dict(elements) if redisplay: self.display_pipeline(p) return True def display_pipeline(self, p): """Display the pipeline definition. """ self.pipebuffer.delete(*self.pipebuffer.get_bounds()) self.pipebuffer.insert_at_cursor(p) return True def get_word_at_cursor(self, b): """Return the word at the cursor position in the given TextBuffer """ cursor_position=b.get_iter_at_mark(b.get_insert()) word_start=cursor_position.copy() # backward_word_start considers - as a word boundary. # Cool in general, but no so for gstreamer element fragment = word_start.backward_search(" ", 0) if fragment is not None: # Found a space. word_start = fragment[0] # Strip the leading space word_start.forward_char() else: word_start.backward_word_start() word_end=cursor_position.copy() if word_end.get_char() != ' ': word_end.forward_word_end() return word_start.get_text(word_end) def initialize_completions(self): if self.inspect_path is None: self.inspect_path=find_program_in_path(INSPECT_PROGRAM) if self.inspect_path is None: self.gst_plugins=[] return True lines = check_output([ self.inspect_path ]).decode('ascii', 'ignore').splitlines() # Remove the filename prefix data=[ l[(l.find(':')+1):] for l in lines ] # Remove the last 2 lines data = [ l[:l.find(':')].strip() for l in data[:-2] ] # Do not consider elements which have a comment, like # v4l2deviceprovider (GstDeviceProviderFactory) # since they are abstract elements. self.gst_plugins=[ l for l in data if not ' ' in l ] return True def get_gst_completions(self, prefix): """Return the list of possible completions. """ res = [ w for w in self.gst_plugins if w.startswith(prefix) and w != prefix ] return res def format_time (self, val=None): """Formats a value (in milliseconds) into a time string. @param val: the value @type val: int @return: the formatted string @rtype: string """ if val is None: return '--:--:--.---' elif val < 0: val = 0 (s, ms) = divmod(int(val), 1000) # Format: HH:MM:SS.mmm return "%s.%03d" % (time.strftime("%H:%M:%S", time.gmtime(s)), ms) def update_status(self): if self.pipeline is None: self.statuslabel.set_text("No valid pipeline") return True try: pos, format = self.pipeline.query_position(Gst.Format.TIME) except: position = None else: position = pos * 1.0 / Gst.MSECOND self.statuslabel.set_text("%s - %s" % (self.pipeline.get_state(100)[1].value_nick, self.format_time(position))) return True def build_widget(self): self.ev=Evaluator(globals_=globals(), locals_={}, display_info_widget=True) window=self.ev.popup(embedded=False) window.connect ("destroy", lambda e: Gtk.main_quit()) def pipe_keypress_cb(widget=None, event=None): # Initialize inspect_path if necessary if self.inspect_path is None: self.inspect_path=find_program_in_path(INSPECT_PROGRAM) if self.inspect_path is None: m=Gtk.MessageDialog(type=Gtk.MessageType.ERROR, flags=Gtk.DialogFlags.MODAL | Gtk.DialogFlags.DESTROY_WITH_PARENT, buttons=Gtk.ButtonsType.CLOSE, message_format="Error: %s cannot be found in the path." % INSPECT_PROGRAM) m.run() m.destroy() return True # Try to fetch information about the current symbol # (selected, or at the cursor location) if self.pipebuffer.get_selection_bounds(): w=self.pipebuffer.get_text(*self.pipebuffer.get_selection_bounds() + (False,)) else: w=self.get_word_at_cursor(self.pipebuffer) self.ev.clear_output() data = check_output([ self.inspect_path, w ]).decode('utf-8', 'ignore') if 'No such element' in data: output = check_output([ self.inspect_path ]).decode('utf-8', 'ignore') data="".join(l for l in output.splitlines() if w in l) p.close() self.ev.log("Elements matching %s:\n%s" % (w, data)) else: self.ev.log("Information about %s\n\n" % w) self.ev.log(data) return True def pipeline_focus_cb(widget=None, event=None): if self.ev.source.has_focus(): self.pipewidget.grab_focus() else: self.ev.source.grab_focus() return True self.ev.control_shortcuts[Gdk.KEY_m] = pipe_keypress_cb self.ev.control_shortcuts[Gdk.KEY_f] = pipeline_focus_cb self.ev.__doc__ += "\n Control-f: toggle focus between expression and pipeline definition" self.ev.__doc__ += "\n Control-m: display information about the selected gstreamer element" self.ev.__doc__ += "\n Control-Return in pipeline widget: interpret new pipeline" self.pipewidget=Gtk.TextView () self.pipewidget.set_editable(True) self.pipewidget.set_wrap_mode (Gtk.WrapMode.CHAR) self.pipebuffer=self.pipewidget.get_buffer() # Automatic completion with gst plugin names self.completer=Completer(self.pipewidget, self.get_gst_completions) f=Gtk.Frame.new("Pipeline") s=Gtk.ScrolledWindow() s.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC) s.add(self.pipewidget) f.add(s) self.ev.widget.pack_start(f, True, True, 0) f.show_all() self.statuslabel = Gtk.Label() self.ev.widget.pack_start(self.statuslabel, False, False, 0) self.statuslabel.show() GLib.timeout_add(200, self.update_status) hb=Gtk.HButtonBox() self.ev.widget.pack_start(hb, False, True, 0) hb.show() def refresh_pipeline(*p): pipe = self.pipebuffer.get_text(*self.pipebuffer.get_bounds() + (False,)) self.set_pipeline(pipe, redisplay=False) # Save the pipeline definition to a history file f=open('/tmp/pyGst.log', 'a') f.write(pipe) f.write("\n") f.close() return True for (stock, action) in ( ("_Quit", lambda e: Gtk.main_quit()), ("_Play", lambda e: self.pipeline.set_state(Gst.State.PLAYING)), ("P_ause", lambda e: self.pipeline.set_state(Gst.State.PAUSED)), ("_Refresh", lambda e: refresh_pipeline()), ("_Info", lambda e: self.ev.clear_output() and self.ev.log('Pipeline structure\n-----------------\n' + '\n'.join(self.dump_element(self.pipeline)))), ): b=Gtk.Button.new_with_mnemonic(stock) b.connect("clicked", action) hb.add(b) b.show() # relpath, dump_bin and dump_element implementation based on Daniel Lenski # posted on gst-dev mailing list on 20070913 def relpath(self, p1, p2): sep = os.path.sep # get common prefix (up to a slash) common = os.path.commonprefix((p1, p2)) common = common[:common.rfind(sep)] # remove common prefix p1 = p1[len(common)+1:] p2 = p2[len(common)+1:] # number of seps in p1 is # of ..'s needed return "../" * p1.count(sep) + p2 def dump_bin(self, bin, depth=0, recurse=-1, showcaps=True): return [ l for e in reversed(list(bin.iterate_elements())) for l in self.dump_element(e, depth, recurse - 1) ] def dump_element(self, e, depth=0, recurse=-1, showcaps=True): ret=[] indentstr = depth * 8 * ' ' # print element path and factory path = e.get_path_string() + (isinstance(e, Gst.Bin) and '/' or '') factory = e.get_factory() if factory is not None: ret.append( '%s%s (%s)' % (indentstr, path, factory.get_name()) ) else: ret.append( '%s%s (No factory)' % (indentstr, path) ) # print info about each pad for p in e.pads: name = p.get_name() # negotiated capabilities caps = p.get_current_caps() if caps: capsname = caps.get_structure(0).get_name() elif showcaps: capsname = '; '.join(s.to_string() for s in set(p.get_current_caps() or [])) else: capsname = None # flags flags = [] if not p.is_active(): flags.append('INACTIVE') if p.is_blocked(): flags.append('BLOCKED') # direction direc = (p.get_direction() is Gst.PadDirection.SRC) and "=>" or "<=" # peer peer = p.get_peer() if peer: peerpath = self.relpath(path, peer.get_path_string()) else: peerpath = None # ghost target if isinstance(p, Gst.GhostPad): target = p.get_target() if target: ghostpath = target.get_path_string() else: ghostpath = None else: ghostpath = None line=[ indentstr, " " ] if flags: line.append( ','.join(flags) ) line.append(".%s" % name) if capsname: line.append( '[%s]' % capsname ) if ghostpath: line.append( "ghosts %s" % self.relpath(path, ghostpath) ) line.append( "%s %s" % (direc, peerpath) ) #if peerpath and peerpath.find('proxy')!=-1: print peer ret.append( ''.join(line) ) if recurse and isinstance(e, Gst.Bin): ret.extend( self.dump_bin(e, depth+1, recurse) ) return ret def str_element(self, element): return "\n".join(self.dump_element(element)) class Completer: def __init__(self, textview=None, get_completions=None, abbreviations=None): self.textview = textview self.preferences = {} self.abbreviations = abbreviations self.get_completions = get_completions self.is_visible = False self.word_list = None self.widget = self.build_widget() self.connect() def connect(self): """Register the various callbacks for completion. """ logger.debug("Connecting completer to %s" , self.textview) self.textview.connect('key-press-event', self.key_press_event_cb) self.textview.connect('focus-out-event', self.hide_completion_window) self.textview.connect('paste-clipboard', self.hide_completion_window) self.textview.connect_after('paste-clipboard', self.hide_completion_window) if isinstance(self.textview, Gtk.TextView): self.textview.get_buffer().connect('delete-range', self.hide_completion_window) self.textview.get_buffer().connect_after('insert-text', self.insert_text_cb) else: self.textview.connect('delete-text', self.hide_completion_window) self.textview.connect_after('insert-text', self.insert_text_cb) return True def insert_text_cb(self, textbuffer, iterator, text, length): """Handles callback when the "insert-text" signal is emitted. """ if self.abbreviations is not None and text == ' ': # Find previous word w, begin, end = self.get_word_before_cursor() w = w.strip() repl = self.abbreviations.get(w, None) if repl is not None: textbuffer.delete(begin, end) textbuffer.insert(begin, repl + " ") return False if length > 1: self.hide_completion_window() else: self.check_completion() return False def hide_completion_window(self, *p): self.widget.hide() self.is_visible=False def show_completion_window(self, *p): size = self.treeview.get_preferred_size().natural_size width, height = size.width, size.height width = max(width, 180) width += 24 height += 24 self.widget.resize(width, height) self.widget.set_property("width-request", width) self.widget.set_property("height-request", height) self.position_window(width, height) self.widget.set_size_request(width, height) self.widget.show_all() self.position_window(width, height) self.is_visible=True def get_cursor_rectangle(self): b = self.textview.get_buffer() cursor_iterator=b.get_iter_at_mark(b.get_insert()) rectangle = self.textview.get_iter_location(cursor_iterator) return rectangle def get_cursor_textview_coordinates(self): rectangle=self.get_cursor_rectangle() # Get the cursor's window coordinates. position = self.textview.buffer_to_window_coords(Gtk.TextWindowType.TEXT, rectangle.x, rectangle.y) cursor_x = position[0] cursor_y = position[1] return cursor_x, cursor_y def get_cursor_size(self): """Get the cursor's size. """ rectangle = self.get_cursor_rectangle() return rectangle.width, rectangle.height def position_window(self, width, height): """Position the completion window in the text editor's buffer. @param width: The completion window's width. @type width: int @param height: The completion window's height. @type height: int """ if isinstance(self.textview, Gtk.Entry): allocation = self.textview.get_allocation() origin = self.textview.get_window().get_origin() position_x, position_y = (origin.x + allocation.x, origin.y + allocation.y + allocation.height) cursor_x = 0 cursor_y = 0 else: # Textview # Get the cursor's coordinate and size. cursor_x, cursor_y = self.get_cursor_textview_coordinates() cursor_height = self.get_cursor_size()[1] # Get the text editor's textview coordinate and size. window = self.textview.get_window(Gtk.TextWindowType.TEXT) origin = window.get_origin() # Determine where to position the completion window. position_x = origin.x + cursor_x position_y = origin.y + cursor_y + cursor_height monitor = Gdk.Display.get_default().get_primary_monitor() geometry = monitor.get_geometry() scale_factor = monitor.get_scale_factor() screen_width = scale_factor * geometry.width screen_height = scale_factor * geometry.height if position_x + width > screen_width: position_x = origin.x + cursor_x - width if position_y + height > screen_height: position_y = origin.y + cursor_y - height #if not_(self.__signals_are_blocked): x, y = self.widget.get_position() if position_y != y: position_x = x logger.debug("Position completion window %d %d", position_x, position_y) if position_x != x or position_y != y: # Set the window's new position. self.widget.move(position_x, position_y) def populate_model(self, completion_list): """Populate the view's data model. @param self: Reference to the CompletionTreeView instance. @type self: A CompletionTreeView object. """ if completion_list != self.word_list: self.word_list = completion_list self.model.clear() for word in self.word_list: self.model.append([word]) self.treeview.columns_autosize() self.treeview.get_selection().select_path(0) def get_word_before_cursor(self): """Return the word to complete with its position (word_start, cursor_position) If the component is a TextView/GtkSource, return the position as Gtk.TextIter. If the component is a Gtk.Entry, return the position as int (cursor position) """ if isinstance(self.textview, Gtk.Entry): cursor_position = self.textview.props.cursor_position text = self.textview.get_text()[:cursor_position] rx = re.compile(r'(.*?)([\w\d_]+)$', re.UNICODE) # Find the last word match = rx.search(text) if match: word = match.group(2) word_start = cursor_position - len(word) else: word = "" word_start = cursor_position else: # Gtk.TextView or GtkSource.View b = self.textview.get_buffer() cursor_position = b.get_iter_at_mark(b.get_insert()) word_start=cursor_position.copy() # backward_word_start considers - as a word boundary. # Cool in general, but no so for gstreamer element fragment = word_start.backward_search(" ", 0) if fragment is not None: # Found a space. word_start = fragment[0] # Strip the leading space word_start.forward_char() else: word_start.backward_word_start() word = word_start.get_text(cursor_position) logger.debug("get_word_before_cursor %s %s %s", word, word_start, cursor_position) return word, word_start, cursor_position def insert_word_completion(self, path): """Insert item selected in the completion window into the text editor's buffer. @param path: The selected row in the completion window. @type path: A Gtk.TreeRow object. """ # Get the selected completion string. completion_string = self.model[path[0]][0] word, begin, end = self.get_word_before_cursor() complete = completion_string.replace(word, '', 1) logger.debug("insert_word_completion %s -> %s to %s (%s, %s)", completion_string, complete, word, begin, end) if isinstance(self.textview, Gtk.Entry): self.textview.insert_text(complete, end) self.textview.set_position(end + len(complete)) else: b = self.textview.get_buffer() b.begin_user_action() b.insert_at_cursor(complete) b.end_user_action() return def check_completion(self): word, begin, end = self.get_word_before_cursor() if word: if len(word) < 2: return False matches=sorted(self.get_completions(word), key=len) logger.debug("check_completion %s %s", word, matches) if matches: self.populate_model(matches) self.show_completion_window() else: # Hide the window self.hide_completion_window() else: self.hide_completion_window() return False def key_press_event_cb(self, widget, event): """Handles "key-press-event" for the treeview and textview. This function allows the "Up" and "Down" arrow keys to work in the word completion window. """ logger.debug("key_press_event %s", event.keyval) if not self.is_visible: return False if event.keyval in (Gdk.KEY_Tab, Gdk.KEY_Right, Gdk.KEY_Left, Gdk.KEY_Home, Gdk.KEY_End, Gdk.KEY_Insert, Gdk.KEY_Delete, Gdk.KEY_Page_Up, Gdk.KEY_Page_Down, Gdk.KEY_Escape): self.hide_completion_window() return True # Get the selected item on the completion window. selection = self.treeview.get_selection() # Get the model and iterator of the selected item. model, iterator = selection.get_selected() # If for whatever reason the selection is lost, select the first row # automatically when the up or down arrow key is pressed. if not iterator: selection.select_path((0,)) model, iterator = selection.get_selected() path = model.get_path(iterator) if event.keyval == Gdk.KEY_Return: # Insert the selected item into the editor's buffer when the enter key # event is detected. self.treeview.row_activated(path, self.treeview.get_column(0)) elif event.keyval == Gdk.KEY_Up: # If the up key is pressed check to see if the first row is selected. # If it is, select the last row. Otherwise, get the path to the row # above and select it. if not path[0]: number_of_rows = len(model) selection.select_path(number_of_rows - 1) self.treeview.scroll_to_cell(number_of_rows - 1) else: selection.select_path((path[0] - 1, )) self.treeview.scroll_to_cell((path[0] - 1, )) elif event.keyval == Gdk.KEY_Down: # Get the iterator of the next row. next_iterator = model.iter_next(iterator) # If the next row exists, select it, if not select the first row. if next_iterator: selection.select_iter(next_iterator) path = model.get_path(next_iterator) self.treeview.scroll_to_cell(path) else: selection.select_path(0) self.treeview.scroll_to_cell(0) else: return False return True def build_widget(self): w = Gtk.Window(type=Gtk.WindowType.POPUP) w.set_type_hint(Gdk.WindowTypeHint.MENU) #w.set_size_request(200, 200) self.treeview=Gtk.TreeView() self.model = Gtk.ListStore(str) renderer = Gtk.CellRendererText() col=Gtk.TreeViewColumn("", renderer, text=0) col.set_expand(False) self.treeview.append_column(col) self.treeview.set_headers_visible(False) self.treeview.set_hover_selection(True) self.treeview.set_model(self.model) def treeview_row_activated_cb(treeview, path, column): """Handles "row-activated" in the treeview. """ self.insert_word_completion(path) self.hide_completion_window() return True self.treeview.connect('row-activated', treeview_row_activated_cb) scroll=Gtk.ScrolledWindow() scroll.add(self.treeview) scroll.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC) scroll.set_border_width(2) w.add(scroll) return w if __name__ == '__main__': logging.basicConfig() gstev=GstEvaluator() if sys.argv[1:]: p=' '.join(sys.argv[1:]) gstev.set_pipeline(p) # Redisplay the help message gstev.ev.help() Gtk.main ()