# -*- coding: utf-8 -*-
#
# media_overlays/gst.py
#
# Copyright 2018 Cimbali <me@cimba.li>
#
# 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.
#
"""
: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 https://gstreamer.freedesktop.org/documentation/playback/playsink.html?gi-language=python#GstPlayFlags
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'):
self.sink.props.widget.show()
[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 https://github.com/Cimbali/pympress/issues/240')
raise ValueError('Can not create a gtksink.')
return Gst.version_string()