qop package

Submodules

qop.constants module

This module defines global constants and enums. The numeric values of the integer enums are used to represent the respective value when serializing qop objects into a TaskQueue sqlite3 database.

class qop.constants.ConverterType(value)

Bases: IntEnum

Types of converters. Corresponds to the subclasses of Converter.

COPY = 0

PYDUB = 1

class qop.constants.Command(value)

Bases: IntEnum

Commands codes that can be sent to the QopDaemon via a CommandMessage. See QopClient for usage examples.

DAEMON_START = 101

DAEMON_STOP = 102

DAEMON_IS_ACTIVE = 103

DAEMON_FACTS = 104

QUEUE_START = 201

QUEUE_STOP = 202

QUEUE_IS_ACTIVE = 203

QUEUE_PUT = 204

QUEUE_FLUSH_PENDING = 205

QUEUE_FLUSH_ALL = 206

QUEUE_PROGRESS = 207

QUEUE_ACTIVE_PROCESSES = 208

QUEUE_SHOW = 209

QUEUE_MAX_PROCESSES = 210

class qop.constants.PayloadClass(value)

Bases: IntEnum

Types of payloads that can be part of a CommandMessage or a StatusMessage.

VALUE = 1

TASK = 2

QUEUE_PROGRESS = 3

TASK_LIST = 4

DAEMON_FACTS = 5

class qop.constants.Status(value)

Bases: IntEnum

Status codes are used by

TaskQueue to represent the status of a task in the queue, and

StatusMessage to indicate whether a command sent to the server was successful or not

FAIL = -1

PENDING = 0

OK = 1

SKIP = 2

ACTIVE = 3

class qop.constants.TaskType(value)

Bases: IntEnum

Types of tasks. Corresponds to the subclasses of Task.

ECHO = 1

FILE = 2

DELETE = 3

COPY = 4

MOVE = 5

CONVERT_SIMPLE = 6

FAIL = 7

SLEEP = 8

CONVERT = 9

qop.converters module

Converters are used by ConvertTask and SimpleConvertTask to transcode audiofiles

class qop.converters.Converter

Bases: object

Abstract base class for Converters

remove_art = False

to_dict() → Dict

static from_dict(x: Dict) → Converter: Create a Converter object from python dict that contains the necessary keys

static from_json(s: str) → Converter: Deserialize a Converter from JSON

start(src: Path | str, dst: Path | str)

class qop.converters.CopyConverter(remove_art: bool = False)

Bases: Converter

Converter that copies a file without transcoding (but may modify the files tags!)

start(src: Path | str, dst: Path | str)

to_dict() → Dict

Bases: CopyConverter

Convert audio files using pydub. See pydub.AudioSegment.export() (link) for more details on the meaning of the parameters. Defaults to mp3 via lame with V0 quality (best possible VBR quality).

Param:: remove_art Remove all album art (image) tags during conversion
Param:: parameters named arguments passed on to pydub.AudioSegment.export() (and from there to ffmpeg) when starting the conversion.

property ext: str

start(src: Path | str, dst: Path | str) → None

to_dict() → Dict

qop.daemon module

This module defines classes for the daemon and the client that the command line programs provided by qop are based on.

The client sends CommandMessages via TCP to the daemon to enqueue new tasks, tell it to start/stop processing the queue, query information, etc… .

The daemon manages a single TaskQueue that stores and processes Tasks. For every command it receives, it responds with a StatusMessage that contains the status of the operation (ok, fail, skip) as well as an optional JSON payload.

Communication diagram:

+-------------+                        +-------------+   +-------------+
|             | ---[CommandMessage]--> |             |   |             |
|  QopClient  |                        |  QopDaemon  |---|  TaskQueue  |
|             | <--[StatusMessage]---- |             |   |             |
+-------------+                        +-------------+   +-------------+

class qop.daemon.QopDaemon(port: int = 9393, queue_path: Path | str = PosixPath('/tmp/qop-temp.sqlite3'), persist_queue: bool = False)

Bases: object

stats = None

queue = None

port = None

