Module `calvo_cli_tools.jack_tools.timebase_master`

A simple JACK timebase master.

Expand source code

#!/usr/bin/env python3
#
#  timebase_master.py
#
# MIT License

# Copyright (c) 2019 Christopher Arndt

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""A simple JACK timebase master."""

import argparse
import sys
from threading import Event

import jack


class TimebaseMasterClient(jack.Client):
    def __init__(self, name, *, bpm=120.0, beats_per_bar=4, beat_type=4,
                 ticks_per_beat=1920, conditional=False, debug=False, **kw):
        super().__init__(name, **kw)
        self.beats_per_bar = int(beats_per_bar)
        self.beat_type = int(beat_type)
        self.bpm = bpm
        self.conditional = conditional
        self.debug = debug
        self.ticks_per_beat = int(ticks_per_beat)
        self.stop_event = Event()
        self.set_shutdown_callback(self.shutdown)

    def shutdown(self, status, reason):
        print('JACK shutdown:', reason, status)
        self.stop_event.set()

    def _tb_callback(self, state, nframes, pos, new_pos):
        if self.debug and new_pos:
            print("New pos:", jack.position2dict(pos))

        # Adapted from:
        # https://github.com/jackaudio/jack2/blob/develop/example-clients/transport.c#L66
        if new_pos:
            pos.beats_per_bar = float(self.beats_per_bar)
            pos.beats_per_minute = self.bpm
            pos.beat_type = float(self.beat_type)
            pos.ticks_per_beat = float(self.ticks_per_beat)
            pos.valid |= jack._lib.JackPositionBBT

            minutes = pos.frame / (pos.frame_rate * 60.0)
            abs_tick = minutes * self.bpm * self.ticks_per_beat
            abs_beat = abs_tick / self.ticks_per_beat

            pos.bar = int(abs_beat / self.beats_per_bar)
            pos.beat = int(abs_beat - (pos.bar * self.beats_per_bar) + 1)
            pos.tick = int(abs_tick - (abs_beat * self.ticks_per_beat))
            pos.bar_start_tick = pos.bar * self.beats_per_bar * self.ticks_per_beat
            pos.bar += 1  # adjust start to bar 1
        else:
            # Compute BBT info based on previous period.
            pos.tick += int(nframes * pos.ticks_per_beat *
                            pos.beats_per_minute / (pos.frame_rate * 60))

            while pos.tick >= pos.ticks_per_beat:
                pos.tick -= int(pos.ticks_per_beat)
                pos.beat += 1

                if pos.beat > pos.beats_per_bar:
                    pos.beat = 1
                    pos.bar += 1
                    pos.bar_start_tick += pos.beats_per_bar * pos.ticks_per_beat

                    if self.debug:
                        print("Pos:", jack.position2dict(pos))

    def become_timebase_master(self, conditional=None):
        return self.set_timebase_callback(self._tb_callback, conditional
                                          if conditional is not None
                                          else self.conditional)


def main(args=None):
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument(
        '-d', '--debug',
        action='store_true',
        help="Enable debug messages")
    ap.add_argument(
        '-c', '--conditional',
        action='store_true',
        help="Exit if another timebase master is already active")
    ap.add_argument(
        '-n', '--client-name',
        metavar='NAME',
        default='timebase',
        help="JACK client name (default: %(default)s)")
    ap.add_argument(
        '-m', '--meter',
        default='4/4',
        help="Meter as <beats-per-bar>/<beat-type> (default: %(default)s)")
    ap.add_argument(
        '-t', '--ticks-per-beat',
        type=int,
        metavar='NUM',
        default=1920,
        help="Ticks per beat (default: %(default)s)")
    ap.add_argument(
        'tempo',
        nargs='?',
        type=float,
        default=120.0,
        help="Tempo in beats per minute (0.1-300.0, default: %(default)s)")

    args = ap.parse_args(args)

    try:
        beats_per_bar, beat_type = (int(x) for x in args.meter.split('/', 1))
    except (TypeError, ValueError):
        print("Error: invalid meter: {}\n".format(args.meter))
        ap.print_help()
        return 2

    try:
        tbmaster = TimebaseMasterClient(
            args.client_name,
            bpm=max(0.1, min(300.0, args.tempo)),
            beats_per_bar=beats_per_bar,
            beat_type=beat_type,
            ticks_per_beat=args.ticks_per_beat,
            debug=args.debug)
    except jack.JackError as exc:
        return "Could not create timebase master JACK client: {}".format(exc)

    with tbmaster:
        if tbmaster.become_timebase_master(args.conditional):
            try:
                print("Press Ctrl-C to quit...")
                tbmaster.stop_event.wait()
            except KeyboardInterrupt:
                print('')
            finally:
                try:
                    tbmaster.release_timebase()
                except jack.JackError:
                    # another JACK client might have grabbed timebase master
                    pass
        else:
            return "Timebase master already present. Exiting..."


if __name__ == '__main__':
    sys.exit(main() or 0)

Functions

def main(args=None)

Expand source code

def main(args=None):
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument(
        '-d', '--debug',
        action='store_true',
        help="Enable debug messages")
    ap.add_argument(
        '-c', '--conditional',
        action='store_true',
        help="Exit if another timebase master is already active")
    ap.add_argument(
        '-n', '--client-name',
        metavar='NAME',
        default='timebase',
        help="JACK client name (default: %(default)s)")
    ap.add_argument(
        '-m', '--meter',
        default='4/4',
        help="Meter as <beats-per-bar>/<beat-type> (default: %(default)s)")
    ap.add_argument(
        '-t', '--ticks-per-beat',
        type=int,
        metavar='NUM',
        default=1920,
        help="Ticks per beat (default: %(default)s)")
    ap.add_argument(
        'tempo',
        nargs='?',
        type=float,
        default=120.0,
        help="Tempo in beats per minute (0.1-300.0, default: %(default)s)")

    args = ap.parse_args(args)

    try:
        beats_per_bar, beat_type = (int(x) for x in args.meter.split('/', 1))
    except (TypeError, ValueError):
        print("Error: invalid meter: {}\n".format(args.meter))
        ap.print_help()
        return 2

    try:
        tbmaster = TimebaseMasterClient(
            args.client_name,
            bpm=max(0.1, min(300.0, args.tempo)),
            beats_per_bar=beats_per_bar,
            beat_type=beat_type,
            ticks_per_beat=args.ticks_per_beat,
            debug=args.debug)
    except jack.JackError as exc:
        return "Could not create timebase master JACK client: {}".format(exc)

    with tbmaster:
        if tbmaster.become_timebase_master(args.conditional):
            try:
                print("Press Ctrl-C to quit...")
                tbmaster.stop_event.wait()
            except KeyboardInterrupt:
                print('')
            finally:
                try:
                    tbmaster.release_timebase()
                except jack.JackError:
                    # another JACK client might have grabbed timebase master
                    pass
        else:
            return "Timebase master already present. Exiting..."

Classes

class TimebaseMasterClient (name, *, bpm=120.0, beats_per_bar=4, beat_type=4, ticks_per_beat=1920, conditional=False, debug=False, **kw)

A client that can connect to the JACK audio server.

Create a new JACK client.

A client object is a context manager, i.e. it can be used in a with statement to automatically call activate() in the beginning of the statement and deactivate() and close() on exit.

Parameters

name : str: The desired client name of at most client_name_size() characters. The name scope is local to each server. Unless forbidden by the use_exact_name option, the server will modify this name to create a unique variant, if needed.

Other Parameters

use_exact_name : bool: Whether an error should be raised if name is not unique. See Status.name_not_unique.
no_start_server : bool: Do not automatically start the JACK server when it is not already running. This option is always selected if JACK_NO_START_SERVER is defined in the calling process environment.
servername : str: Selects from among several possible concurrent server instances. Server names are unique to each user. If unspecified, use 'default' unless JACK_DEFAULT_SERVER is defined in the process environment.
session_id : str: Pass a SessionID Token. This allows the sessionmanager to identify the client again.

Raises

JackOpenError: If the session with the JACK server could not be opened.

Expand source code

class TimebaseMasterClient(jack.Client):
    def __init__(self, name, *, bpm=120.0, beats_per_bar=4, beat_type=4,
                 ticks_per_beat=1920, conditional=False, debug=False, **kw):
        super().__init__(name, **kw)
        self.beats_per_bar = int(beats_per_bar)
        self.beat_type = int(beat_type)
        self.bpm = bpm
        self.conditional = conditional
        self.debug = debug
        self.ticks_per_beat = int(ticks_per_beat)
        self.stop_event = Event()
        self.set_shutdown_callback(self.shutdown)

    def shutdown(self, status, reason):
        print('JACK shutdown:', reason, status)
        self.stop_event.set()

    def _tb_callback(self, state, nframes, pos, new_pos):
        if self.debug and new_pos:
            print("New pos:", jack.position2dict(pos))

        # Adapted from:
        # https://github.com/jackaudio/jack2/blob/develop/example-clients/transport.c#L66
        if new_pos:
            pos.beats_per_bar = float(self.beats_per_bar)
            pos.beats_per_minute = self.bpm
            pos.beat_type = float(self.beat_type)
            pos.ticks_per_beat = float(self.ticks_per_beat)
            pos.valid |= jack._lib.JackPositionBBT

            minutes = pos.frame / (pos.frame_rate * 60.0)
            abs_tick = minutes * self.bpm * self.ticks_per_beat
            abs_beat = abs_tick / self.ticks_per_beat

            pos.bar = int(abs_beat / self.beats_per_bar)
            pos.beat = int(abs_beat - (pos.bar * self.beats_per_bar) + 1)
            pos.tick = int(abs_tick - (abs_beat * self.ticks_per_beat))
            pos.bar_start_tick = pos.bar * self.beats_per_bar * self.ticks_per_beat
            pos.bar += 1  # adjust start to bar 1
        else:
            # Compute BBT info based on previous period.
            pos.tick += int(nframes * pos.ticks_per_beat *
                            pos.beats_per_minute / (pos.frame_rate * 60))

            while pos.tick >= pos.ticks_per_beat:
                pos.tick -= int(pos.ticks_per_beat)
                pos.beat += 1

                if pos.beat > pos.beats_per_bar:
                    pos.beat = 1
                    pos.bar += 1
                    pos.bar_start_tick += pos.beats_per_bar * pos.ticks_per_beat

                    if self.debug:
                        print("Pos:", jack.position2dict(pos))

    def become_timebase_master(self, conditional=None):
        return self.set_timebase_callback(self._tb_callback, conditional
                                          if conditional is not None
                                          else self.conditional)

Ancestors

jack.Client

Methods

def become_timebase_master(self, conditional=None)

Expand source code

def become_timebase_master(self, conditional=None):
    return self.set_timebase_callback(self._tb_callback, conditional
                                      if conditional is not None
                                      else self.conditional)

def shutdown(self, status, reason)

Expand source code

def shutdown(self, status, reason):
    print('JACK shutdown:', reason, status)
    self.stop_event.set()