dragonflight/services/capture/deltacast-bridge/main.c

/* services/capture/deltacast-bridge/main.c
 *
 * Deltacast VideoMaster SDI shared multi-port bridge daemon.
 *
 * Opens the board ONCE, opens RX streams for all requested ports, and
 * writes each port's video/audio to named FIFOs in a shared directory.
 * One reader thread + one audio thread per port run concurrently.
 *
 * Usage:
 *   deltacast-bridge --device <N> --ports <csv>
 *                    [--video-pipe-dir /dev/shm/deltacast]
 *                    [--audio-pipe-dir /dev/shm/deltacast]
 *                    [--signal-timeout <sec>]
 *
 * Compat alias: --port <N>  treated as --ports <N> (single port).
 *
 * For each port that acquires signal, emits one JSON line to stderr:
 *   {"port":N,"width":W,"height":H,"fps_num":N,"fps_den":D,
 *    "pix_fmt":"uyvy422","audio_rate":48000,"audio_channels":2}
 *
 * Runs until SIGTERM/SIGINT, then closes all streams and the board.
 */

#include <errno.h>
#include <fcntl.h>
#include <pthread.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>

#include "VideoMasterHD_Core.h"
#include "VideoMasterHD_Sdi.h"
#include "VideoMasterHD_Sdi_Audio.h"

/* ── Globals ──────────────────────────────────────────────────────────── */
static atomic_int g_stop = 0;

static void on_signal(int s) { (void)s; atomic_store(&g_stop, 1); }

/* ── Constants ────────────────────────────────────────────────────────── */
#define MAX_PORTS 8

/* ── Stream type by port index (non-contiguous SDK enum) ────────────── */
static ULONG rx_streamtype(unsigned port) {
    switch (port) {
        case 0: return VHD_ST_RX0;
        case 1: return VHD_ST_RX1;
        case 2: return VHD_ST_RX2;
        case 3: return VHD_ST_RX3;
        case 4: return VHD_ST_RX4;
        case 5: return VHD_ST_RX5;
        case 6: return VHD_ST_RX6;
        case 7: return VHD_ST_RX7;
        default:
            fprintf(stderr, "{\"error\":\"port %u not supported (max 7)\"}\n", port);
            return VHD_ST_RX0;
    }
}

/* ── Loopback board property by port index ───────────────────────────── */
static ULONG loopback_prop(unsigned port) {
    switch (port) {
        case 0: return VHD_CORE_BP_PASSIVE_LOOPBACK_0;
        case 1: return VHD_CORE_BP_PASSIVE_LOOPBACK_1;
        case 2: return VHD_CORE_BP_PASSIVE_LOOPBACK_2;
        case 3: return VHD_CORE_BP_PASSIVE_LOOPBACK_3;
        default: return VHD_CORE_BP_PASSIVE_LOOPBACK_0;
    }
}

/* ── Video standard → width/height/fps/interlaced ───────────────────── */
typedef struct { int width, height, fps_num, fps_den; int interlaced; } VideoInfo;