close()

listen(port=9393)

static handle_request(req)

new_queue(path: Path)

facts() → Dict

class qop.daemon.QopClient(ip: str = '127.0.0.1', port: int = 9393)

Bases: object

A simple client for interacting with QopDaemon. Usage example:

tsk = tasks.EchoTask("foo")
client = daemon.QopClient()
client.send_command(Command.QUEUE_PUT, payload=tsk)
client.send_command(Command.QUEUE_START) 

gather_facts(max_tries=10) → Dict

get_queue_progress(max_tries=10) → QueueProgress

is_daemon_active() → bool

property max_processes: list

property active_tasks: list

property active_processes: int

is_queue_active() → int

send_command(command: Command, payload: None | Dict | Task | list = None) → Dict

class qop.daemon.Message(body: Dict, extra_headers: Dict | None = None)

Bases: object

Container for messages sent between QopDaemon and QopClient.

encode() → bytes

static from_bytes(x: bytes) → Message

class qop.daemon.StatusMessage(status: int, msg: str | None = None, payload=None, payload_class=None)

Bases: Message

Class responses sent by QopDaemon to QopClient.

class qop.daemon.CommandMessage(command: Command, payload=None, payload_class=None)

Bases: Message

Class for commands sent by QopClient to QopDaemon.

qop.exceptions module

Exceptions

exception qop.exceptions.AlreadyUnderEvaluationError

Bases: Exception

This Task is already being processed by a different worker

exception qop.exceptions.FileExistsAndShouldBeSkippedError: Bases: Exception

exception qop.exceptions.FileExistsAndIsIdenticalError: Bases: FileExistsAndShouldBeSkippedError

exception qop.exceptions.FileExistsAndCannotBeComparedError: Bases: FileExistsAndShouldBeSkippedError

qop.scanners module

Used by qop.py to traverse the directory tree when looking for files to transfer

class qop.scanners.Scanner

Bases: object

scan(root: Path | str) → Generator[Path, None, None]

class qop.scanners.PassScanner

Bases: Scanner

scan(root: Path | str) → Generator[Path, None, None]

class qop.scanners.ExcludeScanner(exts: list)

Bases: Scanner

scan(root: Path | str) → Generator[Path, None, None]

class qop.scanners.IncludeScanner(exts: list)

Bases: Scanner

scan(root: Path | str) → Generator[Path, None, None]

qop.tasks module

This module defines classes for working with Tasks. A task is an atomic operation - such as copying, moving or transcoding a file - that can be stored and executed at a later date (and usually only once).

TaskQueue is a persistent (via sqlite3), prioritized queue with multiprocessing support that can store and execute an arbitrary number of tasks.

class qop.tasks.TaskQueue(path: Path | str, max_transfer_processes=1, max_convert_processes=3)

Bases: object

CONVERT_CACHE_DIR = Path(appdirs.user_cache_dir(“qop”)).joinpath(“convert_temp”) A persistent, prioritized queue with multi process support. Use sqlite3 as a storage backend.

Parameters:

transfer_processes – A list of file transfer processes
convert_processes – A list of audio transcode processes

transfer_processes = []

convert_processes = []

put(task: Task, priority: int = 10, parent: int | None = None) → None

Enqueue a Task

Parameters:

task – Task to be added to the queue
priority – (optional) priority for executing task (tasks with lower priority will be executed earlier)
parent – (optional) only for child tasks, oid/_ROWID_ of the task that spawned this task

pop(task_type_include: TaskType | None = None, task_type_exclude: TaskType | None = None) → Task

Retrieves a Task and sets its status in the queue to Status.ACTIVE

Raises:: AlreadyUnderEvaluationError – If trying to pop a tasks that is already being processed (i.e. if a race condition occurs if the queue is processed in parallel)

peek() → Task: Retrieves a Task without changing its status in the queue

print(status: Tuple | int | None = None, n: int = 10) → None

Print an overview of the queue

Parameters:

n – number of tasks to fetch
status – If not None, only fetch Tasks of the given status(es)

