Compare commits
No commits in common. "60c085ad1205d4e2af16b94986b0448c6b2b51db" and "b86b7f7f331ad19052382b5d5a2011829e537a8b" have entirely different histories.
60c085ad12
...
b86b7f7f33
125
app/helpers.py
125
app/helpers.py
@ -3,6 +3,7 @@ from email.message import EmailMessage
|
||||
from typing import Any, Dict, Optional
|
||||
import functools
|
||||
import os
|
||||
import psutil
|
||||
import re
|
||||
import shutil
|
||||
import smtplib
|
||||
@ -21,80 +22,14 @@ from log import log
|
||||
|
||||
start_time_re = re.compile(r"@\d\d:\d\d")
|
||||
|
||||
|
||||
class AudacityManager:
|
||||
"""
|
||||
Manage comms with Audacity
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""
|
||||
Open passed file in Audacity
|
||||
|
||||
Return True if apparently opened successfully, else False
|
||||
"""
|
||||
|
||||
self.to_pipe: str = "/tmp/audacity_script_pipe.to." + str(os.getuid())
|
||||
self.from_pipe: str = "/tmp/audacity_script_pipe.from." + str(os.getuid())
|
||||
self.eol: str = "\n"
|
||||
self.path = ""
|
||||
|
||||
def open_file(self, path: str) -> str:
|
||||
"""Open passed file in Audacity"""
|
||||
|
||||
self.path = path
|
||||
return self._do_command(f'Import2: Filename="{self.path}"')
|
||||
|
||||
def export_file(self) -> str:
|
||||
"""Export current file"""
|
||||
|
||||
if not self.path:
|
||||
return "Error: no path selected"
|
||||
|
||||
sa_response = self._do_command("SelectAll:")
|
||||
if sa_response == "\nBatchCommand finished: OK\n":
|
||||
exp_response = self._do_command(
|
||||
f'Export2: Filename="{self.path}" NumChannels=2'
|
||||
)
|
||||
return exp_response
|
||||
else:
|
||||
return "SelectAll response: " + sa_response
|
||||
|
||||
def _send_command(self, command: str) -> None:
|
||||
"""Send a single command."""
|
||||
self.to_audacity.write(command + self.eol)
|
||||
self.to_audacity.flush()
|
||||
|
||||
def _get_response(self) -> str:
|
||||
"""Return the command response."""
|
||||
|
||||
result: str = ""
|
||||
line: str = ""
|
||||
|
||||
while True:
|
||||
result += line
|
||||
line = self.from_audacity.readline()
|
||||
if line == "\n" and len(result) > 0:
|
||||
break
|
||||
return result
|
||||
|
||||
def _do_command(self, command: str) -> str:
|
||||
"""Send one command, and return the response."""
|
||||
|
||||
with open(self.to_pipe, "w") as self.to_audacity, open(
|
||||
self.from_pipe, "rt"
|
||||
) as self.from_audacity:
|
||||
self._send_command(command)
|
||||
response = self._get_response()
|
||||
return response
|
||||
# Classes are defined after global functions so that classes can use
|
||||
# those functions.
|
||||
|
||||
|
||||
def ask_yes_no(
|
||||
title: str, question: str, default_yes: bool = False, parent: Optional[QMainWindow] = None
|
||||
) -> bool:
|
||||
def ask_yes_no(title: str, question: str, default_yes: bool = False) -> bool:
|
||||
"""Ask question; return True for yes, False for no"""
|
||||
|
||||
dlg = QMessageBox(parent)
|
||||
dlg = QMessageBox()
|
||||
dlg.setWindowTitle(title)
|
||||
dlg.setText(question)
|
||||
dlg.setStandardButtons(
|
||||
@ -379,6 +314,56 @@ def normalise_track(path):
|
||||
os.remove(temp_path)
|
||||
|
||||
|
||||
def open_in_audacity(path: str) -> bool:
|
||||
"""
|
||||
Open passed file in Audacity
|
||||
|
||||
Return True if apparently opened successfully, else False
|
||||
"""
|
||||
|
||||
# Return if audacity not running
|
||||
if "audacity" not in [i.name() for i in psutil.process_iter()]:
|
||||
return False
|
||||
|
||||
# Return if path not given
|
||||
if not path:
|
||||
return False
|
||||
|
||||
to_pipe: str = "/tmp/audacity_script_pipe.to." + str(os.getuid())
|
||||
from_pipe: str = "/tmp/audacity_script_pipe.from." + str(os.getuid())
|
||||
eol: str = "\n"
|
||||
|
||||
def send_command(command: str) -> None:
|
||||
"""Send a single command."""
|
||||
to_audacity.write(command + eol)
|
||||
to_audacity.flush()
|
||||
|
||||
def get_response() -> str:
|
||||
"""Return the command response."""
|
||||
|
||||
result: str = ""
|
||||
line: str = ""
|
||||
|
||||
while True:
|
||||
result += line
|
||||
line = from_audacity.readline()
|
||||
if line == "\n" and len(result) > 0:
|
||||
break
|
||||
return result
|
||||
|
||||
def do_command(command: str) -> str:
|
||||
"""Send one command, and return the response."""
|
||||
|
||||
send_command(command)
|
||||
response = get_response()
|
||||
return response
|
||||
|
||||
with open(to_pipe, "w") as to_audacity, open(from_pipe, "rt") as from_audacity:
|
||||
do_command(f'Import2: Filename="{path}"')
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def send_mail(to_addr, from_addr, subj, body):
|
||||
# From https://docs.python.org/3/library/email.examples.html
|
||||
|
||||
|
||||
@ -29,6 +29,7 @@ from helpers import (
|
||||
file_is_unreadable,
|
||||
get_embedded_time,
|
||||
get_relative_date,
|
||||
open_in_audacity,
|
||||
ms_to_mmss,
|
||||
set_track_metadata,
|
||||
)
|
||||
@ -983,6 +984,15 @@ class PlaylistModel(QAbstractTableModel):
|
||||
log.error(f"OBS SDK error ({e})")
|
||||
return
|
||||
|
||||
def open_in_audacity(self, row_number: int) -> None:
|
||||
"""
|
||||
Open track at passed row number in Audacity
|
||||
"""
|
||||
|
||||
path = self.playlist_rows[row_number].path
|
||||
if path:
|
||||
open_in_audacity(path)
|
||||
|
||||
def previous_track_ended(self) -> None:
|
||||
"""
|
||||
Notification from musicmuster that the previous track has ended.
|
||||
@ -1517,6 +1527,9 @@ class PlaylistProxyModel(QSortFilterProxyModel):
|
||||
header_row_number, existing_prd, note
|
||||
)
|
||||
|
||||
def open_in_audacity(self, row_number: int) -> None:
|
||||
return self.data_model.open_in_audacity(row_number)
|
||||
|
||||
def previous_track_ended(self) -> None:
|
||||
return self.data_model.previous_track_ended()
|
||||
|
||||
|
||||
@ -1,6 +1,5 @@
|
||||
import psutil
|
||||
from pprint import pprint
|
||||
from typing import Callable, cast, List, Optional, TYPE_CHECKING
|
||||
from typing import Callable, cast, List, Optional, overload, TYPE_CHECKING
|
||||
|
||||
from PyQt6.QtCore import (
|
||||
QEvent,
|
||||
@ -8,7 +7,6 @@ from PyQt6.QtCore import (
|
||||
QObject,
|
||||
QItemSelection,
|
||||
Qt,
|
||||
QTimer,
|
||||
)
|
||||
from PyQt6.QtGui import QAction, QKeyEvent
|
||||
from PyQt6.QtWidgets import (
|
||||
@ -35,7 +33,6 @@ from classes import MusicMusterSignals, track_sequence
|
||||
from config import Config
|
||||
from helpers import (
|
||||
ask_yes_no,
|
||||
AudacityManager,
|
||||
ms_to_mmss,
|
||||
show_OK,
|
||||
show_warning,
|
||||
@ -205,7 +202,6 @@ class PlaylistTab(QTableView):
|
||||
# Load playlist rows
|
||||
self.setModel(self.proxy_model)
|
||||
self._set_column_widths()
|
||||
QTimer.singleShot(0, lambda: self.resizeRowsToContents())
|
||||
|
||||
# ########## Overrident class functions ##########
|
||||
|
||||
@ -355,7 +351,7 @@ class PlaylistTab(QTableView):
|
||||
# Open in Audacity
|
||||
if track_row and not current_row:
|
||||
self._add_context_menu(
|
||||
"Open in Audacity", lambda: self.open_in_audacity(model_row_number)
|
||||
"Open in Audacity", lambda: model.open_in_audacity(model_row_number)
|
||||
)
|
||||
|
||||
# Rescan
|
||||
@ -526,20 +522,11 @@ class PlaylistTab(QTableView):
|
||||
return self.data_model.get_row_track_path(model_row_number)
|
||||
|
||||
def get_selected_rows(self) -> List[int]:
|
||||
"""Return a list of model-selected row numbers sorted by row"""
|
||||
"""Return a list of selected row numbers sorted by row"""
|
||||
|
||||
# Use a set to deduplicate result (a selected row will have all
|
||||
# items in that row selected)
|
||||
return sorted(
|
||||
list(
|
||||
set(
|
||||
[
|
||||
self.proxy_model.mapToSource(a).row()
|
||||
for a in self.selectedIndexes()
|
||||
]
|
||||
)
|
||||
)
|
||||
)
|
||||
return sorted(list(set([a.row() for a in self.selectedIndexes()])))
|
||||
|
||||
def _info_row(self, row_number: int) -> None:
|
||||
"""Display popup with info re row"""
|
||||
@ -566,30 +553,6 @@ class PlaylistTab(QTableView):
|
||||
self.data_model.mark_unplayed(row_numbers)
|
||||
self.clear_selection()
|
||||
|
||||
def open_in_audacity(self, row_number: int) -> None:
|
||||
"""
|
||||
Open track in passed row in Audacity
|
||||
"""
|
||||
|
||||
# Notify user if audacity not running
|
||||
if "audacity" not in [i.name() for i in psutil.process_iter()]:
|
||||
show_warning(self.musicmuster, "Audacity", "Audacity is not running")
|
||||
return
|
||||
|
||||
path = self.data_model.get_row_track_path(row_number)
|
||||
if not path:
|
||||
return
|
||||
|
||||
audacity = AudacityManager()
|
||||
audacity.open_file(path)
|
||||
if ask_yes_no(
|
||||
"Export file",
|
||||
"Click yes to export file, no to ignore",
|
||||
parent=self.musicmuster,
|
||||
):
|
||||
audacity.export_file()
|
||||
self._rescan(row_number)
|
||||
|
||||
def _rescan(self, row_number: int) -> None:
|
||||
"""Rescan track"""
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user