The original problem was QmediaPlayer.setPosition led to the app locking up eventually.
The code below has passed testing which the original code would not. Does not necessarily pass the "fixed yeah" stage as I am suspicious still.
So what was done:?
All media_player methods were bundled into one class
_Video_Player
Class Video_Handler was adjusted accordingly
The key part is in the Video_Handler _setup_media_player method. In the snippet below you will note the two sleep statements , one after moving to thread and one after thread.start. With these no app lockups have yet to be observed.
Hope this helps some one!
self._media_player.moveToThread(self._thread)
sleep(
0.1
) # Note: These are important, without them setPosition in the media player sometimes locks the app
media_player_thread = self._media_player.thread()
is_in_main_thread = media_player_thread == qtC.QThread.currentThread()
print(f"Is in main thread: {is_in_main_thread}")
self._thread.start()
sleep(
0.5
) # Note: These are important, without them setPosition in the media player sometimes locks the app
class _Video_Player(qtC.QObject):
"""
Implements a customer video player object
"""
current_frame_handler = qtC.Signal(int)
duration_changed_handler = qtC.Signal(int)
frame_changed_handler = qtC.Signal(qtM.QVideoFrame)
is_available_handler = qtC.Signal(bool)
media_status_changed_handler = qtC.Signal(qtM.QMediaPlayer.MediaStatus)
pause_handler = qtC.Signal()
play_handler = qtC.Signal()
position_changed_handler = qtC.Signal(int)
seekable_changed_handler = qtC.Signal(bool)
set_position_handler = qtC.Signal(int)
stop_handler = qtC.Signal()
def __init__(self, parent: qtC.QObject | None, input_file: str) -> None:
"""
Sets up the video_player object for use
Args:
parent (qtC.QObject | None): Set the parent of the object
input_file (str): Set the source file of the media player
"""
assert parent is None or isinstance(
parent, qtC.QObject
), f"{parent =} must be None or a qtC.QObject"
assert (
isinstance(input_file, str) and input_file.strip() != ""
), f"{input_file =} must be a non-empty str"
super().__init__(parent)
self._current_position = -1
self._video_sink = qtM.QVideoSink()
self._audio_output = qtM.QAudioOutput()
self._media_player = qtM.QMediaPlayer()
self._media_player.setVideoSink(self._video_sink)
self._media_player.setAudioOutput(self._audio_output)
self._audio_output.setVolume(1)
# Set input source video file
self._media_player.setSource(qtC.QUrl.fromLocalFile(input_file))
# Hook up signals
self._video_sink.videoFrameChanged.connect(self._frame_handler)
self._media_player.durationChanged.connect(self._duration_changed)
self._media_player.positionChanged.connect(self._position_changed)
self._media_player.errorOccurred.connect(self._player_error)
self._media_player.mediaStatusChanged.connect(self._media_status_change)
self._media_player.seekableChanged.connect(self._seekable_changed)
self.frame_changed_handler.connect(self._video_sink.setVideoFrame)
self.is_available_handler.connect(self._media_player.isAvailable)
self.play_handler.connect(self._media_player.play)
self.set_position_handler.connect(self.seek)
self.pause_handler.connect(self._media_player.pause)
self.stop_handler.connect(self.stop)
@qtC.Slot()
def _duration_changed(self, duration: int) -> None:
"""Handles a video duration change
Args:
duration (int): The length of the video
"""
self.duration_changed_handler.emit(duration)
@qtC.Slot()
def _frame_handler(self, frame: qtM.QVideoFrame) -> None:
"""Handles the video frame changing signal
Args:
frame (qtM.QVideoFrame): THe video frame to be displayed
"""
self.frame_changed_handler.emit(frame)
@qtC.Slot()
def _position_changed(self, position_milliseconds: int) -> None:
"""
Handles the position changing signal
Args:
position_milliseconds (int): The current position of the media player in milliseconds.
"""
self.position_changed_handler.emit(position_milliseconds)
@qtC.Slot(qtM.QMediaPlayer.Error, str)
def _player_error(self, error, error_string):
"""Called when the media player encounters an error."""
print(f"Error: {error} - {error_string}")
def available(self) -> bool:
"""
Returns whether the media player is available
Returns:
bool: True if Available, False if noe
"""
return self._media_player.isAvailable()
def current_frame(self) -> int:
return self._media_player.position()
@qtC.Slot()
def _media_status_change(self, media_status: qtM.QMediaPlayer.mediaStatus) -> None:
"""Signals the state of the media has changed
Args:
media_status (qtM.QMediaPlayer.mediaStatus): The status of the media player
"""
self.media_status_changed_handler.emit(media_status)
@qtC.Slot()
def _seekable_changed(self, seekable: bool) -> None:
"""
Signals the seekable status has changed
Args:
seekable (bool): True if the media player is seekable, False otherwise.
"""
self.seekable_changed_handler.emot(seekable)
@qtC.Slot()
def seek(self, position: int) -> None:
"""
Seeks to a position
Args:
position (int): THe position in milliseconds to move to
"""
if self._current_position != position:
if (
self._media_player.isSeekable()
and self._media_player.mediaStatus()
== qtM.QMediaPlayer.MediaStatus.BufferedMedia
):
self._current_position = position
self._media_player.setPosition(position)
def state(self) -> str:
playback_state = self._media_player.playbackState()
if playback_state == qtM.QMediaPlayer.PlaybackState.PlayingState:
return "playing"
elif playback_state == qtM.QMediaPlayer.PlaybackState.PausedState:
return "paused"
elif playback_state == qtM.QMediaPlayer.PlaybackState.StoppedState:
return "stop"
def stop(self):
self._media_player.stop()
self._media_player.setVideoSink(None)
self._media_player.setAudioOutput(None)
@dataclasses.dataclass
class Video_Handler:
aspect_ratio: str
input_file: str
output_edit_folder: str
encoding_info: Encoding_Details
video_display: qtg.Label
video_slider: qtg.Slider
frame_display: qtg.LCD
display_width: int
display_height: int
update_slider: bool = True
source_state: Literal[
"NoMedia",
"Loading",
"Loaded",
"Stalled",
"Buffering",
"Buffered",
"EndOfMedia",
"InvalidMedia",
] = "NoMedia"
state_handler: Callable = None
# Private instance variables
_frame_count: int = 0
_frame_rate: float = 25 # Default to 25 frames per second
_frame_width: int = 720
_frame_height: int = 576
_current_frame: int = -1
def __post_init__(self) -> None:
"""Sets-up the instance"""
assert isinstance(self.aspect_ratio, str) and self.aspect_ratio in (
sys_consts.AR169,
sys_consts.AR43,
), f"{self.aspect_ratio=}. Must be a AR169 | AR43"
assert (
isinstance(self.input_file, str) and self.input_file.strip() != ""
), f"{self.input_file=}. Must be a non-empty str"
assert (
isinstance(self.output_edit_folder, str)
and self.output_edit_folder.strip() != ""
), f"{self.output_edit_folder=}. Must be a non-empty str"
assert isinstance(
self.encoding_info, Encoding_Details
), f"{self.encoding_info=}. Must be an instance of Encoding_Details"
assert isinstance(
self.video_display, qtg.Label
), f"{self.video_display=}. Must be a qtg.Label"
assert isinstance(
self.video_slider, qtg.Slider
), f"{self.video_slider=}. Must be a qtg.Slider"
assert isinstance(
self.frame_display, qtg.LCD
), f"{self.frame_display=}. Must be a qtg.Slider"
assert isinstance(
self.display_width, int
), f"{self.display_width=}. Must be an int"
assert isinstance(
self.display_height, int
), f"{self.display_height=}. Must be an int"
assert isinstance(
self.update_slider, bool
), f"{self.update_slider=}. Must be a bool"
self._frame_width = self.encoding_info.video_width
self._frame_height = self.encoding_info.video_height
self._frame_rate = self.encoding_info.video_frame_rate
self._frame_count = self.encoding_info.video_frame_count
self._setup_media_player()
def _setup_media_player(self):
"""Sets up the media_player instance"""
self._media_player = _Video_Player(parent=None, input_file=self.input_file)
self._media_player.frame_changed_handler.connect(self._frame_handler)
self._media_player.is_available_handler.connect(self.available)
self._media_player.media_status_changed_handler.connect(
self._media_status_change
)
self._media_player.position_changed_handler.connect(self._position_changed)
self._thread = qtC.QThread()
self._media_player.moveToThread(self._thread)
sleep(
0.1
) # Note: These are important, without them setPosition in the media player sometimes locks the app
media_player_thread = self._media_player.thread()
is_in_main_thread = media_player_thread == qtC.QThread.currentThread()
print(f"Is in main thread: {is_in_main_thread}")
self._thread.start()
sleep(
0.5
) # Note: These are important, without them setPosition in the media player sometimes locks the app
@qtC.Slot()
def _frame_handler(self, frame: qtM.QVideoFrame) -> None:
"""Handles displaying the video frame
Args:
frame (qtM.QVideoFrame): THe video frame to be displayed
"""
if frame.isValid():
image = frame.toImage().scaled(self.display_width, self.display_height)
pixmap = qtG.QPixmap.fromImage(image)
if shiboken6.isValid(
self.video_display.guiwidget_get
): # Should not need this check but on shutdown I sometimes got the dreaded C++ object deleted error
self.video_display.guiwidget_get.setPixmap(pixmap)
def _media_status_change(self, media_status: qtM.QMediaPlayer.mediaStatus) -> None:
"""When the status of the media player changes this method sets the source_state var and calls the
state_handler if provided.
Args:
media_status (qtM.QMediaPlayer.mediaStatus): The status of the media player
"""
match media_status:
case qtM.QMediaPlayer.MediaStatus.NoMedia:
self.source_state = "NoMedia"
case qtM.QMediaPlayer.MediaStatus.LoadingMedia:
self.source_state = "Loading"
case qtM.QMediaPlayer.MediaStatus.LoadedMedia:
self.source_state = "Loaded"
case qtM.QMediaPlayer.MediaStatus.StalledMedia:
self.source_state = "Stalled"
case qtM.QMediaPlayer.MediaStatus.BufferingMedia:
self.source_state = "Buffering"
case qtM.QMediaPlayer.MediaStatus.BufferedMedia:
self.source_state = "Buffered"
case qtM.QMediaPlayer.MediaStatus.EndOfMedia:
self.source_state = "EndOfMedia"
case qtM.QMediaPlayer.MediaStatus.InvalidMedia:
self.source_state = "InvalidMedia"
if self.state_handler and isinstance(self.state_handler, Callable):
self.state_handler()
@qtC.Slot()
def _position_changed(self, position_milliseconds: int) -> None:
"""
A method that is called when the position of the media player changes.
Converts the current position in milliseconds to the corresponding frame number,
updates the video slider if necessary, and emits a signal indicating that the position has changed.
Args:
position_milliseconds (int): The current position of the media player in milliseconds.
"""
frame_number = int(position_milliseconds * self._frame_rate // 1000)
if self.update_slider and self.video_slider is not None:
self.video_slider.value_set(frame_number)
self.frame_display.value_set(frame_number)
def get_current_frame(self) -> int:
"""
Returns the current frame number based on the current position of the media player and the frame rate of the video.
Returns:
int: The current frame number.
"""
return int(self._media_player.current_frame() * self._frame_rate // 1000)
def available(self) -> bool:
"""Checks if the media player is supported on the platform
Returns:
bool: True if the media player is supported, False otherwise.
"""
return self._media_player.available()
def play(self) -> None:
"""
Starts playing the media.
"""
self._media_player.play_handler.emit()
return None
def pause(self) -> None:
"""
Pauses the media.
"""
self._media_player.pause_handler.emit()
def seek(self, frame: int) -> None:
"""
Seeks to the specified frame number.
Args:
frame (int): The frame number to seek to.
"""
if self._current_frame != frame:
state = self._media_player.state()
if state == "playing":
self._media_player.pause_handler.emit()
sleep(0.2)
self._current_frame = frame
time_offset = int((1000 / self._frame_rate) * frame)
self._media_player.set_position_handler.emit(time_offset)
if state == "playing":
pass
# self._media_player.play_handler.emit() # Leads to stuttering video sometimes
def shutdown(self) -> None:
"""
Stops playing the media and releases the player's resources.
"""
if self._media_player is not None:
self._media_player.stop_handler.emit()
if self._thread and self._thread.isRunning():
self._thread.quit()
self._thread.wait()
self._thread.deleteLater()
self._thread = None
return None
def state(self) -> str:
"""
Returns the current playback state of the media player.
Returns:
str: The current playback state
- "playing": The media player is currently playing.
- "paused": The media player is currently paused.
- "stop": The media player is currently stopped.
"""
return self._media_player.state()