fetch(status: Tuple | int | Status | None = None, n: int | None = None) → List

Retrieve the queue

Parameters:

n – number of tasks to fetch
status – If not None, only fetch Tasks of the given status(es)

:return a dict containing n queued tasks

reset_active_tasks() → None: Reset all active tasks to Status.PENDING

set_status(oid: int, status: Status, lock: str | None = None) → None

Set the Status of the queued task with _ROWID_ oid

Parameters:

oid – _ROWID_ of the task to mark
status – Status to set
lock – Unique ID for locking queued tasks. The lock value can be querried by the executing task to verify that it is the only tasks that attempts to run this specific task. Must be None except for switching tasks to active.

start(ip=None, port=None) → None: Execute all pending tasks

stop() → None

flush(status: Status | int | None = None) → None: empty the queue

facts() → Dict

is_active() → bool

active_processes(type=None)

property n_total: int: Count of all tasks in queue (including failed and completed)

property n_pending: int: Number of pending tasks

property n_active: int: Count of currently active tasks

property n_ok: int: count of completed tasks

property n_fail: int: count of completed tasks

progress(include_children: bool = False) → QueueProgress

class qop.tasks.Task

Bases: object

Abstract class for qop Tasks. Should not be instantiated directly.

oid = None

parent_oid = None

start() → None: Run a task

static from_dict(x: Dict) → Task: Create a Task of the appropriate subclass from a python dict

to_dict() → Dict

color_repr(color=True)

to_json()

class qop.tasks.EchoTask(msg: str)

Bases: Task

Log a message

start() → None: Run a task

color_repr(color=True)

class qop.tasks.SleepTask(seconds: float)

Bases: Task

Log a message

start() → None: Run a task

color_repr(color=True)

class qop.tasks.FailTask

Bases: Task

Log a message

start() → None: Run a task

class qop.tasks.FileTask(src: Path | str)

Bases: Task

Abstract class for all file-based tasks

start() → None: Run a task

class qop.tasks.DeleteTask(src: Path | str, validate: bool = True)

Bases: FileTask

Delete a file

start() → None: Run a task

class qop.tasks.CopyTask(src: Path | str, dst: Path | str)

Bases: FileTask

Copy a file

color_repr(color=True) → str

start() → None: Run a task

class qop.tasks.MoveTask(src: Path | str, dst: Path | str, parent_oid=None)

Bases: CopyTask

Move a file

start() → None: Run a task

color_repr(color=True) → str

class qop.tasks.SimpleConvertTask(src: Path | str, dst: Path | str, converter: Converter)

Bases: CopyTask

convert an audio file

start() → None: Run a task

color_repr(color=True) → str

to_dict() → Dict

class qop.tasks.ConvertTask(src: Path | str, dst: Path | str, converter: Converter, tempdir=PosixPath('/home/runner/.cache/qop/convert_temp'))

Bases: SimpleConvertTask

ConvertTask transcodes an audio file to a temporary directory and then adds a move task to the queue. This makes it possible to cleanly separate transcode and transfer processes.

start() → None: Run a task

spawn() → MoveTask: Spawns a follow-up task. If a task has a spawn() method, TaskQueue calls it internally after fetching the original Task. The follow-up task is then enqueued with maximum priority. If this newly spawned task fails, the parent task will also be considered failed.

color_repr(color=True) → str

to_dict() → Dict

class qop.tasks.TaskQueueElement(task: Task, priority: 1)

Bases: object

An enqueued Task

status = None: Status of the queued Task

task = None: A Task

priority = None: Priority of the queued Task

class qop.tasks.QueueProgress(pending: int, ok: int, skip: int, fail: int, active: int)

Bases: object

Info on the current status of the Queue

property total

static from_dict(x: Dict) → QueueProgress

static from_list(x: List) → QueueProgress: Convert a list (for example as returned by an SQL SELECT statement) to a QueueProgress object

to_dict() → Dict

fmt_summary()

qop.tasks.hammer_commit(con, max_tries=10)

exception qop.tasks.UnknownTaskTypeError: Bases: ValueError