-
Notifications
You must be signed in to change notification settings - Fork 2
/
cli.py
189 lines (153 loc) · 4.73 KB
/
cli.py
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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
import logging
from typing import List
import click
import cloup
from cloup import Context
from saraswati import __appname__, __version__
from saraswati.model import Channel, SaraswatiSettings
from saraswati.recorder import SaraswatiRecorder
from saraswati.util import setup_logging
appname = f"{__appname__} {__version__}"
HELP_EPILOGUE = """
Examples:
# Record a single channel from the built-in microphone.
saraswati record --channel="testdrive source=autoaudiosrc"
# Record two channels with different sine waves.
saraswati record \\
--channel="channel1 source=audiotestsrc wave=3 freq=200; foo=bar" \\
--channel="channel2 source=audiotestsrc wave=3 freq=400"
# Record to a specified location, using a different chunk duration.
saraswati record \\
--channel="testdrive source=autoaudiosrc" \\
--chunk-duration=10 \\
--spool=/var/spool/saraswati
"""
def validate_channel(ctx, param, value):
"""
Parse multiple "--channel" command line options into list of Channel objects.
"""
channels: List[Channel] = []
for channel_raw in value:
try:
name, options_raw = channel_raw.split(" ", 1)
options_list = options_raw.split(";")
options_list = list(map(str.strip, options_list))
options = {}
for option_item in options_list:
key, value = option_item.split("=", 1)
options[key] = value.strip()
source = options["source"]
del options["source"]
channel = Channel(name=name, source=source, options=options)
channels.append(channel)
except Exception as ex:
raise click.BadParameter(
f"\n--channel option needs to be in format "
f"'channel1 source=audiotestsrc wave=3 freq=200; foo=bar'.\n\n"
f"Exception: '{ex.__class__.__name__}: {ex}'\n"
f"Offending value: '{channel_raw}'.\n"
)
return channels
channels_opt = click.option(
"--channel",
"-c",
"channels",
type=click.STRING,
multiple=True,
callback=validate_channel,
help="Define channels to record. Add multiple times to define more channels",
)
spool_opt = click.option(
"--spool",
"spool_path",
type=click.Path(exists=True, file_okay=False, dir_okay=True, writable=True),
help="Absolute path to the spool directory",
)
debug_opt = click.option("--debug", is_flag=True, help="Turn on debug mode")
def print_version(ctx, param, value):
"""
Implement "--version" flag globally.
https://github.com/pallets/click/issues/1180#issuecomment-444815991
"""
if not value or ctx.resilient_parsing:
return
print(appname)
ctx.exit(0)
def print_help(ctx, param, value):
"""
Augment "--help" flag.
"""
if not value or ctx.resilient_parsing:
return
formatter = cloup.formatting.HelpFormatter(max_width=120)
cli.format_help(ctx, formatter)
print(formatter.getvalue())
print(HELP_EPILOGUE)
ctx.exit(0)
CONTEXT_SETTINGS = Context.settings(
max_content_width=180,
)
@cloup.group(
"saraswati", context_settings=CONTEXT_SETTINGS, invoke_without_command=True
)
@click.option(
"--version",
help="Print the package version and exit.",
is_flag=True,
expose_value=False,
is_eager=True,
callback=print_version,
)
@click.option(
"--help",
help="Print the package version and exit.",
is_flag=True,
expose_value=False,
is_eager=True,
callback=print_help,
)
@click.pass_context
def cli(ctx):
if ctx.invoked_subcommand is None:
print_help(ctx, None, "--help")
@cli.command("record")
@channels_opt
@spool_opt
@click.option(
"--chunk-duration",
type=click.INT,
help="Duration of each chunk file (seconds)",
default=60,
)
@click.option(
"--chunk-max-files",
type=click.INT,
help="Maximum number of file fragments",
default=9999,
)
@debug_opt
def record(
channels: List[Channel],
chunk_duration: int,
chunk_max_files: int,
spool_path: str,
debug: bool,
):
# Setup logging
setup_logging(level=logging.DEBUG)
# Create settings container.
settings = SaraswatiSettings(
channels=None,
chunk_duration=chunk_duration,
chunk_max_files=chunk_max_files,
spool_path=spool_path,
)
# Ensure spool directory exists.
settings.spool_path.mkdir(exist_ok=True)
# Create recorder.
recorder = SaraswatiRecorder(settings=settings)
# autoaudiosrc, osxaudiosrc, audiotestsrc wave=3 freq=200, alsasrc device="hw:1"
for channel in channels:
recorder.add_channel(name=channel.name, source=channel.source)
# Invoke recording pipelines.
recorder.run()