static VideoInfo video_info(VHD_VIDEOSTANDARD std, VHD_CLOCKDIVISOR div) {
    int ntsc = (div == VHD_CLOCKDIV_1001);
    switch (std) {
        case VHD_VIDEOSTD_S274M_1080p_25Hz:    return (VideoInfo){1920,1080,25,1,0};
        case VHD_VIDEOSTD_S274M_1080p_30Hz:    return (VideoInfo){1920,1080,ntsc?30000:30,ntsc?1001:1,0};
        case VHD_VIDEOSTD_S274M_1080p_24Hz:    return (VideoInfo){1920,1080,ntsc?24000:24,ntsc?1001:1,0};
        case VHD_VIDEOSTD_S274M_1080p_50Hz:    return (VideoInfo){1920,1080,50,1,0};
        case VHD_VIDEOSTD_S274M_1080p_60Hz:    return (VideoInfo){1920,1080,ntsc?60000:60,ntsc?1001:1,0};
        case VHD_VIDEOSTD_S274M_1080psf_24Hz:  return (VideoInfo){1920,1080,ntsc?24000:24,ntsc?1001:1,0};
        case VHD_VIDEOSTD_S274M_1080psf_25Hz:  return (VideoInfo){1920,1080,25,1,0};
        case VHD_VIDEOSTD_S274M_1080psf_30Hz:  return (VideoInfo){1920,1080,ntsc?30000:30,ntsc?1001:1,0};
        case VHD_VIDEOSTD_S274M_1080i_50Hz:    return (VideoInfo){1920,1080,25,1,1};
        case VHD_VIDEOSTD_S274M_1080i_60Hz:    return (VideoInfo){1920,1080,ntsc?30000:30,ntsc?1001:1,1};
        case VHD_VIDEOSTD_S296M_720p_50Hz:     return (VideoInfo){1280,720,50,1,0};
        case VHD_VIDEOSTD_S296M_720p_60Hz:     return (VideoInfo){1280,720,ntsc?60000:60,ntsc?1001:1,0};
        case VHD_VIDEOSTD_S296M_720p_25Hz:     return (VideoInfo){1280,720,25,1,0};
        case VHD_VIDEOSTD_S296M_720p_30Hz:     return (VideoInfo){1280,720,ntsc?30000:30,ntsc?1001:1,0};
        case VHD_VIDEOSTD_S296M_720p_24Hz:     return (VideoInfo){1280,720,ntsc?24000:24,ntsc?1001:1,0};
        case VHD_VIDEOSTD_3840x2160p_24Hz:     return (VideoInfo){3840,2160,ntsc?24000:24,ntsc?1001:1,0};
        case VHD_VIDEOSTD_3840x2160p_25Hz:     return (VideoInfo){3840,2160,25,1,0};
        case VHD_VIDEOSTD_3840x2160p_30Hz:     return (VideoInfo){3840,2160,ntsc?30000:30,ntsc?1001:1,0};
        case VHD_VIDEOSTD_3840x2160p_50Hz:     return (VideoInfo){3840,2160,50,1,0};
        case VHD_VIDEOSTD_3840x2160p_60Hz:     return (VideoInfo){3840,2160,ntsc?60000:60,ntsc?1001:1,0};
        case VHD_VIDEOSTD_S259M_NTSC_480:      return (VideoInfo){720,480,ntsc?30000:30,ntsc?1001:1,1};
        default:                               return (VideoInfo){1920,1080,25,1,0};
    }
}

/* ── Write-all helper ─────────────────────────────────────────────────── */
static int write_all(int fd, const unsigned char *p, size_t len) {
    size_t off = 0;
    while (off < len) {
        ssize_t n = write(fd, p + off, len - off);
        if (n > 0) { off += (size_t)n; continue; }
        if (n < 0 && errno == EINTR) continue;
        return -1;
    }
    return 0;
}

/* ── Per-port state ───────────────────────────────────────────────────── */
typedef struct {
    HANDLE        board;
    unsigned      port;
    unsigned      device;
    ULONG         video_std;
    ULONG         clock_div;
    VideoInfo     vi;
    char          video_fifo[256];
    char          audio_fifo[256];
    /* threads */
    pthread_t     video_tid;
    pthread_t     audio_tid;
    /* streams (owned by threads, set before thread launch) */
    HANDLE        video_stream;
} PortState;

/* ── Audio thread ──────────────────────────────────────────────────────
 *
 * Identical design to the single-port bridge audio thread:
 *   - Opens FIFO writer FIRST, unconditionally (unblocks ffmpeg input)
 *   - Feeds continuous wall-clock-paced s16le stereo (real or silence)
 *   - Best-effort VHD audio stream; silence fallback on any failure
 */
