Source code for pympress.media_overlays.gst_backend

# -*- coding: utf-8 -*-
#       media_overlays/
#       Copyright 2018 Cimbali <>
#       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
#       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.
:mod:`pympress.media_overlays.gst` -- widget to play videos using Gstreamer's Gst

import logging
logger = logging.getLogger(__name__)

import gi
gi.require_version('Gst', '1.0')
gi.require_version('Gtk', '3.0')
from gi.repository import GLib, Gst

from pympress import util
from pympress.media_overlays import base

[docs] class GstOverlay(base.VideoOverlay): """ Simple Gstramer widget. Wraps a simple gstreamer playbin. """ #: A :class:`~Gst.Playbin` to be play videos playbin = None #: A :class:`~Gst.Base.Sink` to display video content sink = None #: `int` number of milliseconds between updates update_freq = 200 def __init__(self, *args, **kwargs): # Create GStreamer playbin self.playbin = Gst.ElementFactory.make('playbin', None) self.sink = Gst.ElementFactory.make('gtksink', None) self.playbin.set_property('video-sink', self.sink) super(GstOverlay, self).__init__(*args, **kwargs) self.media_overlay.remove(self.movie_zone) self.media_overlay.pack_start(self.sink.props.widget, True, True, 0) self.media_overlay.reorder_child(self.sink.props.widget, 0) self.sink.props.widget.hide() # Create bus to get events from GStreamer playin bus = self.playbin.get_bus() bus.add_signal_watch() bus.enable_sync_message_emission() bus.connect('message::eos', lambda *args: GLib.idle_add(self.handle_end)) bus.connect('message::error', lambda _, msg: logger.error('{} {}'.format(*msg.parse_error()))) bus.connect('message::state-changed', self.on_state_changed) bus.connect('message::duration-changed', lambda *args: GLib.idle_add(self.do_update_duration))
[docs] def is_playing(self): """ Returns whether the media is currently playing (and not paused). Returns: `bool`: `True` iff the media is playing. """ return self.playbin.get_state(0).state == Gst.State.PLAYING
def _set_file(self, filepath): """ Sets the media file to be played by the widget. Args: filepath (`pathlib.Path`): The path to the media file path """ self.playbin.set_property('uri', filepath.as_uri()) self.playbin.set_state(Gst.State.READY)
[docs] def mute(self, value): """ Mutes or unmutes the player. Args: value (`bool`): `True` iff this player should be muted """ flags = self.playbin.get_property('flags') # Fall back to the documented value if introspection fails, # see audio_flag = util.introspect_flag_value(type(flags), 'audio', 0x02) if value: flags = flags & ~audio_flag else: flags = flags | audio_flag self.playbin.set_property('flags', flags) return False
[docs] def on_state_changed(self, bus, msg): """ Callback triggered by playbin state changes. Args: bus (:class:`~Gst.Bus`): the bus that we are connected to msg (:class:`~Gst.Message`): the "state-changed" message """ if msg.src != self.playbin: # ignore the playbin's children return old, new, pending = msg.parse_state_changed() if old == Gst.State.READY and new == Gst.State.PAUSED: # the playbin goes from READY (= stopped) to PLAYING (via PAUSED) self.on_initial_play()
[docs] def on_initial_play(self): """ Set starting position, start scrollbar updates, unhide overlay. """ # set starting position, if needed if self.start_pos: self.do_set_time(self.start_pos) # ensure the scroll bar is updated GLib.idle_add(self.do_update_duration) GLib.timeout_add(self.update_freq, self.do_update_time) # ensure the overlay is visible (if needed) if not self.media_type.startswith('audio'):
[docs] def do_update_duration(self, *args): """ Transmit the change of file duration to the UI to adjust the scroll bar. """ changed, time_ns = self.playbin.query_duration(Gst.Format.TIME) self.update_range(max(0, time_ns) / 1e9)
[docs] def do_update_time(self): """ Update the current position in the progress bar. Returns: `bool`: `True` iff this function should be run again (:func:`~GLib.timeout_add` convention) """ changed, time_ns = self.playbin.query_position(Gst.Format.TIME) time = time_ns / 1e9 self.update_progress(time) if self.last_timestamp <= self.end_pos <= time: self.handle_end() self.last_timestamp = time return self.playbin.get_state(0).state in {Gst.State.PLAYING, Gst.State.PAUSED}
[docs] def do_play(self): """ Start playing the media file. Returns: `bool`: `True` iff this function should be run again (:func:`~GLib.idle_add` convention) """ self.playbin.set_state(Gst.State.PLAYING) return False
[docs] def do_play_pause(self): """ Toggle pause mode of the media. Should run on the main thread to ensure we avoid reentrency problems. Returns: `bool`: `True` iff this function should be run again (:func:`~GLib.idle_add` convention) """ self.playbin.set_state(Gst.State.PLAYING if not self.is_playing() else Gst.State.PAUSED) return False
[docs] def do_stop(self): """ Stops playing in the backend player. """ self.playbin.set_state(Gst.State.NULL) self.playbin.set_state(Gst.State.READY) self.sink.props.widget.hide() return False
[docs] def do_set_time(self, time): """ Set the player at time `~time`. Should run on the main thread to ensure we avoid reentrency problems. Args: time (`float`): the timestamp, in s Returns: `bool`: `True` iff this function should be run again (:func:`~GLib.idle_add` convention) """ self.last_timestamp = time self.playbin.seek_simple(Gst.Format.TIME, Gst.SeekFlags.FLUSH, time * Gst.SECOND) return False
[docs] @classmethod def setup_backend(cls, gst_opts = []): """ Prepare/check the Gst backend. Returns: `str`: the version of Gst used by the backend """ Gst.init(gst_opts) if Gst.ElementFactory.make('gtksink', None) is None: logger.warning('Can not create a gtksink. Check the gtk plugin for GStreamer is installed.') logger.warning('See') raise ValueError('Can not create a gtksink.') return Gst.version_string()