Source code for pisak.media
"""
Module offering tools for managing and playing different kinds of media data.
"""
from multiprocessing import Process, Manager, Event
from gi.repository import GObject, ClutterGst, Clutter
import pisak
from pisak import res
from pisak import properties, configurator
[docs]class MediaPlaybackIface:
"""
Interface of players of different kinds of media data.
"""
[docs] def is_playing(self):
"""
Return the status of the media stream.
"""
raise NotImplementedError
[docs] def rewind_to(self, position):
"""
Rewind the media stream to the given position.
:param position: new position
"""
raise NotImplementedError
[docs] def rewind_backward(self):
"""
Rewind the media stream backward.
"""
raise NotImplementedError
[docs] def increase_volume(self):
"""
Increase volume of the media stream.
"""
raise NotImplementedError
[docs] def decrease_volume(self):
"""
Decrease volume of the media stream.
"""
raise NotImplementedError
[docs] def set_volume(self, value):
"""
Set volume of the media stream to the given value.
:param value: new volume value
"""
raise NotImplementedError
[docs] def get_elapsed_time(self):
"""
Get elapsed time of the media stream.
"""
raise NotImplementedError
[docs]class MediaPlayback(Clutter.Actor, MediaPlaybackIface,
properties.PropertyAdapter, configurator.Configurable):
"""
Tool for controlling playback of different kinds of media data,
being a wrapper of an inernal ClutterMedia.
"""
__gsignals__ = {
"progressed": (
GObject.SIGNAL_RUN_FIRST, None,
(GObject.TYPE_FLOAT, GObject.TYPE_FLOAT,)),
"started-playing": (
GObject.SIGNAL_RUN_FIRST, None, ()),
"eos": (
GObject.SIGNAL_RUN_FIRST, None, ()),
"limit-declared": (
GObject.SIGNAL_RUN_FIRST, None, (GObject.TYPE_FLOAT,))
}
__gproperties__ = {
"rewind_step": (
GObject.TYPE_FLOAT, None, None,
0, GObject.G_MAXUINT, 2, GObject.PARAM_READWRITE),
"skip_step": (
GObject.TYPE_FLOAT, None, None,
0, GObject.G_MAXUINT, 30, GObject.PARAM_READWRITE),
"volume_step": (
GObject.TYPE_FLOAT, None, None,
0, 1, 0.05, GObject.PARAM_READWRITE),
"engine": (
Clutter.Actor.__gtype__,
"", "", GObject.PARAM_READWRITE)
}
def __init__(self):
super().__init__()
self._engine = None
self.rewind_direction = None
self.rewind_step = 3
self.skip_step = 30
self.volume_step = 0.1
self.volume = pisak.config.as_int('sound_effects_volume') / 100
self.rewind_pace = 200 # pace of rewinding in miliseconds
self.rewind_timer_handler = None
self.rewind_timer = Clutter.Timeline.new(self.rewind_pace)
self.rewind_timer.set_repeat_count(-1)
self.filename = ""
# needed as fail safe for sudden progress jumps
self.previous_progress = 0
self.apply_props()
@property
def engine(self):
return self._engine
@engine.setter
def engine(self, value):
"""
Engine to be used for streaming media content. At the moment it
should be ClutterMedia. When set, default audio volume level is applied
and handler is connected to signal emitted on media progress changes.
:param value: instance of an engine.
"""
self._engine = value
if value is not None:
value.connect("notify::progress", self._on_progressed)
value.set_audio_volume(self.volume)
@property
def rewind_step(self):
"""
Step of rewinding in seconds.
"""
return self._rewind_step
@rewind_step.setter
def rewind_step(self, value):
self._rewind_step = value
@property
def skip_step(self):
"""
Step of skipping in seconds.
"""
return self._skip_step
@skip_step.setter
def skip_step(self, value):
self._skip_step = value
@property
def volume_step(self):
"""
Step of volume changes.
"""
return self._volume_step
@volume_step.setter
def volume_step(self, value):
self._volume_step = value
@property
def filename(self):
"""
Path to the file with media stream. Setting this property
cause loading of a media stream to the player.
"""
return self._filename
@filename.setter
def filename(self, value):
self._filename = value
if value is not None and self.engine is not None:
self._engine.set_filename(value)
self._declare_stream_length()
def _on_progressed(self, source, event):
progress = self._engine.get_progress()
if not progress - self.previous_progress == 1:
self.emit("progressed", progress, progress * self._engine.get_duration())
if progress >= 1:
self._on_eos()
self.previous_progress = progress
def _on_eos(self):
self.rewind_timer.stop()
self.stop()
self.emit("eos")
def _declare_stream_length(self):
self.emit("limit-declared", self._engine.get_duration())
def _rewind_forward(self):
self._move_forward(self.rewind_step)
def _rewind_backward(self):
self._move_backward(self.rewind_step)
def _move_forward(self, step):
try:
to = min(self._engine.get_progress() +
step / self._engine.get_duration(), 1)
self._engine.set_progress(to)
except ZeroDivisionError:
pass
def _move_backward(self, step):
try:
to = max(self._engine.get_progress() -
step / self._engine.get_duration(), 0)
self._engine.set_progress(to)
except ZeroDivisionError:
pass
def _toggle_rewind(self, where):
was_rewinding = self.rewind_timer.is_playing()
if was_rewinding:
self.rewind_timer.stop()
if not was_rewinding or self.rewind_direction != where:
if self.rewind_timer_handler is not None:
self.rewind_timer.disconnect(self.rewind_timer_handler)
self.rewind_timer_handler = \
self.rewind_timer.connect("completed", lambda *_: where())
self.rewind_timer.start()
self.rewind_direction = where
[docs] def get_duration(self):
"""
Get duration of the media stream in seconds.
:return: duration, integer.
"""
return self._engine.get_duration()
[docs] def move_to(self, position):
"""
Move to the given position.
:param position: normalized value between 0 and 1.
"""
self._engine.set_progress(position)
[docs] def get_elapsed_time(self):
"""
Get media stream elapsed time in seconds.
:return: elapsed time, float.
"""
return self._engine.get_duration() * self._engine.get_progress()
[docs] def play(self):
"""
Start playing the media stream.
"""
self.stop_rewind()
self._engine.set_playing(True)
self.emit("started-playing")
[docs] def stop(self):
"""
Stop playing the media stream and move to the beginning.
"""
self.stop_rewind()
self._engine.set_playing(False)
self._engine.set_progress(0)
[docs] def pause(self):
"""
Pause playing of the media stream.
"""
self.stop_rewind()
self._engine.set_playing(False)
[docs] def is_playing(self):
"""
Return the status of the media stream.
:return: boolean.
"""
return self._engine.get_playing()
[docs] def set_volume(self, value):
"""
Set volume of the media stream.
:param value: normalized volume value between 0 and 1.
"""
self._engine.set_audio_volume(value)
[docs] def skip_forward(self):
"""
Skip the media stream forward by the skip step.
"""
self._move_forward(self.skip_step)
[docs] def skip_backward(self):
"""
Skip the media stream backward by the skip step.
"""
self._move_backward(self.skip_step)
[docs] def toggle_rewind_forward(self):
"""
Start or stop rewind the media stream forward.
"""
self._toggle_rewind(self._rewind_forward)
[docs] def toggle_rewind_backward(self):
"""
Start or stop rewind the media stream backward.
"""
self._toggle_rewind(self._rewind_backward)
[docs] def increase_volume(self):
"""
Increase volume of the media stream by the volume step.
"""
self.volume = min(self.volume + self.volume_step, 1)
self._engine.set_audio_volume(self.volume)
pisak.config['sound_effects_volume'] = str(int(self.volume * 100))
[docs] def decrease_volume(self):
"""
Decrease volume of the media stream by the volume step.
"""
self.volume = max(self.volume - self.volume_step, 0)
self._engine.set_audio_volume(self.volume)
pisak.config['sound_effects_volume'] = str(int(self.volume * 100))
[docs] def stop_rewind(self):
"""
Stop any on-going process of media stream rewinding.
"""
if self.rewind_timer.is_playing():
self.rewind_timer.stop()
[docs]class AudioPlayback(MediaPlayback):
"""
Tool for controlling playback of audio stream.
"""
__gtype_name__ = "PisakAudioPlayback"
def __init__(self):
super().__init__()
self.engine = ClutterGst.VideoTexture()
[docs]class VideoPlayback(MediaPlayback):
"""
Tool for controlling playback of video stream. Out of order, so far.
"""
__gtype_name__ = "PisakVideoPlayback"
def __init__(self):
super().__init__()
[docs] def set_subtitle_from_file(self, path):
"""
Set subtitle for the current video stream. Format
of a subtitle file should be one of these: TXT.
:param path: path to a file containing subtitle.
"""
self._engine.set_subtitle_uri(path)
[docs] def set_subtitle_font(self, font):
"""
Set font of a subtitle.
:param font: font of a subtitle.
"""
self._engine.set_subtitle_font_name(font)