static void *audio_thread(void *arg) {
    PortState *ps = (PortState *)arg;

    int fd = open(ps->audio_fifo, O_WRONLY);
    if (fd < 0) {
        fprintf(stderr, "[audio:%u] open FIFO failed: %s\n", ps->port, strerror(errno));
        return NULL;
    }

    const int    AUDIO_RATE    = 48000;
    const int    CHANNELS      = 2;
    const size_t FRAME_BYTES   = (size_t)CHANNELS * 2; /* s16le stereo */
    int fps_num = ps->vi.fps_num > 0 ? ps->vi.fps_num : 25;
    int fps_den = ps->vi.fps_den > 0 ? ps->vi.fps_den : 1;
    long samples_per_frame = ((long)AUDIO_RATE * fps_den + fps_num / 2) / fps_num;
    if (samples_per_frame < 1) samples_per_frame = 1;
    size_t tick_bytes = (size_t)samples_per_frame * FRAME_BYTES;

    ULONG max_samples = VHD_GetNbSamples((VHD_VIDEOSTANDARD)ps->video_std,
                                         (VHD_CLOCKDIVISOR)ps->clock_div,
                                         VHD_ASR_48000, 0);
    ULONG block_size  = VHD_GetBlockSize(VHD_AF_16, VHD_AM_STEREO);
    size_t vhd_buf_sz = ((size_t)max_samples + 64) * (block_size ? block_size : FRAME_BYTES);
    size_t buf_sz     = vhd_buf_sz > tick_bytes ? vhd_buf_sz : tick_bytes;
    unsigned char *buf = calloc(1, buf_sz);
    if (!buf) { close(fd); return NULL; }

    HANDLE stream = NULL;
    int have_vhd_audio = 0;
    VHD_AUDIOINFO ai;
    memset(&ai, 0, sizeof(ai));

    ULONG r = VHD_OpenStreamHandle(ps->board, rx_streamtype(ps->port),
                                   VHD_SDI_STPROC_DISJOINED_ANC,
                                   NULL, &stream, NULL);
    if (r == VHDERR_NOERROR) {
        VHD_SetStreamProperty(stream, VHD_SDI_SP_VIDEO_STANDARD,    ps->video_std);
        VHD_SetStreamProperty(stream, VHD_SDI_SP_CLOCK_SYSTEM,      ps->clock_div);
        VHD_SetStreamProperty(stream, VHD_CORE_SP_TRANSFER_SCHEME,  VHD_TRANSFER_SLAVED);

        ai.pAudioGroups[0].pAudioChannels[0].Mode         = VHD_AM_STEREO;
        ai.pAudioGroups[0].pAudioChannels[0].BufferFormat = VHD_AF_16;
        ai.pAudioGroups[0].pAudioChannels[0].pData        = buf;

        if (VHD_StartStream(stream) == VHDERR_NOERROR) {
            have_vhd_audio = 1;
        } else {
            fprintf(stderr, "[audio:%u] VHD_StartStream failed — feeding silence\n", ps->port);
            VHD_CloseStreamHandle(stream);
            stream = NULL;
        }
    } else {
        fprintf(stderr, "[audio:%u] VHD_OpenStreamHandle failed (%lu) — feeding silence\n",
                ps->port, r);
    }

    struct timespec next;
    clock_gettime(CLOCK_MONOTONIC, &next);
    long frame_ns = (long)(1000000000.0 * (double)fps_den / (double)fps_num);
    HANDLE slot = NULL;

    while (!atomic_load(&g_stop)) {
        size_t out_bytes = 0;

        if (have_vhd_audio) {
            r = VHD_LockSlotHandle(stream, &slot);
            if (r == VHDERR_NOERROR) {
                ai.pAudioGroups[0].pAudioChannels[0].DataSize = (ULONG)buf_sz;
                if (VHD_SlotExtractAudio(slot, &ai) == VHDERR_NOERROR) {
                    ULONG sz = ai.pAudioGroups[0].pAudioChannels[0].DataSize;
                    if (sz > 0 && (size_t)sz <= buf_sz) out_bytes = (size_t)sz;
                }
                VHD_UnlockSlotHandle(slot);
            } else if (r != VHDERR_TIMEOUT) {
                fprintf(stderr, "[audio:%u] lock error %lu — degrading to silence\n",
                        ps->port, r);
                VHD_StopStream(stream);
                VHD_CloseStreamHandle(stream);
                stream = NULL;
                have_vhd_audio = 0;
            }
        }

        if (out_bytes == 0) {
            memset(buf, 0, tick_bytes);
            out_bytes = tick_bytes;
        }

        if (write_all(fd, buf, out_bytes) < 0) {
            atomic_store(&g_stop, 1);
            break;
        }

        next.tv_nsec += frame_ns;
        while (next.tv_nsec >= 1000000000L) { next.tv_nsec -= 1000000000L; next.tv_sec += 1; }
        struct timespec now;
        clock_gettime(CLOCK_MONOTONIC, &now);
        if (next.tv_sec > now.tv_sec ||
            (next.tv_sec == now.tv_sec && next.tv_nsec > now.tv_nsec)) {
            clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, &next, NULL);
        } else {
            next = now;
        }
    }

    close(fd);
    if (stream) {
        VHD_StopStream(stream);
        VHD_CloseStreamHandle(stream);
    }
    free(buf);
    return NULL;
}

