File: timebase_master.py

package info (click to toggle)
python-jack-client 0.5.3-1
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, bullseye, forky, sid, trixie
  • size: 356 kB
  • sloc: python: 2,123; makefile: 6
file content (141 lines) | stat: -rwxr-xr-x 4,872 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
#!/usr/bin/env python3
"""A simple JACK timebase master."""
import argparse
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 = beats_per_bar
        self.beat_type = beat_type
        self.bpm = bpm
        self.conditional = conditional
        self.debug = debug
        self.ticks_per_beat = ticks_per_beat
        self.stop_event = Event()
        self.set_shutdown_callback(self.shutdown_callback)

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

    def timebase_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 = 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.POSITION_BBT

            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:
            assert pos.valid & jack.POSITION_BBT
            # 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):
        if conditional is None:
            conditiona = self.conditional
        return self.set_timebase_callback(self.timebase_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', '--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 (default: %(default)s)')

    args = ap.parse_args(args)

    try:
        beats_per_bar, beat_type = map(float, args.meter.split('/'))
    except ValueError:
        ap.error('Invalid meter: ' + args.meter)

    try:
        tbmaster = TimebaseMasterClient(
            args.client_name,
            bpm=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:
        ap.exit('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:
            ap.exit('Timebase master already present. Exiting...')


if __name__ == '__main__':
    main()