ewsdr/TCIStreams.pas

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 — либо не записано ничего,
      либо окно уже истекло (по документу это одно и то же: записи нет). }
    function  Take: TTCIPcm;
    { Окно закрылось по часам. Спрашивает тик сервера: у мёртвого приёмника
      Feed не зовут вовсе, и без этого вопроса память жила бы до Stop. }
    function  Expired: Boolean;
    property Rx:   Integer read FRx;
    property Rate: Integer read FRate;
    property Owner: TObject read FOwner;
  end;

  { Писатель WAV в своём потоке: файл до 60 МБ, а зовут сохранение из тика
    сервера — блокировать его на секунду диска нельзя. Данные забирает себе. }
  TTCIWavWriter = class(TThread)
  private
    FPath: string;
    FData: TTCIPcm;
    FRate: Integer;
  protected
    procedure Execute; override;
  public
    constructor Create(const APath: string; const AData: TTCIPcm;
                       ARateHz: Integer);
  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: TTCIPcm;
var
  C, n, Left: Integer;
begin
  Result := nil;
  FLock.Enter;
  try
    // Проверка срока и здесь: аудио могло не идти вовсе (мьют, стоящий
    // приёмник), и тогда Feed часы не смотрел ни разу.
    if (not FExpired) and (GetTickCount64 >= FEndsAt) then DropData;
    if FExpired or (FCount <= 0) then Exit;
    // Склейка кусков в один буфер — единственное копирование за всю запись, и
    // оно идёт уже вне DSP-потока: SAVE вынимает рекордер из таблицы раньше,
    // чем зовёт Take, так что кормить его больше некому.
    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;   // SAVE завершает запись (§4.3)
  finally
    FLock.Leave;
  end;
end;

{ ═══════════════════════════════════════════════════════════════════════════
  WAV
  ═══════════════════════════════════════════════════════════════════════════ }

constructor TTCIWavWriter.Create(const APath: string;
  const AData: TTCIPcm; ARateHz: Integer);
begin
  inherited Create(True);
  FreeOnTerminate := True;
  FPath := APath;
  FData := AData;
  FRate := ARateHz;
  Start;
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;
  FData := nil;
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.