/* ── Video thread ─────────────────────────────────────────────────────── */
static void *video_thread(void *arg) {
    PortState *ps = (PortState *)arg;

    int fd = open(ps->video_fifo, O_WRONLY);
    if (fd < 0) {
        fprintf(stderr, "[video:%u] open FIFO failed: %s\n", ps->port, strerror(errno));
        return NULL;
    }

    HANDLE slot = NULL;
    while (!atomic_load(&g_stop)) {
        ULONG r = VHD_LockSlotHandle(ps->video_stream, &slot);
        if (r == VHDERR_NOERROR) {
            BYTE  *buf = NULL;
            ULONG  sz  = 0;
            if (VHD_GetSlotBuffer(slot, VHD_SDI_BT_VIDEO, &buf, &sz) == VHDERR_NOERROR) {
                if (write_all(fd, buf, sz) < 0) {
                    atomic_store(&g_stop, 1);
                    VHD_UnlockSlotHandle(slot);
                    break;
                }
            }
            VHD_UnlockSlotHandle(slot);
        } else if (r != VHDERR_TIMEOUT) {
            fprintf(stderr, "[video:%u] VHD_LockSlotHandle error %lu — stopping\n",
                    ps->port, r);
            atomic_store(&g_stop, 1);
            break;
        }
    }

    close(fd);
    return NULL;
}

/* ── Parse comma-separated port list ─────────────────────────────────── */
static int parse_ports(const char *csv, unsigned *ports, int max) {
    int count = 0;
    char buf[256];
    strncpy(buf, csv, sizeof(buf) - 1);
    buf[sizeof(buf) - 1] = '\0';
    char *tok = strtok(buf, ",");
    while (tok && count < max) {
        ports[count++] = (unsigned)atoi(tok);
        tok = strtok(NULL, ",");
    }
    return count;
}

