550 lines
25 KiB
Python
Executable File
550 lines
25 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Measure Mini 2 room response, fit bounded EQ and operate the station over SSH."""
|
||
import argparse
|
||
import csv
|
||
import json
|
||
from pathlib import Path
|
||
import secrets
|
||
import select
|
||
import shlex
|
||
import subprocess
|
||
import time
|
||
import wave
|
||
|
||
import numpy as np
|
||
|
||
|
||
RATE = 48000
|
||
PEAK = 10 ** (-16 / 20)
|
||
START = 0.90
|
||
SWEEP = 4.5
|
||
TAIL = 0.60
|
||
MARK1 = 0.50
|
||
MARK2 = START + SWEEP + TAIL
|
||
GRID = 100 * 2 ** (np.arange(0, 73) / 12)
|
||
MAX_RECORDING_SECONDS = 35
|
||
|
||
|
||
def chirp(duration, start, end, logarithmic=False):
|
||
t = np.arange(round(duration * RATE)) / RATE
|
||
if logarithmic:
|
||
k = np.log(end / start) / duration
|
||
cycles = start * np.expm1(k * t) / k
|
||
else:
|
||
cycles = start * t + (end - start) / (2 * duration) * t**2
|
||
fade = round(0.025 * RATE)
|
||
envelope = np.ones(len(t))
|
||
envelope[:fade] = np.sin(np.linspace(0, np.pi / 2, fade)) ** 2
|
||
envelope[-fade:] = envelope[:fade][::-1]
|
||
return PEAK * np.sin(2 * np.pi * cycles) * envelope
|
||
|
||
|
||
FIRST_MARK = chirp(0.15, 1700, 6400)
|
||
LAST_MARK = chirp(0.15, 6800, 2100)
|
||
EXCITATION = chirp(SWEEP, 95, 11500, logarithmic=True)
|
||
|
||
|
||
def probe():
|
||
samples = np.zeros(round((MARK2 + 0.45) * RATE))
|
||
for offset, signal in [(MARK1, FIRST_MARK), (START, EXCITATION), (MARK2, LAST_MARK)]:
|
||
first = round(offset * RATE)
|
||
samples[first:first + len(signal)] = signal
|
||
return np.round(samples * 32767).astype('<i2')
|
||
|
||
|
||
def read_recording(path, internal):
|
||
if internal:
|
||
samples = np.fromfile(path, dtype='<i2')
|
||
if len(samples) != 9 * RATE * 6:
|
||
raise ValueError(f'{path}: ожидается 9 секунд шестиканального PCM')
|
||
return samples.reshape(-1, 6)[:, :4].astype(float) / 32768
|
||
with wave.open(str(path)) as reader:
|
||
rate = reader.getframerate()
|
||
width = reader.getsampwidth()
|
||
channels = reader.getnchannels()
|
||
n = reader.getnframes()
|
||
if width not in (2, 3, 4) or not 1 <= channels <= 8 or not 16000 <= rate <= 96000:
|
||
raise ValueError(f'{path}: нужен WAV PCM 16/24/32 бит, 16–96 кГц')
|
||
if not 6.4 * rate <= n <= MAX_RECORDING_SECONDS * rate:
|
||
raise ValueError(f'{path}: ожидается запись 7–35 секунд')
|
||
raw = reader.readframes(n)
|
||
if width == 3:
|
||
b = np.frombuffer(raw, dtype=np.uint8).reshape(-1, 3).astype(np.int32)
|
||
values = (b[:, 0] | (b[:, 1] << 8) | (b[:, 2] << 16))
|
||
values = ((values ^ 0x800000) - 0x800000) / 8388608
|
||
else:
|
||
values = np.frombuffer(raw, dtype='<i2' if width == 2 else '<i4') / (2 ** (width * 8 - 1))
|
||
samples = values.reshape(-1, channels)
|
||
if rate != RATE:
|
||
times = np.arange(round(len(samples) * RATE / rate)) * rate / RATE
|
||
samples = np.column_stack([np.interp(times, np.arange(len(samples)), channel) for channel in samples.T])
|
||
return samples
|
||
|
||
|
||
def marker_scores(samples, marker, first, last):
|
||
first = max(0, first)
|
||
last = min(last, len(samples) - len(marker))
|
||
if last <= first:
|
||
raise ValueError('не хватает начала или конца записи')
|
||
segment = samples[first:last + len(marker)]
|
||
length = 1 << (len(segment) + len(marker) - 2).bit_length()
|
||
correlation = np.fft.irfft(
|
||
np.fft.rfft(segment, length) * np.conj(np.fft.rfft(marker, length)), length
|
||
)[:last - first + 1]
|
||
squared = np.r_[0, np.cumsum(segment**2)]
|
||
rms = np.sqrt(np.maximum(0, squared[len(marker):] - squared[:-len(marker)]))
|
||
score = np.abs(correlation) / np.maximum(1e-12, rms * np.linalg.norm(marker))
|
||
# Reject false maxima that occur in almost silent segments.
|
||
score[rms < max(1e-5, rms.max() * 0.08)] = 0
|
||
return first, score
|
||
|
||
|
||
def synchronize(samples, span):
|
||
tolerance = round(0.001 * span)
|
||
first, starts = marker_scores(samples, FIRST_MARK, round(0.15 * RATE), len(samples) - span)
|
||
last, ends = marker_scores(samples, LAST_MARK, first + span - tolerance, len(samples))
|
||
best = None
|
||
for _ in range(24):
|
||
peak = int(np.argmax(starts))
|
||
first_score = float(starts[peak])
|
||
if first_score < 0.16:
|
||
break
|
||
start = first + peak
|
||
lo = start + span - tolerance - last
|
||
hi = min(len(ends), lo + 2 * tolerance + 1)
|
||
if hi > lo:
|
||
end = lo + int(np.argmax(ends[lo:hi]))
|
||
last_score = float(ends[end])
|
||
score = first_score * last_score
|
||
if last_score >= 0.16 and (best is None or score > best[0]):
|
||
best = score, start, last + end, min(first_score, last_score)
|
||
exclusion = RATE // 200
|
||
starts[max(0, peak - exclusion):peak + exclusion + 1] = 0
|
||
if best is None:
|
||
raise ValueError('не найдены две согласованные синхрометки')
|
||
_, start, end, confidence = best
|
||
return start, (end - start) / span, confidence
|
||
|
||
|
||
def sweep_snr(received, noise):
|
||
n = 8192
|
||
window = np.hanning(n)
|
||
frequencies = np.fft.rfftfreq(n, 1 / RATE)
|
||
floor = np.abs(np.fft.rfft(noise[-n:] * window)) ** 2
|
||
result = np.full(len(GRID), -100.0)
|
||
for index, frequency in enumerate(GRID):
|
||
center = round(SWEEP * RATE * np.log(frequency / 95) / np.log(11500 / 95))
|
||
if center < n // 2:
|
||
continue
|
||
section = received[center - n // 2:center + n // 2]
|
||
power = np.abs(np.fft.rfft(section * window)) ** 2
|
||
band = np.abs(frequencies - frequency) < max(12, 0.10 * frequency)
|
||
ratio = max(1e-20, power[band].sum()) / max(1e-20, floor[band].sum())
|
||
result[index] = 10 * np.log10(ratio)
|
||
return result
|
||
|
||
|
||
def response(samples):
|
||
if np.max(np.abs(samples)) >= 0.985:
|
||
raise ValueError('микрофон перегружен, снизь уровень записи')
|
||
span = round((MARK2 - MARK1) * RATE)
|
||
start, drift, confidence = synchronize(samples, span)
|
||
offset = (START - MARK1) * RATE
|
||
n = len(EXCITATION) + round(0.44 * RATE)
|
||
times = start + drift * (offset + np.arange(n))
|
||
received = np.interp(times, np.arange(len(samples)), samples)
|
||
noise = samples[max(0, start - RATE // 4):max(0, start - RATE // 25)]
|
||
if len(noise) < 8192:
|
||
raise ValueError('недостаточно тишины перед сигналом')
|
||
snr = 20 * np.log10(max(1e-12, np.std(received)) / max(1e-9, np.std(noise)))
|
||
band_snr = sweep_snr(received, noise)
|
||
mid_snr = np.median(band_snr[(GRID >= 125) & (GRID <= 920)])
|
||
if snr < 6 or mid_snr < 15:
|
||
raise ValueError(f'слишком много шума: {snr:.0f} дБ в записи, {mid_snr:.0f} дБ в полосе коррекции')
|
||
fft_size = 1 << (n - 1).bit_length()
|
||
excitation = np.fft.rfft(EXCITATION, fft_size)
|
||
power = np.abs(excitation) ** 2
|
||
transfer = np.fft.rfft(received, fft_size) * np.conj(excitation) / (power + power.max() * 1e-8)
|
||
ir = np.fft.irfft(transfer)
|
||
# Keep nearby reflections; taper the end so they don't create a sharp window edge.
|
||
window = np.roll(ir, round(0.008 * RATE))[:round(0.388 * RATE)].copy()
|
||
fade = RATE // 10
|
||
window[-fade:] *= np.cos(np.linspace(0, np.pi / 2, fade)) ** 2
|
||
spectrum = np.abs(np.fft.rfft(window, fft_size))
|
||
frequencies = np.fft.rfftfreq(fft_size, 1 / RATE)
|
||
# Fractional-octave power averaging keeps narrow nulls out of the fit.
|
||
db = np.array([
|
||
10 * np.log10(max(1e-20, np.mean(spectrum[(frequencies >= f / 2**(1/12)) &
|
||
(frequencies <= f * 2**(1/12))] ** 2)))
|
||
for f in GRID
|
||
])
|
||
reference = np.median(db[(GRID >= 750) & (GRID <= 1900)])
|
||
return db - reference, {
|
||
'snrDb': round(float(snr), 1),
|
||
'correctionBandSnrDb': round(float(mid_snr), 1),
|
||
'driftPpm': round((drift - 1) * 1e6),
|
||
'sync': round(confidence, 2),
|
||
'_bandSnr': band_snr,
|
||
}
|
||
|
||
|
||
def measurement(path, internal, keep_channels=False):
|
||
channels = read_recording(path, internal)
|
||
traces = []
|
||
quality = []
|
||
failures = []
|
||
for channel in channels.T:
|
||
try:
|
||
trace, q = response(channel)
|
||
traces.append(trace)
|
||
quality.append(q)
|
||
except ValueError as exc:
|
||
failures.append(str(exc))
|
||
if not traces or (internal and len(traces) < 2):
|
||
raise ValueError(f'{path}: ' + '; '.join(dict.fromkeys(failures)))
|
||
# Each external file is one position, even when the recorder supplies stereo.
|
||
return (np.array(traces) if keep_channels else np.median(traces, axis=0)), quality
|
||
|
||
|
||
def filter_curve(frequency, q, gain):
|
||
omega = 2 * np.pi * frequency / RATE
|
||
a = 10 ** (gain / 40)
|
||
alpha = np.sin(omega) / (2 * q)
|
||
z = np.exp(-2j * np.pi * GRID / RATE)
|
||
b = (1 + alpha * a) - 2 * np.cos(omega) * z + (1 - alpha * a) * z**2
|
||
d = (1 + alpha / a) - 2 * np.cos(omega) * z + (1 - alpha / a) * z**2
|
||
return 20 * np.log10(np.abs(b / d))
|
||
|
||
|
||
def fit(traces, internal, reference=None, reliable=None):
|
||
measured = np.median(traces, axis=0)
|
||
spread = np.percentile(traces, 75, axis=0) - np.percentile(traces, 25, axis=0)
|
||
target = -0.7 * np.log2(GRID / 1000)
|
||
if reference is not None:
|
||
target = reference
|
||
mask = (GRID >= (150 if internal else 125)) & (GRID <= (500 if internal else 920))
|
||
if reliable is not None:
|
||
mask &= reliable
|
||
# Only cut peaks observed in most positions. Keep a margin for mic coloration.
|
||
common = np.percentile(traces, 25, axis=0) - target
|
||
margin = 3.0 if internal or len(traces) == 1 else 1.6
|
||
excess = np.where(mask, np.maximum(0, common - margin - 0.35 * spread), 0)
|
||
curve = np.zeros_like(GRID)
|
||
filters = []
|
||
q_values = [0.7, 1.0, 1.2] if internal else [0.7, 1.0, 1.4, 2.0, 2.8]
|
||
max_gain = 2 if internal else 3 if len(traces) == 1 else 5
|
||
total_cut = 3 if internal else 6
|
||
for _ in range(2 if internal else 5):
|
||
residual = excess + curve
|
||
best = 0.0
|
||
choice = None
|
||
base_loss = np.sum((curve + excess) ** 2 * np.where(mask, 1, 1.7))
|
||
for frequency in GRID[mask & (residual > 0.9)]:
|
||
for q in q_values:
|
||
for gain in np.arange(1, max_gain + 0.1, 0.5):
|
||
correction = filter_curve(round(frequency), q, -gain)
|
||
proposed = curve + correction
|
||
if -proposed.min() > total_cut + 0.005:
|
||
continue
|
||
# Penalize reductions where one or more seats already have a dip.
|
||
dip = np.maximum(0, -common - 1)
|
||
loss = np.sum((proposed + excess)**2 * np.where(mask, 1, 1.7) +
|
||
np.abs(proposed) * np.minimum(dip, 8) * 0.3)
|
||
improvement = base_loss - loss
|
||
if improvement > best:
|
||
best, choice = improvement, (round(frequency), q, -gain, proposed)
|
||
if choice is None or best < 0.8:
|
||
break
|
||
frequency, q, gain, curve = choice
|
||
filters.append({'hz': frequency, 'q': q, 'db': float(gain)})
|
||
preamp = -0.6 if filters else 0
|
||
# Pink-weighted estimate, for an initial level match before listening.
|
||
match_band = (GRID >= 160) & (GRID <= 4500)
|
||
dry = preamp + 10 * np.log10(np.mean(10 ** (curve[match_band] / 10)))
|
||
return measured, spread, target, curve, filters, preamp, round(float(dry), 2)
|
||
|
||
|
||
def write_analysis(args):
|
||
measured = [measurement(Path(path), args.internal, keep_channels=args.internal) for path in args.files]
|
||
traces = np.array([np.median(trace, axis=0) if args.internal else trace for trace, _ in measured])
|
||
fit_traces = np.concatenate([trace for trace, _ in measured]) if args.internal else traces
|
||
reference = None
|
||
if args.reference:
|
||
reference = np.median([measurement(Path(path), True)[0] for path in args.reference], axis=0)
|
||
band_snr = [[channel.pop('_bandSnr') for channel in quality] for _, quality in measured]
|
||
if args.internal:
|
||
reliable = np.percentile(np.concatenate(band_snr), 25, axis=0) >= 12
|
||
else:
|
||
reliable = np.min([np.median(channels, axis=0) for channels in band_snr], axis=0) >= 12
|
||
before, spread, target, curve, filters, preamp, dry = fit(fit_traces, args.internal, reference, reliable)
|
||
report = {
|
||
'level': 'built-in' if args.internal else 'external',
|
||
'positions': len(traces),
|
||
'placementReference': bool(args.reference),
|
||
'quality': [quality for _, quality in measured],
|
||
'filters': filters,
|
||
'preampDb': preamp,
|
||
'referenceGainDb': dry,
|
||
}
|
||
prefix = Path(args.output)
|
||
prefix.parent.mkdir(parents=True, exist_ok=True)
|
||
prefix.with_suffix('.json').write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n')
|
||
after = before + curve
|
||
with prefix.with_suffix('.csv').open('w', newline='') as file:
|
||
writer = csv.writer(file)
|
||
writer.writerow(['Hz', 'before_db', 'predicted_db', 'target_db', 'correction_band', 'interquartile_db',
|
||
*[f'position_{i+1}_db' for i in range(len(traces))]])
|
||
band = (GRID >= (150 if args.internal else 125)) & (GRID <= (500 if args.internal else 920))
|
||
for row in zip(GRID, before, after, target, band, spread, *traces):
|
||
writer.writerow([round(float(x), 3) for x in row])
|
||
print(json.dumps(report, ensure_ascii=False, indent=2))
|
||
print(f'Профиль и кривые: {prefix}.json, {prefix}.csv')
|
||
if not filters:
|
||
print('Устойчивых пиков для безопасной коррекции не найдено.')
|
||
|
||
|
||
def verify(args):
|
||
with Path(args.before).open(newline='') as file:
|
||
rows = list(csv.DictReader(file))
|
||
frequencies = np.array([float(row['Hz']) for row in rows])
|
||
if len(frequencies) != len(GRID) or not np.allclose(frequencies, GRID, atol=0.001):
|
||
raise ValueError('CSV должен быть получен командой fit')
|
||
positions = [key for key in rows[0] if key.startswith('position_')]
|
||
if len(positions) != len(args.files):
|
||
raise ValueError('нужен повторный замер в каждой исходной точке и в том же порядке')
|
||
baseline = np.array([[float(row[key]) for row in rows] for key in positions])
|
||
after = np.array([measurement(Path(path), args.internal)[0] for path in args.files])
|
||
target = np.array([float(row['target_db']) for row in rows])
|
||
band = (GRID >= (150 if args.internal else 125)) & (GRID <= (500 if args.internal else 920))
|
||
error = lambda traces: np.sqrt(np.mean((traces[:, band] - target[band]) ** 2, axis=1))
|
||
result = {
|
||
'rmsDeviationBeforeDb': np.round(error(baseline), 2).tolist(),
|
||
'rmsDeviationAfterDb': np.round(error(after), 2).tolist(),
|
||
'meanChangeDb': round(float(np.mean(error(after) - error(baseline))), 2),
|
||
}
|
||
prefix = Path(args.output)
|
||
prefix.parent.mkdir(parents=True, exist_ok=True)
|
||
prefix.with_suffix('.json').write_text(json.dumps(result, indent=2) + '\n')
|
||
with prefix.with_suffix('.csv').open('w', newline='') as file:
|
||
writer = csv.writer(file)
|
||
writer.writerow(['Hz', 'before_db', 'after_db', 'target_db', 'correction_band',
|
||
*[f'position_{i+1}_db' for i in range(len(baseline))],
|
||
*[f'after_position_{i+1}_db' for i in range(len(after))]])
|
||
for row in zip(GRID, np.median(baseline, axis=0), np.median(after, axis=0), target,
|
||
band, *baseline, *after):
|
||
writer.writerow([round(float(x), 3) for x in row])
|
||
print(json.dumps(result, indent=2))
|
||
|
||
|
||
def plot(args):
|
||
import matplotlib.pyplot as plt
|
||
|
||
with Path(args.csv).open(newline='') as file:
|
||
rows = list(csv.DictReader(file))
|
||
frequencies = np.array([float(row['Hz']) for row in rows])
|
||
fig, axis = plt.subplots(figsize=(9, 4.6), layout='constrained')
|
||
for prefix, color in [('position_', '#425979'), ('after_position_', '#1b8c78')]:
|
||
columns = [key for key in rows[0] if key.startswith(prefix)]
|
||
if len(columns) > 1:
|
||
for key in columns:
|
||
axis.semilogx(frequencies, [float(row[key]) for row in rows], color=color, alpha=0.20)
|
||
for key, label, color, style in (
|
||
('before_db', 'До коррекции', '#425979', '-'),
|
||
('predicted_db', 'Расчёт после коррекции', '#1b8c78', '--'),
|
||
('after_db', 'Повторный замер', '#1b8c78', '-'),
|
||
('target_db', 'Цель в рабочем диапазоне', '#8c7290', ':'),
|
||
):
|
||
if key in rows[0]:
|
||
levels = np.array([float(row[key]) for row in rows])
|
||
if key == 'predicted_db' and np.allclose(levels, [float(row['before_db']) for row in rows]):
|
||
continue
|
||
keep = (np.array([float(row['correction_band']) for row in rows]) > 0
|
||
if key == 'target_db' else np.ones(len(rows), dtype=bool))
|
||
axis.semilogx(frequencies[keep], levels[keep],
|
||
label=label, color=color, linestyle=style, linewidth=2)
|
||
axis.set(xlabel='Частота, Гц', ylabel='Относительный уровень, дБ', xlim=(100, 6400))
|
||
axis.grid(alpha=0.25, which='both')
|
||
axis.legend()
|
||
fig.savefig(args.output, dpi=150)
|
||
|
||
|
||
def record_mac(args):
|
||
output = Path(args.output)
|
||
if output.exists():
|
||
raise ValueError(f'файл уже существует: {output}')
|
||
root = Path(__file__).resolve().parent
|
||
cache = Path.home() / 'Library/Caches/alice-room-eq'
|
||
cache.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||
binary = cache / 'record-mac'
|
||
source, plist = root / 'record-mac.swift', root / 'record-mac.plist'
|
||
if not binary.exists() or binary.stat().st_mtime < max(source.stat().st_mtime, plist.stat().st_mtime):
|
||
compiled = cache / f'record-mac-{secrets.token_hex(8)}'
|
||
try:
|
||
subprocess.run(['swiftc', '-parse-as-library', '-O', '-Xlinker', '-sectcreate',
|
||
'-Xlinker', '__TEXT', '-Xlinker', '__info_plist', '-Xlinker', str(plist),
|
||
str(source), '-o', str(compiled)], check=True)
|
||
compiled.replace(binary)
|
||
finally:
|
||
compiled.unlink(missing_ok=True)
|
||
with subprocess.Popen([str(binary), str(output)], stdout=subprocess.PIPE) as recorder:
|
||
try:
|
||
if not select.select([recorder.stdout], [], [], 30)[0] or recorder.stdout.readline() != b'ready\n':
|
||
raise ValueError('микрофон не готов: проверь разрешение macOS и выбранный вход')
|
||
time.sleep(0.8)
|
||
ssh(args, '/data/opt/alice-backend/room-eq.sh play ' + args.preset)
|
||
if recorder.wait(timeout=17) != 0:
|
||
raise ValueError('запись не завершилась')
|
||
except BaseException:
|
||
if recorder.poll() is None:
|
||
recorder.kill()
|
||
recorder.wait()
|
||
output.unlink(missing_ok=True)
|
||
raise
|
||
|
||
|
||
def ssh(args, command, **kwargs):
|
||
return subprocess.run(
|
||
['ssh', '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=8',
|
||
'-i', str(Path(args.identity).expanduser()), '-p', str(args.port), args.host, command],
|
||
check=True, **kwargs
|
||
)
|
||
|
||
|
||
def install(args):
|
||
root = Path(__file__).resolve().parent
|
||
remote = '/data/opt/alice-backend'
|
||
ssh(args, f'mkdir -p {remote}/room-eq')
|
||
for local, target in [
|
||
(Path(args.build) / 'module-ladspa-sink.so', 'spotify/lib/module-ladspa-sink.so'),
|
||
(Path(args.build) / 'alice_room_eq.so', 'room-eq/alice_room_eq.so'),
|
||
(root.parent / 'station/room-eq.sh', 'room-eq.sh'),
|
||
]:
|
||
with local.open('rb') as source:
|
||
ssh(args, f'umask 077; cat > {remote}/{target}.new && mv {remote}/{target}.new {remote}/{target}',
|
||
stdin=source)
|
||
ssh(args, f'umask 077; cat > {remote}/room-eq/probe.pcm.new && '
|
||
f'mv {remote}/room-eq/probe.pcm.new {remote}/room-eq/probe.pcm && chmod 700 {remote}/room-eq.sh',
|
||
input=probe().tobytes())
|
||
|
||
|
||
def apply(args):
|
||
profile = json.loads(Path(args.profile).read_text())
|
||
try:
|
||
filters = [
|
||
{'hz': float(band['hz']), 'q': float(band['q']), 'db': float(band['db'])}
|
||
for band in profile['filters']
|
||
]
|
||
preamp = float(profile['preampDb'])
|
||
reference = float(profile['referenceGainDb'])
|
||
except (KeyError, TypeError, ValueError) as exc:
|
||
raise ValueError('некорректный формат профиля') from exc
|
||
if not 1 <= len(filters) <= 5:
|
||
raise ValueError('профиль должен содержать от одного до пяти фильтров')
|
||
if not -12 <= reference <= 0 or not -12 <= preamp <= 0:
|
||
raise ValueError('предусиление должно быть в диапазоне от −12 до 0 дБ')
|
||
curve = np.zeros_like(GRID)
|
||
for band in filters:
|
||
if not (100 <= band['hz'] <= 1600 and 0.5 <= band['q'] <= 3 and -6 <= band['db'] <= 0):
|
||
raise ValueError('фильтр выходит за допустимые пределы')
|
||
curve += filter_curve(band['hz'], band['q'], band['db'])
|
||
if curve.min() < -6.01:
|
||
raise ValueError('суммарный срез не должен превышать 6 дБ')
|
||
|
||
def controls(preamp, bands):
|
||
values = [preamp]
|
||
for band in bands:
|
||
values.extend([band['hz'], band['q'], band['db']])
|
||
values.extend([300, 1, 0] * (5 - len(bands)))
|
||
return ','.join(f'{float(x):.3f}' for x in values)
|
||
eq = controls(preamp, filters)
|
||
ref = controls(reference, [])
|
||
ssh(args, '/data/opt/alice-backend/room-eq.sh apply', input=f'{args.mode}\n{eq}\n{ref}\n'.encode())
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument('--host', default='root@192.168.1.135')
|
||
parser.add_argument('--port', type=int, default=2222)
|
||
parser.add_argument('--identity', default='~/.ssh/yandex_mini2')
|
||
sub = parser.add_subparsers(dest='command', required=True)
|
||
sub.add_parser('install').add_argument('--build', default='room-eq/build')
|
||
sub.add_parser('probe').add_argument('output')
|
||
capture = sub.add_parser('capture')
|
||
capture.add_argument('output')
|
||
capture.add_argument('--preset', choices=['baseline', 'corrected'], default='baseline')
|
||
recorder = sub.add_parser('record-mac')
|
||
recorder.add_argument('output')
|
||
recorder.add_argument('--preset', choices=['baseline', 'corrected'], default='baseline')
|
||
play = sub.add_parser('play')
|
||
play.add_argument('preset', choices=['baseline', 'corrected'], default='baseline', nargs='?')
|
||
analysis = sub.add_parser('fit')
|
||
analysis.add_argument('files', nargs='+')
|
||
analysis.add_argument('--internal', action='store_true')
|
||
analysis.add_argument('--reference', nargs='+', help='эталонные сырые записи со встроенных микрофонов')
|
||
analysis.add_argument('--output', default='room-profile')
|
||
check = sub.add_parser('verify')
|
||
check.add_argument('before')
|
||
check.add_argument('files', nargs='+')
|
||
check.add_argument('--internal', action='store_true')
|
||
check.add_argument('--output', default='room-verification')
|
||
chart = sub.add_parser('plot')
|
||
chart.add_argument('csv')
|
||
chart.add_argument('output')
|
||
application = sub.add_parser('apply')
|
||
application.add_argument('profile')
|
||
application.add_argument('--mode', choices=['eq', 'ref', 'off'], default='off')
|
||
for name in ('eq', 'ref', 'off', 'disable', 'status', 'ab'):
|
||
sub.add_parser(name)
|
||
args = parser.parse_args()
|
||
try:
|
||
if args.command == 'fit':
|
||
if args.reference and not args.internal:
|
||
raise ValueError('--reference поддерживается только для встроенных микрофонов')
|
||
write_analysis(args)
|
||
elif args.command == 'verify':
|
||
verify(args)
|
||
elif args.command == 'plot':
|
||
plot(args)
|
||
elif args.command == 'record-mac':
|
||
record_mac(args)
|
||
elif args.command == 'probe':
|
||
with wave.open(args.output, 'wb') as file:
|
||
file.setparams((1, 2, RATE, 0, 'NONE', 'not compressed'))
|
||
file.writeframes(probe().tobytes())
|
||
elif args.command == 'install':
|
||
install(args)
|
||
elif args.command == 'apply':
|
||
apply(args)
|
||
elif args.command == 'ab':
|
||
order = ['eq', 'ref'] if secrets.randbits(1) else ['ref', 'eq']
|
||
try:
|
||
while True:
|
||
try:
|
||
choice = input('a/b — переключить, ? — раскрыть, q — выйти: ').strip().lower()
|
||
except EOFError:
|
||
break
|
||
if choice == 'q':
|
||
break
|
||
if choice == '?':
|
||
print(f'A = {order[0]}, B = {order[1]}')
|
||
elif choice in ('a', 'b'):
|
||
ssh(args, '/data/opt/alice-backend/room-eq.sh ' + order[choice == 'b'])
|
||
finally:
|
||
ssh(args, '/data/opt/alice-backend/room-eq.sh off')
|
||
elif args.command == 'capture':
|
||
with Path(args.output).open('xb') as file:
|
||
try:
|
||
ssh(args, '/data/opt/alice-backend/room-eq.sh measure ' + args.preset, stdout=file)
|
||
except BaseException:
|
||
Path(args.output).unlink()
|
||
raise
|
||
else:
|
||
command = args.command + (' ' + shlex.quote(args.preset) if args.command == 'play' else '')
|
||
ssh(args, '/data/opt/alice-backend/room-eq.sh ' + command)
|
||
except (ValueError, OSError, ImportError, wave.Error, subprocess.SubprocessError) as exc:
|
||
parser.exit(1, f'{exc}\n')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|