unit TCIStreams; { TCIStreams.pas — бинарные потоки TCI (§3.4): нарезка сэмплов на блоки, пересчёт частоты дискретизации, запись линейного выхода в файл. Чистый слой обработки: ни контроллера, ни движка тут нет — только TCIProtocol (формат блока) и TCIServer (очередь клиента). Кто и откуда кормит эти объекты, решает TCIAdapter. ★Главное про потоки исполнения. Feed* зовёт DSP-ПОТОК (тап аудио/IQ), то есть тот самый, который считает WDSP. Поэтому здесь нет ни одного вызова, который может ждать: блок уходит в кольцо клиента (микросекунды под его локом), а в сокет его пишет собственный поток клиента. Отставший клиент теряет свои блоки (TTCIClient.SendBin выбрасывает самый старый) и никого больше не задерживает. Пересчёт частоты: • вниз (RX-аудио 48 кГц → 8/12/24, IQ 384 → 48/96/192) — FIR-дециматор с целым коэффициентом. Без фильтра тут нельзя: широкий ФМ-канал или шум за полосой сложились бы в звуковую полосу зеркалом; • вверх (TX-аудио клиента 8/12/24 кГц → 48 кГц тракта) — линейная интерполяция. Образы от неё лежат на 8..12 кГц и выше, то есть заведомо за полосой TX-фильтра (максимум 4 кГц), а завал в полосе — 0.2 дБ на 3 кГц. Городить ради этого второй FIR смысла нет. } {$IFDEF FPC} {$MODE Delphi} {$LONGSTRINGS ON} {$ENDIF} interface uses // ★Windows идёт ПЕРВЫМ намеренно: он объявляет свой TCriticalSection (запись, // а не класс), и стоя последним перекрыл бы SyncObjs — на win64 это уже // ловилось в DX-кластере. Порядок здесь и есть лечение. {$IFDEF WINDOWS}Windows,{$ENDIF} {$IFDEF UNIX}BaseUnix,{$ENDIF} Classes, SysUtils, Math, SyncObjs, TCIProtocol, TCIServer; const // Длина FIR на каждую ступень прореживания. Ntaps = TCI_FIR_PER_FACTOR×M+1 // ⇒ стоимость на ВХОДНОЙ сэмпл постоянна (≈8 умножений) независимо от M. TCI_FIR_PER_FACTOR = 8; TCI_FIR_MAX_TAPS = 257; // Кусок, которым поток перемалывает подачу от движка. Все рабочие буферы // заведены под него в конструкторе: SetLength в DSP-потоке на каждый блок // аудио — это тысячи обращений к куче в секунду на ровном месте. TCI_FEED_CHUNK = 4096; type { Накопленная запись: 16-битный PCM с чередованием L/R. } TTCIPcm = array of SmallInt; { Дециматор с целым коэффициентом. Прямая свёртка по линии задержки: выход считается только на нужной фазе, поэтому цена не зависит от M. } TTCIDecimator = class private FTaps: array of Single; FHist: array of Single; // линия задержки, кольцо FN: Integer; // длина FIR FHalf: Integer; // FN div 2 — число симметричных пар FPos: Integer; FPhase: Integer; FFactor: Integer; public { Одиночный дециматор: полоса 0.45 от новой частоты Найквиста, длина FIR по коэффициенту. Для одной ступени этого достаточно. } constructor Create(AFactor: Integer); { Ступень каскада: полосу и длину задаёт вызывающий — ранним ступеням узкая переходная полоса не нужна (см. TTCIDecimChain). } constructor CreateDesigned(AFactor, ATaps: Integer; ACutoff: Double); procedure Reset; { N входных сэмплов → до N/Factor выходных. Dst обязан вмещать столько. } function Process(const Src: array of Single; N: Integer; var Dst: array of Single): Integer; property Factor: Integer read FFactor; property Taps: Integer read FN; end; { Каскад дециматоров. Одной ступенью большие коэффициенты не берутся: длина FIR растёт вместе с коэффициентом, а упираясь в потолок TCI_FIR_MAX_TAPS, одноступенчатый дециматор перестаёт быть фильтром вовсе. Замерено на 5760→48 кГц (коэффициент 120, верхний rate Pluto): завал 1.3 дБ в полосе и подавление зеркала всего 16 дБ — то есть поток IQ с мусором. Поэтому коэффициент раскладывается на множители (по убыванию — самая дорогая ступень первой). Спецификацию фильтра каждой ступени задаёт ИТОГОВАЯ полоса, а не её собственная: ранняя ступень обязана убрать лишь те узкие зоны, которые в конце сложатся в полезную полосу, поэтому её переходная полоса шире в десятки раз, а фильтр во столько же короче. Итог замера: −83 дБ по зеркалу на любом коэффициенте, цена ≈10% ядра на потоке 5.76 МГц (было 20% и мусор). Свёртка идёт со сложением симметричных пар — умножений вдвое меньше при том же результате. } TTCIDecimChain = class private FStages: array of TTCIDecimator; FTmp: array[0..1] of array of Single; FFactor: Integer; public constructor Create(AFactor: Integer); destructor Destroy; override; procedure Reset; function Process(const Src: array of Single; N: Integer; var Dst: array of Single): Integer; property Factor: Integer read FFactor; end; { Интерполятор для TX-аудио: целое отношение, линейная интерполяция. } TTCIInterpolator = class private FFactor: Integer; FPrev: Single; FHas: Boolean; public constructor Create(AFactor: Integer); procedure Reset; function Process(const Src: array of Single; N: Integer; var Dst: array of Double): Integer; property Factor: Integer read FFactor; end; { Исходящий поток одного клиента: один тип, один приёмник. Живёт от START до STOP (или до ухода клиента) и владеет своими дециматорами и накопителем. } TTCIStreamOut = class private FClient: TTCIClient; FKind: TTCIStreamType; FRx: Integer; FSrcRate: Integer; FOutRate: Integer; FChannels: Integer; FSampleT: TTCISampleType; FBlock: Integer; // сэмплов НА КАНАЛ в блоке FDec: array[0..1] of TTCIDecimChain; FTmp: array[0..1] of array of Single; // выход дециматора FIn: array of Single; // вход одного канала, кусок FWantRate: Integer; // о чём просил клиент (для пересборки) FAcc: array of Single; // накопитель с чередованием каналов FAccCount: Integer; // сэмплов на канал в накопителе FPacked: array of Byte; procedure EmitFull; procedure PushPair(A, B: Single); public constructor Create(AClient: TTCIClient; AKind: TTCIStreamType; ARx, ASrcRate, AWantRate, AChannels: Integer; ASampleT: TTCISampleType; ABlock: Integer); destructor Destroy; override; { Аудио 48 кГц: стерео от движка. Один канал — усреднение (моно). } procedure FeedAudio(const L, R: array of Single; N: Integer); { IQ: комплексные отсчёты, всегда два канала. } procedure FeedIQ(PI_, PQ_: PDouble; N: Integer); { Совпадает ли поток с (клиент, тип, приёмник) — для поиска в списке. } function Matches(AClient: TTCIClient; AKind: TTCIStreamType; ARx: Integer): Boolean; { Частота источника сменилась на ходу (другой sample rate устройства или rate DDC пана). Пересобирает прореживание; накопленный блок бросаем — склеивать в один блок сэмплы двух разных частот нельзя. } procedure SetSourceRate(ANewRate: Integer); property Client: TTCIClient read FClient; property Kind: TTCIStreamType read FKind; property Rx: Integer read FRx; property OutRate: Integer read FOutRate; property SrcRate: Integer read FSrcRate; end; { Запись линейного выхода (LINE_OUT_RECORDER_*, §4.3). Буфер на MaxSec секунд 48 кГц стерео в int16: во-первых, ровно то, что уйдёт в WAV, а во-вторых, float32 на предельных 300 с — это 115 МБ вместо 57. ★Не кольцо. Документ говорит про arg2 «максимальное время записи», и дальше прямо: «по истечении времени запись УДАЛЯЕТСЯ, чтобы сохранить запись в файл необходимо в заданном временном интервале выслать LINE_OUT_RECORDER_SAVE». То есть START открывает окно длиной MaxSec, и всё, что не сохранили внутри него, пропадает. Кольцо «последние N секунд» вело себя иначе в обе стороны: начало активной записи затирало само себя, а SAVE через час после START отдавал файл, которого у ExpertSDR3 давно бы не было. Срок считаем по ЧАСАМ, а не по накопленным сэмплам: линейный выход молчит (мьют, пауза приёмника), а время записи всё равно идёт. } { ★Память набирается КУСКАМИ по мере записи, а не вся сразу на START. Раньше конструктор выделял MaxSec × 48 кГц × 2 канала × int16 — до 57.6 МБ на одну строку из сети; семь приёмников давали 403 МБ, и удержать их мог любой клиент (авторизации в TCI нет). Теперь START не стоит ни байта: молчащий, мёртвый или замьюченный приёмник не занимает ничего, а растущая запись спрашивает разрешения на каждый кусок у общего бюджета TCI_RECORD_MAX_BYTES. Куски не перевыделяются (никакого realloc в DSP-потоке) и склеиваются один раз, в Take — а он идёт уже после того, как рекордер вынут из таблицы, то есть без DSP-потока на плечах. } TTCIRecorder = class private FLock: TCriticalSection; FChunks: array of TTCIPcm; // куски по FChunk сэмплов на канал, L/R вперемешку FChunk: Integer; // сэмплов на канал в куске FCap: Integer; // потолок в сэмплах на канал FCount: Integer; // накоплено сэмплов на канал FBytes: Int64; // занято под FChunks (столько же взято у бюджета) FRate: Integer; FRx: Integer; FOwner: TObject; // клиент, попросивший START (для его ухода) FExpired: Boolean; // окно записи закрылось, данные удалены FStarved: Boolean; // бюджет не дал расти — пишем сколько влезло FEndsAt: QWord; // GetTickCount64 конца окна procedure DropData; // под FLock function EnsureRoom: Boolean; // под FLock: место под ещё один сэмпл public constructor Create(ARx, ARateHz, AMaxSec: Integer; AOwner: TObject); destructor Destroy; override; procedure Feed(const L, R: array of Single; N: Integer); { Забрать накопленное и завершить запись. nil — либо не записано ничего, либо окно уже истекло (по документу это одно и то же: записи нет). ★Reserved — сколько байт бюджета УХОДИТ ВМЕСТЕ С ДАННЫМИ: копия живёт дальше в писателе, и пока он её не отпустит, она обязана оставаться учтённой. Вызывающий передаёт это число писателю (или возвращает сам через TCIRecBudgetFree, если писателя не будет). } function Take(out Reserved: Int64): TTCIPcm; { Окно закрылось по часам. Спрашивает тик сервера: у мёртвого приёмника Feed не зовут вовсе, и без этого вопроса память жила бы до Stop. } function Expired: Boolean; property Rx: Integer read FRx; property Rate: Integer read FRate; property Owner: TObject read FOwner; end; { Писатель WAV в своём потоке: файл до 60 МБ, а зовут сохранение из тика сервера — блокировать его на секунду диска нельзя. Данные забирает себе ВМЕСТЕ С ИХ РЕЗЕРВОМ в общем бюджете (AReserved из TTCIRecorder.Take) и отпускает его, только когда данные больше не нужны. Пока писатель ждёт медленный диск, эти байты остаются занятыми — так потолок и держит очередь сохранений, а не только сами записи. } TTCIWavWriter = class(TThread) private FPath: string; FData: TTCIPcm; FRate: Integer; FReserved: Int64; procedure ReleaseData; protected procedure Execute; override; public constructor Create(const APath: string; const AData: TTCIPcm; ARateHz: Integer; AReserved: Int64); end; { Коэффициент прореживания SrcRate → WantRate: наибольший целый делитель, дающий не меньше запрошенного. Апсемплинг наружу не делаем никогда — клиенту уходит настоящая частота (она же в заголовке блока). } function TCIDecimFactor(SrcRate, WantRate: Integer): Integer; { Частота IQ, которую реально можно отдать: наибольшая ЗАКОННАЯ по протоколу (48/96/192/384 кГц), не выше запрошенной и делящая частоту источника нацело. Нужна из-за частот дискретизации Pluto: 576 и 960 кГц на 384 не делятся, и без этого выбора клиент, попросивший 384 кГц, получал бы поток на 576/480 — и не по протоколу, и вчетверо толще, чем он ждёт. Если законной не нашлось вовсе (чужой rate), возвращаем просьбу как есть: дальше её обработает TCIDecimFactor, а настоящая частота уйдёт в заголовке блока. } function TCIPickIQRate(SrcRate, WantRate: Integer): Integer; { ═══ Общий бюджет памяти рекордеров ═══════════════════════════════════════ Взять/вернуть можно из любого потока (Take зовёт DSP-поток, Free — тик, поток клиента и поток контроллера). TCIRecBudgetTake возвращает False, если запрошенное не влезает в TCI_RECORD_MAX_BYTES. } function TCIRecBudgetTake(Bytes: Int64): Boolean; procedure TCIRecBudgetFree(Bytes: Int64); function TCIRecBudgetUsed: Int64; // для стенда и диагностики { ═══ Имя файла записи ═════════════════════════════════════════════════════ ★Из строки LINE_OUT_RECORDER_SAVE берётся ТОЛЬКО имя файла, а каталог — всегда настроенный каталог записей. Причина простая: авторизации в TCI нет (§3.1), а bind наружу разрешён, поэтому любой, кто дотянулся до порта, писал бы файлы от имени EWSDR куда угодно — путь уходил в fmCreate почти как пришёл, вместе с '..' и абсолютными путями. Документ (§4.3) в примерах даёт полный путь, но полный путь на СЕРВЕРЕ клиенту всё равно бесполезен: файл ложится не на его машину, а на нашу. Поэтому каталог из просьбы отбрасываем молча (клиент получает разумный файл, а не ошибку), а имя проверяем: пусто, '.', '..', управляющие символы и слишком длинное — отказ; расширение обязано быть .wav (кодировать MP3 нам нечем). Выйти за каталог после ExtractFileName нечем — разделителей в имени уже не осталось. Result = '' — имя не годится; иначе полный путь внутри BaseDir. } function TCIRecordPath(const BaseDir, Req: string): string; { Создать файл, НЕ перезаписывая существующий и не идя по симлинку. Отдельно от TFileStream: fmCreate молча затирает то, что уже лежит по этому пути, а имя приходит из сети. THandle(-1) — не вышло. } function TCICreateNewFile(const Path: string): THandle; implementation { ═══════════════════════════════════════════════════════════════════════════ Дециматор ═══════════════════════════════════════════════════════════════════════════ } constructor TTCIDecimator.Create(AFactor: Integer); var N: Integer; begin if AFactor < 1 then AFactor := 1; N := TCI_FIR_PER_FACTOR * AFactor + 1; CreateDesigned(AFactor, N, 0.45 / AFactor); end; constructor TTCIDecimator.CreateDesigned(AFactor, ATaps: Integer; ACutoff: Double); var i, C: Integer; X, W, Sum: Double; begin inherited Create; if AFactor < 1 then AFactor := 1; FFactor := AFactor; FN := ATaps; if FN < 9 then FN := 9; if FN > TCI_FIR_MAX_TAPS then FN := TCI_FIR_MAX_TAPS; if (FN and 1) = 0 then Inc(FN); // нечётная длина: линейная фаза и // целая задержка FHalf := FN div 2; SetLength(FTaps, FN); SetLength(FHist, FN); // Окно Блэкмана поверх sinc. Оно, в отличие от Хэмминга, даёт −74 дБ вместо // −53 в полосе задержания, а платим за это только длиной — и как раз длину // многоступенчатая схема экономит (см. TTCIDecimChain). C := FN div 2; Sum := 0; for i := 0 to FN - 1 do begin X := i - C; if Abs(X) < 1E-9 then W := 2 * ACutoff else W := Sin(2 * Pi * ACutoff * X) / (Pi * X); W := W * (0.42 - 0.5 * Cos(2 * Pi * i / (FN - 1)) + 0.08 * Cos(4 * Pi * i / (FN - 1))); FTaps[i] := W; Sum := Sum + W; end; // Нормировка по единичному усилению на постоянном токе: без неё уровень // аудио гулял бы на доли дБ от коэффициента прореживания. if Abs(Sum) > 1E-12 then for i := 0 to FN - 1 do FTaps[i] := FTaps[i] / Sum; Reset; end; procedure TTCIDecimator.Reset; var i: Integer; begin for i := 0 to FN - 1 do FHist[i] := 0; FPos := 0; FPhase := 0; end; function TTCIDecimator.Process(const Src: array of Single; N: Integer; var Dst: array of Single): Integer; var i, k, t, a, b: Integer; Acc: Double; begin Result := 0; if N > Length(Src) then N := Length(Src); if FFactor = 1 then begin // Прореживать нечего — фильтр в этом случае только съел бы верх полосы. k := N; if k > Length(Dst) then k := Length(Dst); for i := 0 to k - 1 do Dst[i] := Src[i]; Exit(k); end; k := 0; for i := 0 to N - 1 do begin FHist[FPos] := Src[i]; Inc(FPos); if FPos >= FN then FPos := 0; Inc(FPhase); if FPhase < FFactor then Continue; FPhase := 0; if k >= Length(Dst) then Break; // переполнение приёмника — молча режем // Свёртка со сложением симметричных пар: фильтр линейнофазовый, значит // FTaps[t] = FTaps[N-1-t], и умножений вдвое меньше при том же результате. // На 5.76 МГц (верхний rate Pluto) это разница между 16% и 10% ядра. Acc := 0; a := FPos; // самый старый отсчёт (это отвод N-1) b := FPos + FN - 1; // самый свежий (это отвод 0) if b >= FN then Dec(b, FN); for t := 0 to FHalf - 1 do begin Acc := Acc + FTaps[t] * (FHist[a] + FHist[b]); Inc(a); if a >= FN then a := 0; Dec(b); if b < 0 then b := FN - 1; end; Dst[k] := Acc + FTaps[FHalf] * FHist[a]; // центральный отвод Inc(k); end; Result := k; end; { ═══════════════════════════════════════════════════════════════════════════ Каскад дециматоров ═══════════════════════════════════════════════════════════════════════════ } constructor TTCIDecimChain.Create(AFactor: Integer); var Rest, F, i, N: Integer; Cur, FPass, FStop: Double; Fac: array of Integer; procedure AddFactor(V: Integer); begin SetLength(Fac, Length(Fac) + 1); Fac[High(Fac)] := V; end; begin inherited Create; if AFactor < 1 then AFactor := 1; FFactor := AFactor; Rest := AFactor; // Раскладываем на простые. Всё, что не разложилось (простое число больше // семи — у частот дискретизации не встречается, но бывает у чужого железа), // остаётся одной ступенью: она хотя бы не хуже прежнего поведения. for F in [2, 3, 5, 7] do while (Rest mod F = 0) and (Rest > 1) do begin AddFactor(F); Rest := Rest div F; end; if Rest > 1 then AddFactor(Rest); // По убыванию: первая ступень самая «дорогая», и после неё частота, на // которой работают остальные, уже сбита. for i := 0 to High(Fac) - 1 do for F := 0 to High(Fac) - 1 - i do if Fac[F] < Fac[F + 1] then begin Rest := Fac[F]; Fac[F] := Fac[F + 1]; Fac[F + 1] := Rest; end; // Спецификация фильтра каждой ступени считается от ИТОГОВОЙ полосы, а не от // её собственной. Ранняя ступень отдаёт наверх широкий поток, и всё, что она // обязана убрать, — те узкие зоны, которые в конце сложатся в полезную // полосу; переходная полоса у неё получается в десятки раз шире, а значит и // фильтр во столько же раз короче. Ради этого многоступенчатую схему и // делают: на 5760→48 кГц первая ступень обходится 30 отводами вместо 257, // которых всё равно не хватало. SetLength(FStages, Length(Fac)); Cur := 1.0; // доля от исходной частоты на входе ступени for i := 0 to High(Fac) do begin // Полоса, которую обязаны сохранить, — 0.45 от итоговой Найквиста, // в единицах ВХОДНОЙ частоты этой ступени. FPass := (0.45 * 0.5 / FFactor) / Cur; FStop := (1.0 / Fac[i]) - FPass; // сюда сложится всё лишнее if FStop <= FPass * 1.05 then // последняя ступень: запаса уже нет begin FPass := 0.45 / Fac[i]; FStop := 0.55 / Fac[i]; end; // Ширина переходной полосы ↔ длина окна Блэкмана: N ≈ 5.5/Δf. N := Ceil(5.5 / (FStop - FPass)) + 1; FStages[i] := TTCIDecimator.CreateDesigned(Fac[i], N, (FPass + FStop) * 0.5); Cur := Cur / Fac[i]; end; SetLength(FTmp[0], TCI_FEED_CHUNK); SetLength(FTmp[1], TCI_FEED_CHUNK); end; destructor TTCIDecimChain.Destroy; var i: Integer; begin for i := 0 to High(FStages) do FStages[i].Free; inherited; end; procedure TTCIDecimChain.Reset; var i: Integer; begin for i := 0 to High(FStages) do FStages[i].Reset; end; function TTCIDecimChain.Process(const Src: array of Single; N: Integer; var Dst: array of Single): Integer; var i, Cur: Integer; begin if N > Length(Src) then N := Length(Src); if Length(FStages) = 0 then begin if N > Length(Dst) then N := Length(Dst); for i := 0 to N - 1 do Dst[i] := Src[i]; Exit(N); end; if Length(FStages) = 1 then Exit(FStages[0].Process(Src, N, Dst)); // Пинг-понг между двумя буферами; последняя ступень пишет сразу в Dst. Cur := 0; N := FStages[0].Process(Src, N, FTmp[0]); for i := 1 to High(FStages) - 1 do begin N := FStages[i].Process(FTmp[Cur], N, FTmp[1 - Cur]); Cur := 1 - Cur; end; Result := FStages[High(FStages)].Process(FTmp[Cur], N, Dst); end; { ═══════════════════════════════════════════════════════════════════════════ Интерполятор (TX-аудио клиента → 48 кГц тракта) ═══════════════════════════════════════════════════════════════════════════ } constructor TTCIInterpolator.Create(AFactor: Integer); begin inherited Create; if AFactor < 1 then AFactor := 1; FFactor := AFactor; Reset; end; procedure TTCIInterpolator.Reset; begin FPrev := 0; FHas := False; end; function TTCIInterpolator.Process(const Src: array of Single; N: Integer; var Dst: array of Double): Integer; var i, p, k: Integer; A, B: Single; begin if N > Length(Src) then N := Length(Src); k := 0; if FFactor = 1 then begin for i := 0 to N - 1 do begin if k >= Length(Dst) then Break; Dst[k] := Src[i]; Inc(k); end; Exit(k); end; for i := 0 to N - 1 do begin B := Src[i]; if FHas then A := FPrev else A := B; // самый первый блок: без скачка от нуля for p := 0 to FFactor - 1 do begin if k >= Length(Dst) then Break; Dst[k] := A + (B - A) * (p / FFactor); Inc(k); end; FPrev := B; FHas := True; end; Result := k; end; { ═══════════════════════════════════════════════════════════════════════════ Исходящий поток ═══════════════════════════════════════════════════════════════════════════ } constructor TTCIStreamOut.Create(AClient: TTCIClient; AKind: TTCIStreamType; ARx, ASrcRate, AWantRate, AChannels: Integer; ASampleT: TTCISampleType; ABlock: Integer); var F, MaxB, i: Integer; begin inherited Create; FClient := AClient; FKind := AKind; FRx := ARx; FSrcRate := ASrcRate; FWantRate := AWantRate; FChannels := EnsureRange(AChannels, 1, 2); FSampleT := ASampleT; if FKind = tstIQ then AWantRate := TCIPickIQRate(ASrcRate, AWantRate); F := TCIDecimFactor(ASrcRate, AWantRate); FOutRate := ASrcRate div F; // Блок не имеет права вылезти за data[16384]: столько ExpertSDR3 объявил // потолком, и клиенты держат приёмный буфер ровно под него. MaxB := TCIMaxBlockSamples(FSampleT, FChannels); FBlock := EnsureRange(ABlock, 1, MaxB); for i := 0 to FChannels - 1 do begin FDec[i] := TTCIDecimChain.Create(F); SetLength(FTmp[i], TCI_FEED_CHUNK); end; SetLength(FIn, TCI_FEED_CHUNK); SetLength(FAcc, FBlock * FChannels); SetLength(FPacked, FBlock * FChannels * TCISampleBytes(FSampleT)); FAccCount := 0; end; destructor TTCIStreamOut.Destroy; var i: Integer; begin for i := 0 to 1 do FreeAndNil(FDec[i]); inherited; end; function TTCIStreamOut.Matches(AClient: TTCIClient; AKind: TTCIStreamType; ARx: Integer): Boolean; begin Result := (FClient = AClient) and (FKind = AKind) and (FRx = ARx); end; procedure TTCIStreamOut.SetSourceRate(ANewRate: Integer); var i, F, W: Integer; begin if (ANewRate <= 0) or (ANewRate = FSrcRate) then Exit; FSrcRate := ANewRate; // Просьбу клиента храним как есть, а законную частоту пересчитываем: у // нового источника делители другие (576 кГц Pluto не делится на 384). if FKind = tstIQ then W := TCIPickIQRate(FSrcRate, FWantRate) else W := FWantRate; F := TCIDecimFactor(FSrcRate, W); FOutRate := FSrcRate div F; for i := 0 to FChannels - 1 do begin FDec[i].Free; FDec[i] := TTCIDecimChain.Create(F); end; FAccCount := 0; end; procedure TTCIStreamOut.EmitFull; var H: TTCIStreamHeader; Bytes: Integer; begin TCIFillHeader(H, FKind, FRx, FOutRate, FSampleT, FBlock, FChannels); Bytes := TCIPackSamples(FAcc, FBlock * FChannels, FSampleT, @FPacked[0]); FClient.SendBin(H, @FPacked[0], Bytes); FAccCount := 0; end; procedure TTCIStreamOut.PushPair(A, B: Single); begin if FChannels = 1 then FAcc[FAccCount] := A else begin FAcc[FAccCount * 2] := A; FAcc[FAccCount * 2 + 1] := B; end; Inc(FAccCount); if FAccCount >= FBlock then EmitFull; end; procedure TTCIStreamOut.FeedAudio(const L, R: array of Single; N: Integer); var i, n0, n1, Chunk, Off: Integer; begin if (N <= 0) or (FClient = nil) then Exit; if N > Length(L) then N := Length(L); if N > Length(R) then N := Length(R); Off := 0; while Off < N do begin // Кусками по TCI_FEED_CHUNK: движок вправе отдать блок любой длины, а // рабочие буферы у нас фиксированные (см. константу). Chunk := N - Off; if Chunk > TCI_FEED_CHUNK then Chunk := TCI_FEED_CHUNK; if FChannels = 1 then begin for i := 0 to Chunk - 1 do FIn[i] := (L[Off + i] + R[Off + i]) * 0.5; n0 := FDec[0].Process(FIn, Chunk, FTmp[0]); for i := 0 to n0 - 1 do PushPair(FTmp[0][i], 0); end else begin for i := 0 to Chunk - 1 do FIn[i] := L[Off + i]; n0 := FDec[0].Process(FIn, Chunk, FTmp[0]); for i := 0 to Chunk - 1 do FIn[i] := R[Off + i]; n1 := FDec[1].Process(FIn, Chunk, FTmp[1]); if n1 < n0 then n0 := n1; for i := 0 to n0 - 1 do PushPair(FTmp[0][i], FTmp[1][i]); end; Inc(Off, Chunk); end; end; procedure TTCIStreamOut.FeedIQ(PI_, PQ_: PDouble; N: Integer); var i, n0, n1, Chunk, Off: Integer; SI, SQ: PDouble; begin if (N <= 0) or (FClient = nil) or (PI_ = nil) or (PQ_ = nil) then Exit; Off := 0; while Off < N do begin Chunk := N - Off; if Chunk > TCI_FEED_CHUNK then Chunk := TCI_FEED_CHUNK; SI := PI_; Inc(SI, Off); for i := 0 to Chunk - 1 do begin FIn[i] := SI^; Inc(SI); end; n0 := FDec[0].Process(FIn, Chunk, FTmp[0]); if FChannels >= 2 then begin // Q считаем ТЕМ ЖЕ проходом, что и I: разошедшиеся по длине выходы // означали бы сдвиг фазы между каналами, то есть поворот спектра. SQ := PQ_; Inc(SQ, Off); for i := 0 to Chunk - 1 do begin FIn[i] := SQ^; Inc(SQ); end; n1 := FDec[1].Process(FIn, Chunk, FTmp[1]); if n1 < n0 then n0 := n1; for i := 0 to n0 - 1 do PushPair(FTmp[0][i], FTmp[1][i]); end else for i := 0 to n0 - 1 do PushPair(FTmp[0][i], 0); Inc(Off, Chunk); end; end; { ═══════════════════════════════════════════════════════════════════════════ Рекордер линейного выхода ═══════════════════════════════════════════════════════════════════════════ } constructor TTCIRecorder.Create(ARx, ARateHz, AMaxSec: Integer; AOwner: TObject); begin inherited Create; FLock := TCriticalSection.Create; FRx := ARx; FRate := ARateHz; FOwner := AOwner; if AMaxSec < 1 then AMaxSec := 1; if AMaxSec > TCI_RECORD_MAX_SEC then AMaxSec := TCI_RECORD_MAX_SEC; FCap := ARateHz * AMaxSec; FChunk := ARateHz * TCI_RECORD_CHUNK_SEC; if FChunk < 1 then FChunk := 1; if FChunk > FCap then FChunk := FCap; // ★Ни одного байта под звук здесь не выделяется — см. шапку объявления. FCount := 0; FBytes := 0; FExpired := False; FStarved := False; // Окно открывается прямо здесь: клиент отсчитывает его от своей команды // START, и ждать первого блока аудио, чтобы завести часы, нельзя. FEndsAt := GetTickCount64 + QWord(AMaxSec) * 1000; end; destructor TTCIRecorder.Destroy; begin // Бюджет возвращаем и на аварийном пути: рекордер освобождают и по уходу // клиента, и по исчезновению приёмника, и на Stop — DropData зовут не все. TCIRecBudgetFree(FBytes); FBytes := 0; FLock.Free; inherited; end; function TTCIRecorder.EnsureRoom: Boolean; // Под FLock. True — в последнем куске есть место хотя бы под один сэмпл. var Need: Int64; N: Integer; begin Result := False; if FCount >= FCap then Exit; // набрали свои MaxSec N := Length(FChunks); if FCount < N * FChunk then Exit(True); // место в текущем куске if FStarved then Exit; // бюджет уже отказал — не долбим его Need := Int64(FChunk) * 2 * SizeOf(SmallInt); if not TCIRecBudgetTake(Need) then begin // Отказ не рушит запись: то, что успели набрать, остаётся сохраняемым. FStarved := True; Exit; end; SetLength(FChunks, N + 1); SetLength(FChunks[N], FChunk * 2); Inc(FBytes, Need); Result := True; end; procedure TTCIRecorder.DropData; // Под FLock. Память отдаём сразу: истёкшая запись на 300 с держала бы 57 МБ // до тех пор, пока клиент не вспомнит про BREAK. var i: Integer; begin FCount := 0; FExpired := True; for i := 0 to High(FChunks) do FChunks[i] := nil; FChunks := nil; TCIRecBudgetFree(FBytes); FBytes := 0; end; function TTCIRecorder.Expired: Boolean; begin FLock.Enter; try if (not FExpired) and (GetTickCount64 >= FEndsAt) then DropData; Result := FExpired; finally FLock.Leave; end; end; procedure TTCIRecorder.Feed(const L, R: array of Single; N: Integer); // DSP-поток. Пишем линейно до конца окна; истекло — данных больше нет (§4.3). var i, j, C: Integer; A, B: Single; begin if (N <= 0) or (FCap <= 0) then Exit; if N > Length(L) then N := Length(L); if N > Length(R) then N := Length(R); FLock.Enter; try if FExpired then Exit; // Часы и ёмкость закрывают окно вместе: ёмкость — это те же MaxSec звука, // но при паузе в аудио доживёт до конца только счёт по часам. if GetTickCount64 >= FEndsAt then begin DropData; Exit; end; if N > FCap - FCount then N := FCap - FCount; for i := 0 to N - 1 do begin // Место спрашиваем на каждый сэмпл: кусок мог кончиться посередине // блока, а бюджет — отказать (тогда дописываем ровно до его границы). if not EnsureRoom then Break; A := L[i]; B := R[i]; if A > 1.0 then A := 1.0; if A < -1.0 then A := -1.0; if B > 1.0 then B := 1.0; if B < -1.0 then B := -1.0; C := FCount div FChunk; j := (FCount mod FChunk) * 2; FChunks[C][j] := Round(A * 32767); FChunks[C][j + 1] := Round(B * 32767); Inc(FCount); end; finally FLock.Leave; end; end; function TTCIRecorder.Take(out Reserved: Int64): TTCIPcm; var C, n, Left: Integer; Copy_: Int64; begin Result := nil; Reserved := 0; FLock.Enter; try // Проверка срока и здесь: аудио могло не идти вовсе (мьют, стоящий // приёмник), и тогда Feed часы не смотрел ни разу. if (not FExpired) and (GetTickCount64 >= FEndsAt) then DropData; if FExpired or (FCount <= 0) then Exit; // Склейка кусков в один буфер — единственное копирование за всю запись, и // оно идёт уже вне DSP-потока: SAVE вынимает рекордер из таблицы раньше, // чем зовёт Take, так что кормить его больше некому. // ★Копия НЕ берёт у бюджета новых денег и НЕ отпускает старых: она // наследует резерв кусков, из которых собрана. Иначе SAVE был бы дырой в // потолке — DropData возвращал бы куски сразу, а копия жила бы дальше в // писателе неучтённой, и на медленном (или зависшем) сетевом каталоге // writer-потоки копились бы без всякой границы. Copy_ := Int64(FCount) * 2 * SizeOf(SmallInt); if Copy_ > FBytes then Copy_ := FBytes; // не отдать больше, чем занято SetLength(Result, FCount * 2); Left := FCount; for C := 0 to High(FChunks) do begin if Left <= 0 then Break; n := FChunk; if n > Left then n := Left; Move(FChunks[C][0], Result[(FCount - Left) * 2], n * 2 * SizeOf(SmallInt)); Dec(Left, n); end; // Из общего резерва оставляем ровно то, что перешло в копию, остальное // (хвост последнего куска) возвращаем — этим и займётся DropData. Reserved := Copy_; Dec(FBytes, Copy_); DropData; // SAVE завершает запись (§4.3) finally FLock.Leave; end; end; { ═══════════════════════════════════════════════════════════════════════════ WAV ═══════════════════════════════════════════════════════════════════════════ } constructor TTCIWavWriter.Create(const APath: string; const AData: TTCIPcm; ARateHz: Integer; AReserved: Int64); begin inherited Create(True); FreeOnTerminate := True; FPath := APath; FData := AData; FRate := ARateHz; FReserved := AReserved; Start; end; procedure TTCIWavWriter.ReleaseData; // Данные и их место в бюджете уходят вместе и ровно один раз — сколько бы // путей выхода ни было у Execute. begin FData := nil; TCIRecBudgetFree(FReserved); FReserved := 0; end; procedure TTCIWavWriter.Execute; // Заголовок собираем в буфере: WAV — это фиксированные 44 байта, и городить // два десятка отдельных Write ради них незачем (а строковые литералы в // нетипизированный Write в FPC ещё и передаются не тем, чем кажется). var FS: THandleStream; H: THandle; Hdr: array[0..43] of Byte; DataBytes: LongWord; procedure PutTag(Ofs: Integer; const Tag: string); var i: Integer; begin for i := 1 to Length(Tag) do Hdr[Ofs + i - 1] := Byte(Tag[i]); end; procedure PutU32(Ofs: Integer; V: LongWord); begin Hdr[Ofs] := Byte(V); Hdr[Ofs + 1] := Byte(V shr 8); Hdr[Ofs + 2] := Byte(V shr 16); Hdr[Ofs + 3] := Byte(V shr 24); end; procedure PutU16(Ofs: Integer; V: Word); begin Hdr[Ofs] := Byte(V); Hdr[Ofs + 1] := Byte(V shr 8); end; begin try DataBytes := LongWord(Length(FData) * SizeOf(SmallInt)); FillChar(Hdr, SizeOf(Hdr), 0); PutTag(0, 'RIFF'); PutU32(4, 36 + DataBytes); PutTag(8, 'WAVE'); PutTag(12, 'fmt '); PutU32(16, 16); // размер fmt-блока PutU16(20, 1); // PCM PutU16(22, 2); // каналов PutU32(24, LongWord(FRate)); PutU32(28, LongWord(FRate) * 2 * 2); // байт в секунду PutU16(32, 4); // выравнивание блока PutU16(34, 16); // бит на сэмпл PutTag(36, 'data'); PutU32(40, DataBytes); // ★Не TFileStream/fmCreate: имя пришло из сети, и затирать им чужой файл // нельзя. TCICreateNewFile создаёт только новый и не идёт по симлинку. H := TCICreateNewFile(FPath); // Не вышло (файл уже есть, нет прав, нет каталога) — просто уходим: выйти // отсюда через Exit нельзя, ниже ещё возврат памяти под данные. if H <> THandle(-1) then begin FS := THandleStream.Create(H); try FS.Write(Hdr[0], SizeOf(Hdr)); if DataBytes > 0 then FS.Write(FData[0], DataBytes); finally FS.Free; FileClose(H); end; end; except // Записать не вышло (нет прав, нет каталога, диск полон) — сказать об этом // клиенту уже некому: команда давно подтверждена. Молчим, но и не падаем: // исключение из потока утащило бы за собой процесс. end; ReleaseData; end; { ═══════════════════════════════════════════════════════════════════════════ Утилиты ═══════════════════════════════════════════════════════════════════════════ } function TCIDecimFactor(SrcRate, WantRate: Integer): Integer; var F: Integer; begin Result := 1; if (SrcRate <= 0) or (WantRate <= 0) or (WantRate >= SrcRate) then Exit; // Идём от большего прореживания к меньшему и берём первое, которое делит // входную частоту нацело и не опускает нас ниже запрошенной. F := SrcRate div WantRate; while F > 1 do begin if (SrcRate mod F = 0) and (SrcRate div F >= WantRate) then Exit(F); Dec(F); end; end; function TCIPickIQRate(SrcRate, WantRate: Integer): Integer; const LEGAL: array[0..3] of Integer = (384000, 192000, 96000, 48000); var i: Integer; begin Result := WantRate; if (SrcRate <= 0) or (WantRate <= 0) then Exit; for i := 0 to High(LEGAL) do if (LEGAL[i] <= WantRate) and (LEGAL[i] <= SrcRate) and (SrcRate mod LEGAL[i] = 0) then Exit(LEGAL[i]); end; { ═══════════════════════════════════════════════════════════════════════════ Бюджет памяти рекордеров ═══════════════════════════════════════════════════════════════════════════ } var RecBudgetLock: TCriticalSection = nil; RecBudgetUsed: Int64 = 0; function TCIRecBudgetTake(Bytes: Int64): Boolean; begin Result := False; if Bytes <= 0 then Exit(True); if RecBudgetLock = nil then Exit; RecBudgetLock.Enter; try if RecBudgetUsed + Bytes > TCI_RECORD_MAX_BYTES then Exit; Inc(RecBudgetUsed, Bytes); Result := True; finally RecBudgetLock.Leave; end; end; procedure TCIRecBudgetFree(Bytes: Int64); begin if (Bytes <= 0) or (RecBudgetLock = nil) then Exit; RecBudgetLock.Enter; try Dec(RecBudgetUsed, Bytes); if RecBudgetUsed < 0 then RecBudgetUsed := 0; finally RecBudgetLock.Leave; end; end; function TCIRecBudgetUsed: Int64; begin Result := 0; if RecBudgetLock = nil then Exit; RecBudgetLock.Enter; try Result := RecBudgetUsed; finally RecBudgetLock.Leave; end; end; { ═══════════════════════════════════════════════════════════════════════════ Имя файла записи ═══════════════════════════════════════════════════════════════════════════ } function TCIRecordPath(const BaseDir, Req: string): string; const MAX_NAME = 120; var S, Name: string; i: Integer; begin Result := ''; if BaseDir = '' then Exit; S := Req; // ':' в протоколе запрещён и заменён на '|' (§4.3) — возвращаем на место, // иначе 'D|\rec\a.wav' стало бы каталогом с именем 'D|'. Слэши допускаются // любые, приводим к своему: без этого ExtractFileName на Linux не увидит // разделителя в 'D:\rec\a.wav' и примет всю строку за имя файла. for i := 1 to Length(S) do if S[i] = '|' then S[i] := ':'; for i := 1 to Length(S) do if (S[i] = '/') or (S[i] = '\') then S[i] := PathDelim; // ★Каталог из просьбы отбрасываем целиком — вместе с '..', абсолютным путём // и буквой диска. Именно здесь закрывается запись в произвольный файл. Name := ExtractFileName(S); if (Name = '') or (Name = '.') or (Name = '..') then Exit; if Length(Name) > MAX_NAME then Exit; for i := 1 to Length(Name) do if (Name[i] < ' ') or (Name[i] = ':') then Exit; // Пишем мы только WAV, а расширение — единственное, по чему клиент потом // узнает файл. Чужое расширение здесь честнее молчаливой подмены. if not SameText(ExtractFileExt(Name), '.wav') then Exit; Result := IncludeTrailingPathDelimiter(BaseDir) + Name; end; function TCICreateNewFile(const Path: string): THandle; begin {$IFDEF UNIX} // O_EXCL — не трогать существующий файл, O_NOFOLLOW — не идти по симлинку, // подложенному вместо него. Обе проверки делает ядро одним вызовом: FileExists // перед FileCreate оставлял бы окно между проверкой и созданием. Result := THandle(FpOpen(PChar(Path), O_WRONLY or O_CREAT or O_EXCL or O_NOFOLLOW, &644)); {$ELSE} // CREATE_NEW = «создать, только если файла нет»; симлинки в Windows без прав // администратора не создаются, отдельной защиты от них не нужно. Result := CreateFileW(PWideChar(UnicodeString(Path)), GENERIC_WRITE, 0, nil, CREATE_NEW, FILE_ATTRIBUTE_NORMAL, 0); {$ENDIF} end; initialization RecBudgetLock := TCriticalSection.Create; finalization FreeAndNil(RecBudgetLock); end.