/* ── Main ─────────────────────────────────────────────────────────────── */
int main(int argc, char *argv[]) {
    unsigned device_id         = 0;
    unsigned ports[MAX_PORTS]  = {0};
    int      port_count        = 0;
    int      sig_timeout       = 30;
    const char *video_pipe_dir = "/dev/shm/deltacast";
    const char *audio_pipe_dir = "/dev/shm/deltacast";

    for (int i = 1; i < argc; i++) {
        if (!strcmp(argv[i], "--device") && i+1 < argc) {
            device_id = (unsigned)atoi(argv[++i]);
        } else if (!strcmp(argv[i], "--ports") && i+1 < argc) {
            port_count = parse_ports(argv[++i], ports, MAX_PORTS);
        } else if (!strcmp(argv[i], "--port") && i+1 < argc) {
            /* single-port compat alias */
            ports[0] = (unsigned)atoi(argv[++i]);
            port_count = 1;
        } else if (!strcmp(argv[i], "--video-pipe-dir") && i+1 < argc) {
            video_pipe_dir = argv[++i];
        } else if (!strcmp(argv[i], "--audio-pipe-dir") && i+1 < argc) {
            audio_pipe_dir = argv[++i];
        } else if (!strcmp(argv[i], "--signal-timeout") && i+1 < argc) {
            sig_timeout = atoi(argv[++i]);
        }
    }

    if (port_count == 0) {
        fprintf(stderr, "{\"error\":\"no ports specified — use --ports 0,1,2,...\"}\n");
        return 1;
    }

    signal(SIGINT,  on_signal);
    signal(SIGTERM, on_signal);
    signal(SIGPIPE, SIG_IGN);

    /* ── Init API ────────────────────────────────────────────────────── */
    ULONG dll_ver, nb_boards;
    if (VHD_GetApiInfo(&dll_ver, &nb_boards) != VHDERR_NOERROR) {
        fprintf(stderr, "{\"error\":\"VHD_GetApiInfo failed\"}\n");
        return 1;
    }
    if (device_id >= nb_boards) {
        fprintf(stderr, "{\"error\":\"board %u not found (%lu detected)\"}\n",
                device_id, nb_boards);
        return 1;
    }

    /* ── Open board ONCE ─────────────────────────────────────────────── */
    HANDLE board = NULL;
    if (VHD_OpenBoardHandle(device_id, &board, NULL, 0) != VHDERR_NOERROR) {
        fprintf(stderr, "{\"error\":\"VHD_OpenBoardHandle failed for board %u\"}\n", device_id);
        return 1;
    }
    fprintf(stderr, "[board] opened board %u with %d port(s)\n", device_id, port_count);

    /* Disable passive loopback for each requested port (ports 0-3 only in SDK). */
    for (int pi = 0; pi < port_count; pi++) {
        unsigned p = ports[pi];
        if (p < 4) VHD_SetBoardProperty(board, loopback_prop(p), FALSE);
    }

    /* ── Wait for signal on all ports ───────────────────────────────── */
    ULONG video_stds[MAX_PORTS] = {0};
    ULONG clock_divs[MAX_PORTS] = {0};
    int   locked[MAX_PORTS]     = {0};

    for (int pi = 0; pi < port_count; pi++) {
        video_stds[pi] = (ULONG)NB_VHD_VIDEOSTANDARDS;
        clock_divs[pi] = VHD_CLOCKDIV_1;
    }

    struct timespec deadline;
    clock_gettime(CLOCK_MONOTONIC, &deadline);
    deadline.tv_sec += sig_timeout;

    while (!atomic_load(&g_stop)) {
        struct timespec now;
        clock_gettime(CLOCK_MONOTONIC, &now);
        if (now.tv_sec > deadline.tv_sec ||
            (now.tv_sec == deadline.tv_sec && now.tv_nsec >= deadline.tv_nsec)) break;

        int all_locked = 1;
        for (int pi = 0; pi < port_count; pi++) {
            if (locked[pi]) continue;
            VHD_GetChannelProperty(board, VHD_RX_CHANNEL, ports[pi],
                                   VHD_SDI_CP_VIDEO_STANDARD, &video_stds[pi]);
            if (video_stds[pi] != (ULONG)NB_VHD_VIDEOSTANDARDS) {
                VHD_GetChannelProperty(board, VHD_RX_CHANNEL, ports[pi],
                                       VHD_SDI_CP_CLOCK_DIVISOR, &clock_divs[pi]);
                locked[pi] = 1;
                fprintf(stderr, "[board] port %u signal locked (std=%lu)\n",
                        ports[pi], video_stds[pi]);
            } else {
                all_locked = 0;
            }
        }
        if (all_locked) break;

        struct timespec ts = {0, 200000000L}; /* 200ms poll */
        nanosleep(&ts, NULL);
    }

    /* Report results — continue with whatever locked, abort only if NONE locked. */
    int any_locked = 0;
    for (int pi = 0; pi < port_count; pi++) {
        if (locked[pi]) { any_locked = 1; }
        else {
            fprintf(stderr,
                "{\"error\":\"no signal on board %u port %u within %ds\"}\n",
                device_id, ports[pi], sig_timeout);
        }
    }
    if (!any_locked || atomic_load(&g_stop)) {
        VHD_CloseBoardHandle(board);
        return 1;
    }

    /* ── Create FIFOs and open streams for each locked port ─────────── */
    PortState ps[MAX_PORTS];
    memset(ps, 0, sizeof(ps));
    int active_count = 0;

    for (int pi = 0; pi < port_count; pi++) {
        if (!locked[pi]) continue;
        PortState *p = &ps[active_count];
        p->board      = board;
        p->port       = ports[pi];
        p->device     = device_id;
        p->video_std  = video_stds[pi];
        p->clock_div  = clock_divs[pi];
        p->vi         = video_info((VHD_VIDEOSTANDARD)video_stds[pi],
                                   (VHD_CLOCKDIVISOR)clock_divs[pi]);

        snprintf(p->video_fifo, sizeof(p->video_fifo),
                 "%s/video-%u.fifo", video_pipe_dir, ports[pi]);
        snprintf(p->audio_fifo, sizeof(p->audio_fifo),
                 "%s/audio-%u.fifo", audio_pipe_dir, ports[pi]);

        /* Create FIFOs (mkfifo; ignore EEXIST). */
        if (mkfifo(p->video_fifo, 0666) != 0 && errno != EEXIST) {
            fprintf(stderr, "[port:%u] mkfifo video failed: %s\n", ports[pi], strerror(errno));
            continue;
        }
        if (mkfifo(p->audio_fifo, 0666) != 0 && errno != EEXIST) {
            fprintf(stderr, "[port:%u] mkfifo audio failed: %s\n", ports[pi], strerror(errno));
            continue;
        }

        /* Open video stream. */
        HANDLE vs = NULL;
        ULONG r = VHD_OpenStreamHandle(board, rx_streamtype(ports[pi]),
                                       VHD_SDI_STPROC_DISJOINED_VIDEO,
                                       NULL, &vs, NULL);
        if (r != VHDERR_NOERROR) {
            fprintf(stderr, "{\"error\":\"VHD_OpenStreamHandle video failed port %u rc=%lu\"}\n",
                    ports[pi], r);
            continue;
        }
        VHD_SetStreamProperty(vs, VHD_SDI_SP_VIDEO_STANDARD,      p->video_std);
        VHD_SetStreamProperty(vs, VHD_SDI_SP_CLOCK_SYSTEM,        p->clock_div);
        VHD_SetStreamProperty(vs, VHD_CORE_SP_TRANSFER_SCHEME,    VHD_TRANSFER_SLAVED);
        VHD_SetStreamProperty(vs, VHD_CORE_SP_BUFFERQUEUE_DEPTH,  8);
        p->video_stream = vs;

        if (VHD_StartStream(vs) != VHDERR_NOERROR) {
            fprintf(stderr, "{\"error\":\"VHD_StartStream video failed port %u\"}\n", ports[pi]);
            VHD_CloseStreamHandle(vs);
            p->video_stream = NULL;
            continue;
        }

        /* Emit format JSON to stderr (one line per port on signal lock). */
        fprintf(stderr,
            "{\"port\":%u,\"width\":%d,\"height\":%d,"
            "\"fps_num\":%d,\"fps_den\":%d,"
            "\"interlaced\":%s,"
            "\"pix_fmt\":\"uyvy422\","
            "\"audio_channels\":2,\"audio_rate\":48000,"
            "\"device\":%u}\n",
            ports[pi],
            p->vi.width, p->vi.height,
            p->vi.fps_num, p->vi.fps_den,
            p->vi.interlaced ? "true" : "false",
            device_id);
        fflush(stderr);

        /* Launch audio thread (blocks until reader connects to audio FIFO). */
        pthread_create(&p->audio_tid, NULL, audio_thread, p);

        /* Launch video thread (blocks until reader connects to video FIFO). */
        pthread_create(&p->video_tid, NULL, video_thread, p);

        active_count++;
    }

    if (active_count == 0) {
        fprintf(stderr, "{\"error\":\"no ports successfully started\"}\n");
        VHD_CloseBoardHandle(board);
        return 1;
    }

    /* ── Wait for all threads to finish ─────────────────────────────── */
    for (int i = 0; i < active_count; i++) {
        if (ps[i].video_tid) pthread_join(ps[i].video_tid, NULL);
        if (ps[i].audio_tid) pthread_join(ps[i].audio_tid, NULL);
    }

    /* ── Cleanup ─────────────────────────────────────────────────────── */
    for (int i = 0; i < active_count; i++) {
        if (ps[i].video_stream) {
            VHD_StopStream(ps[i].video_stream);
            VHD_CloseStreamHandle(ps[i].video_stream);
        }
    }
    VHD_CloseBoardHandle(board);

    return 0;
}