Query and manipulate JACK transport state and provide timebase information using jackclient-python

timebase.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
#  timebase.py
#
"""A simple JACK timebase master."""

import argparse
import sys
import time

# https://github.com/spatialaudio/jackclient-python/
import jack


class TimebaseMaster:
    def __init__(self, client=None, bpm=120.0, meter='4/4', ticks_per_beat=960,
                 conditional=False, debug=False):
        self.bpm = bpm
        self.ticks_per_beat = int(ticks_per_beat)
        self.debug = debug

        try:
            self.beats_per_bar, self.beat_type = (int(x) for x in meter.split('/', 1))
        except (TypeError, ValueError):
            raise ValueError("Invalid meter: %s" % meter)


        if client:
            self.activate(client, conditional)

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

        if new_pos:
            pos.beats_per_bar = self.beats_per_bar
            pos.beats_per_minute = self.bpm
            pos.beat_type = self.beat_type
            pos.ticks_per_beat = self.ticks_per_beat
            pos.valid |= jack._lib.JackPositionBBT

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

            pos.bar = int(abs_beat / pos.beats_per_bar)
            pos.beat = int(abs_beat - (pos.bar * pos.beats_per_bar) + 1)
            pos.tick = int(abs_tick - (abs_beat * pos.ticks_per_beat))
            pos.bar_start_tick = pos.bar * pos.beats_per_bar * pos.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 activate(self, client, conditional=False):
        client.set_timebase_callback(self.callback, 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', '--client-name',
        default="timebase",
        help="JACK client name (default: %(default)s9")
    ap.add_argument(
        '-m', '--meter',
        default='4/4',
        help="Meter as <beats-per-bar>/<denominator> (default: %(default)s)")
    ap.add_argument(
        'tempo',
        type=float,
        default=120.0,
        help="Tempo in beats per minute (default: %(default)s)")

    args = ap.parse_args(args if args is not None else sys.argsv[1:])

    try:
        client = jack.Client(args.client_name)
        tm = TimebaseMaster(
            client,
            bpm=max(0.0, min(240.0, args.tempo)),
            meter=args.meter,
            debug=args.debug)
    except jack.JackError as exc:
        return "Could not create JACK client:: %s" % exc
    except ValueError as exc:
        return "Could not create timebase master: %s" % exc

    client.activate()

    try:
        while True:
            time.sleep(0.1)
    except KeyboardInterrupt:
        print("\nBye!")
    finally:
        try:
            client.release_timebase()
        except (AttributeError, JackError):
            pass
        client.deactivate()
        client.close()


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]) or 0)

transporter.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
#  transporter.py
#
"""Query or change the JACK transport state."""

import argparse
import string
import sys

# https://github.com/spatialaudio/jackclient-python/
import jack


STATE_LABELS = {
    jack.ROLLING: "rolling",
    jack.STOPPED: "stopped",
    jack.STARTING: "starting",
    jack.NETSTARTING: "waiting for network sync",
}


def main(args=None):
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument(
        '-c', '--client-name',
        default="transporter",
        help="JACK client name (default: %(default)s)")
    ap.add_argument(
        'command',
        nargs='?',
        default='status',
        choices=['query', 'rewind', 'start', 'status', 'stop', 'toggle'],
        help="Transport command")

    args = ap.parse_args(args if args is not None else sys.argv[1:])

    try:
        client = jack.Client(args.client_name)
    except jack.JackError as exc:
        return "Could not create JACK client:: %s" % exc

    state = client.transport_state
    result = 0

    if args.command == 'status':
        print("JACK transport is %s." % STATE_LABELS[state._code])
        result = 1 if state == jack.STOPPED else 0
    elif args.command == 'query':
        print("State: %s" % STATE_LABELS[state._code])
        info = client.transport_query()[1]
        for field in sorted(info):
            label = string.capwords(field.replace('_', ' '))
            print("%s: %s" % (label, info[field]))
        result = 1 if state == jack.STOPPED else 0
    elif args.command == 'start':
        if state == jack.STOPPED:
            client.transport_start()
    elif args.command == 'stop':
        if state != jack.STOPPED:
            client.transport_stop()
    elif args.command == 'toggle':
        if state == jack.STOPPED:
            client.transport_start()
        else:
            client.transport_stop()
    elif args.command == 'rewind':
        client.transport_frame = 0

    client.close()
    return result


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]) or 0)