Add lab restream pipeline with pirate_drift logo and lib/env.sh.

Encode-once/fanout scripts, HLS output, and single overloadable env source. Commits assets/pirate_drift.png as default watermark for cluster runs. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-29 04:38:08 +03:00 · 2026-07-11 15:27:24 +02:00
parent d990862b64
commit 5ca8a9f539
27 changed files with 911 additions and 1 deletions
--- a/lib/common.sh
+++ b/lib/common.sh
@@ -0,0 +1,98 @@
+#!/usr/bin/env bash
+# Shared bootstrap: ROOT, env defaults, encoding presets, pipe helpers.
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+export ROOT
+
+# shellcheck disable=SC1091
+source "${ROOT}/lib/logging.sh"
+# shellcheck disable=SC1091
+source "${ROOT}/lib/env.sh"
+
+apply_encoding_preset() {
+  case "${OUTPUT_PRESET}" in
+    720p)
+      WIDTH=1280
+      HEIGHT=720
+      FASTNESS=veryfast
+      BV=2500k
+      BUFSIZE=5000k
+      ;;
+    1080p|*)
+      WIDTH=1920
+      HEIGHT=1080
+      FASTNESS=faster
+      BV=5000k
+      BUFSIZE=10000k
+      ;;
+  esac
+  export WIDTH HEIGHT FASTNESS BV BUFSIZE
+  RES="${WIDTH}x${HEIGHT}"
+  PROFILE=high
+  KEYFRAMES_INTL="${FPS}"
+  GOP_INTL=$((KEYFRAMES_INTL * 2))
+  PIX_FMT="-pix_fmt yuv420p"
+  PRESET="-preset ${FASTNESS} -tune zerolatency -profile:v ${PROFILE} ${PIX_FMT} -x264-params keyint=${KEYFRAMES_INTL}:min-keyint=${KEYFRAMES_INTL} -g ${GOP_INTL} -flags +global_header"
+  BITRATE="-b:v ${BV} -maxrate ${BV} -minrate ${BV}"
+  VCODEC="-c:v libx264"
+  ACODEC="-c:a aac -b:a 128k"
+  export RES PROFILE PRESET BITRATE VCODEC ACODEC
+}
+
+apply_encoding_preset
+
+FFMPEG_TS_FLAGS=(
+  -fflags +genpts+igndts
+  -avoid_negative_ts make_zero
+  -sn
+  -movflags +faststart
+)
+
+read_rate_flag() {
+  if [[ "${NEAR_REALTIME}" == "1" ]]; then
+    RE_FLAG=(-re)
+  else
+    RE_FLAG=()
+  fi
+}
+
+ensure_storage() {
+  mkdir -p "${STORAGE_DIR}" "${RUN_DIR}"
+}
+
+ensure_pipe() {
+  if [[ ! -p "${PIPE_PATH}" ]]; then
+    mkfifo "${PIPE_PATH}"
+    if command -v perl >/dev/null 2>&1; then
+      perl -MFcntl -e 'fcntl(STDIN, 1031, shift) or die $!' "${PIPE_FIFO_SIZE}" <> "${PIPE_PATH}" || true
+    fi
+  fi
+  exec 7<> "${PIPE_PATH}"
+}
+
+remove_pipe() {
+  exec 7>&- 7<&- 2>/dev/null || true
+  rm -f "${PIPE_PATH}" 2>/dev/null || true
+}
+
+next_record_path() {
+  local counter=0 name
+  ensure_storage
+  while true; do
+    name="$(printf "video%04d.%s" "${counter}" "${RECORD_FORMAT}")"
+    if [[ ! -f "${STORAGE_DIR}/${name}" ]]; then
+      echo "${STORAGE_DIR}/${name}"
+      return
+    fi
+    counter=$((counter + 1))
+  done
+}
+
+require_cmd() {
+  command -v "$1" >/dev/null 2>&1 || die "required command not found: $1"
+}
+
+require_logo() {
+  [[ -f "${LOGO_PATH}" ]] || die "logo not found at ${LOGO_PATH} (expected ${ROOT}/assets/pirate_drift.png)"
+}
--- a/lib/env.sh
+++ b/lib/env.sh
@@ -0,0 +1,65 @@
+#!/usr/bin/env bash
+# Single source of truth for pipeline variables.
+# Override: export VAR=value before running, or set in config/local.env (sourced last).
+#
+# Requires ROOT (set by lib/common.sh before this file is sourced).
+
+if [[ -z "${ROOT:-}" ]]; then
+  ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+  export ROOT
+fi
+
+# --- binaries ---
+export FFMPEG_BIN="${FFMPEG_BIN:-ffmpeg}"
+export FFPROBE_BIN="${FFPROBE_BIN:-ffprobe}"
+export YTDLP_BIN="${YTDLP_BIN:-yt-dlp}"
+
+# --- paths (logo default: assets/pirate_drift.png in repo root) ---
+export LOGO_PATH="${LOGO_PATH:-${ROOT}/assets/pirate_drift.png}"
+export PIPE_NAME="${PIPE_NAME:-.stream_pipe}"
+export PIPE_PATH="${PIPE_PATH:-${ROOT}/${PIPE_NAME}}"
+export PIPE_FIFO_SIZE="${PIPE_FIFO_SIZE:-1048576}"
+export STORAGE_DIR="${STORAGE_DIR:-${ROOT}/storage}"
+export RUN_DIR="${RUN_DIR:-${ROOT}/run}"
+export SOURCES_LIST="${SOURCES_LIST:-${ROOT}/config/sources.list}"
+export SINKS_LIST="${SINKS_LIST:-${ROOT}/config/sinks.list}"
+export HLS_OUTPUT_DIR="${HLS_OUTPUT_DIR:-}"
+
+# --- encode quality ---
+export OUTPUT_PRESET="${OUTPUT_PRESET:-1080p}"
+export FILTER_PROFILE="${FILTER_PROFILE:-default}"
+export FPS="${FPS:-30}"
+export THREADS="${THREADS:-8}"
+export THREAD_QUEUE_SIZE="${THREAD_QUEUE_SIZE:-512}"
+export RECORD_FORMAT="${RECORD_FORMAT:-mkv}"
+
+# --- ingest behaviour ---
+export YTDLP_ENABLED="${YTDLP_ENABLED:-1}"
+export NEAR_REALTIME="${NEAR_REALTIME:-1}"
+export FANOUT_RETRIES="${FANOUT_RETRIES:-5}"
+
+# optional delogo rectangle (source coords): x:y:w:h
+export DELOGO="${DELOGO:-}"
+
+# separate audio when video track has no audio
+export AUDIO_SOURCE="${AUDIO_SOURCE:-}"
+
+# --- optional file overrides (highest priority among defaults) ---
+_mp_load_env_file() {
+  local f="$1"
+  if [[ -f "${f}" ]]; then
+    set -a
+    # shellcheck disable=SC1090
+    source "${f}"
+    set +a
+  fi
+}
+_mp_load_env_file "${ROOT}/config/local.env"
+
+# Re-apply path defaults if local.env used relative paths
+export LOGO_PATH="${LOGO_PATH:-${ROOT}/assets/pirate_drift.png}"
+export PIPE_PATH="${PIPE_PATH:-${ROOT}/${PIPE_NAME}}"
+export STORAGE_DIR="${STORAGE_DIR:-${ROOT}/storage}"
+export RUN_DIR="${RUN_DIR:-${ROOT}/run}"
+export SOURCES_LIST="${SOURCES_LIST:-${ROOT}/config/sources.list}"
+export SINKS_LIST="${SINKS_LIST:-${ROOT}/config/sinks.list}"
--- a/lib/filters.sh
+++ b/lib/filters.sh
@@ -0,0 +1,45 @@
+#!/usr/bin/env bash
+# filter_complex builders: logo overlay, optional delogo, blurred variant.
+set -euo pipefail
+
+# shellcheck disable=SC1091
+source "$(dirname "${BASH_SOURCE[0]}")/common.sh"
+
+build_filter_complex() {
+  local v_in="${1:-0:v}"
+  local logo_in="${2:-1:v}"
+  local profile="${FILTER_PROFILE}"
+  local chain scaled delogo filter
+
+  case "${profile}" in
+    nologo)
+      FILTER_COMPLEX="[${v_in}]fps=${FPS},scale=${WIDTH}:${HEIGHT}[ovl]"
+      return
+      ;;
+    blurred)
+      chain="[${v_in}]scale=1446:1040,crop=1280:920,scale=1446:1040,crop=1280:920,noise=alls=25:allf=t+u,avgblur=16:16"
+      ;;
+    default|*)
+      chain="[${v_in}]fps=${FPS},scale=${WIDTH}:${HEIGHT}"
+      ;;
+  esac
+
+  if [[ -n "${DELOGO:-}" ]]; then
+    IFS=: read -r dx dy dw dh <<<"${DELOGO}"
+    chain="${chain},delogo=x=${dx}:y=${dy}:w=${dw}:h=${dh}:show=0"
+  fi
+
+  scaled="${chain}[scaled]"
+  filter="[${logo_in}][scaled]scale2ref=(${WIDTH}/${HEIGHT})*ih/8:ih/8[wm][base];"
+  filter+="[wm]setsar=1[wmsar];[base][wmsar]overlay=x=main_w-overlay_w-(main_w*0.01):y=main_h*0.01[ovl]"
+  FILTER_COMPLEX="${scaled};${filter}"
+}
+
+map_video_out() {
+  echo '[ovl]'
+}
+
+map_audio_out() {
+  local audio_input="${AUDIO_MAP_INPUT:-0}"
+  echo "${audio_input}:a:0"
+}
--- a/lib/logging.sh
+++ b/lib/logging.sh
@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+log() { printf '[%s] %s\n' "$(date '+%H:%M:%S')" "$*" >&2; }
+warn() { log "WARN: $*"; }
+die() { log "ERROR: $*"; exit 1; }
--- a/lib/outputs.sh
+++ b/lib/outputs.sh
@@ -0,0 +1,68 @@
+#!/usr/bin/env bash
+# Tee muxer targets: local file, named pipe, HLS dir, RTMP sinks list.
+set -euo pipefail
+
+# shellcheck disable=SC1091
+source "$(dirname "${BASH_SOURCE[0]}")/common.sh"
+
+read_sinks() {
+  SINKS=()
+  [[ -f "${SINKS_LIST}" ]] || return 0
+  while IFS= read -r line; do
+    line="${line%%#*}"
+    line="$(echo "${line}" | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')"
+    [[ -n "${line}" ]] && SINKS+=("${line}")
+  done < "${SINKS_LIST}"
+}
+
+pipe_tee_entry() {
+  local pipe="$1"
+  printf "[select='v:0,a':f=flv:onfail=ignore:use_fifo=1:fifo_options='fifo_format=flv:drop_pkts_on_overflow=1:attempt_recovery=1:recovery_wait_time=1']%s" "${pipe}"
+}
+
+file_tee_entry() {
+  local path="$1"
+  local fmt="${RECORD_FORMAT}"
+  case "${fmt}" in
+    mkv) printf "[select='v:0,a':f=matroska:onfail=ignore]%s" "${path}" ;;
+    flv|*) printf "[select='v:0,a':f=flv:onfail=ignore]%s" "${path}" ;;
+  esac
+}
+
+hls_tee_entry() {
+  local dir="$1"
+  mkdir -p "${dir}"
+  printf "[select='v:0,a':f=hls:onfail=ignore:hls_time=2:hls_list_size=6:hls_flags=delete_segments+append_list]%s/index.m3u8" "${dir}"
+}
+
+rtmp_tee_entry() {
+  local url="$1"
+  printf "[select='v:0,a':f=flv:onfail=ignore]%s" "${url}"
+}
+
+# Encode stage: record + pipe (+ optional HLS). RTMP fanout is copy stage.
+build_encode_tee_spec() {
+  local record="$1"
+  local pipe="$2"
+  local parts=()
+  parts+=("$(file_tee_entry "${record}")")
+  parts+=("$(pipe_tee_entry "${pipe}")")
+  if [[ -n "${HLS_OUTPUT_DIR}" ]]; then
+    parts+=("$(hls_tee_entry "${HLS_OUTPUT_DIR}")")
+  fi
+  local IFS='|'
+  echo "${parts[*]}"
+}
+
+# Fanout stage: copy from pipe to all sinks (no re-encode).
+build_fanout_tee_spec() {
+  read_sinks
+  local parts=()
+  local s
+  for s in "${SINKS[@]}"; do
+    parts+=("$(rtmp_tee_entry "${s}")")
+  done
+  [[ ${#parts[@]} -gt 0 ]] || die "no sinks in ${SINKS_LIST}"
+  local IFS='|'
+  echo "${parts[*]}"
+}
--- a/lib/source.sh
+++ b/lib/source.sh
@@ -0,0 +1,118 @@
+#!/usr/bin/env bash
+# Resolve pipeline input from CLI arg, sources.list index, or env SOURCE_URL.
+set -euo pipefail
+
+# shellcheck disable=SC1091
+source "$(dirname "${BASH_SOURCE[0]}")/common.sh"
+
+SOURCE_URL=""
+SOURCE_FMT=""
+SOURCE_VIDEO_STREAM="v:0"
+SOURCE_AUDIO_STREAM="a:0"
+AUDIO_INPUT_ARGS=()
+USE_YTDLP=0
+
+is_page_url() {
+  local url="$1"
+  [[ "$url" =~ ^https?:// ]] || return 1
+  [[ "$url" =~ \.(m3u8|mpd|mp4|mkv|mov|ts|flv)(\?|$) ]] && return 1
+  [[ -f "$url" ]] && return 1
+  return 0
+}
+
+resolve_source() {
+  local arg="${1:-}"
+  SOURCE_URL="${SOURCE_URL:-}"
+  SOURCE_FMT=""
+
+  if [[ -n "${SOURCE_URL}" ]]; then
+    :
+  elif [[ -n "${arg}" && -f "${arg}" && "${arg}" != *".list" ]]; then
+    SOURCE_URL="${arg}"
+  elif [[ -n "${arg}" && "${arg}" =~ ^https?:// ]]; then
+    SOURCE_URL="${arg}"
+  elif [[ -n "${arg}" && "${arg}" =~ ^[0-9]+$ ]]; then
+    if [[ ! -f "${SOURCES_LIST}" ]]; then
+      die "sources list missing: ${SOURCES_LIST} (copy sources.example.list)"
+    fi
+    local line
+    line="$(sed -n "${arg}p" "${SOURCES_LIST}" | sed 's/#.*//' | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')"
+    [[ -n "${line}" ]] || die "no source on line ${arg} of ${SOURCES_LIST}"
+    SOURCE_URL="$(awk '{print $1}' <<<"${line}")"
+    SOURCE_FMT="$(awk '{print $2}' <<<"${line}")"
+  elif [[ -f "${SOURCES_LIST}" ]]; then
+    local line
+    line="$(grep -v '^[[:space:]]*#' "${SOURCES_LIST}" | grep -v '^[[:space:]]*$' | head -1)"
+    SOURCE_URL="$(awk '{print $1}' <<<"${line}")"
+    SOURCE_FMT="$(awk '{print $2}' <<<"${line}")"
+  else
+    die "usage: pass URL, file path, or 1-based index into ${SOURCES_LIST}"
+  fi
+
+  [[ -n "${SOURCE_URL}" ]] || die "empty source URL"
+
+  if [[ -n "${AUDIO_SOURCE:-}" ]]; then
+    read_rate_flag
+    AUDIO_INPUT_ARGS=("${RE_FLAG[@]}" -thread_queue_size "${THREAD_QUEUE_SIZE}" -i "${AUDIO_SOURCE}")
+    SOURCE_AUDIO_STREAM="a:0"
+  fi
+
+  if [[ "${YTDLP_ENABLED}" == "1" ]] && is_page_url "${SOURCE_URL}"; then
+    USE_YTDLP=1
+  fi
+
+  echo "${SOURCE_URL}" > "${ROOT}/.current_source"
+  log "source: ${SOURCE_URL} (ytdlp=${USE_YTDLP})"
+}
+
+ytdlp_auth_args() {
+  local auth="${ROOT}/config/yt-dlp.auth.env"
+  if [[ -f "${auth}" ]]; then
+    # shellcheck disable=SC1090
+    source "${auth}"
+  fi
+  local args=()
+  [[ -n "${YTDLP_USERNAME:-}" ]] && args+=(--username "${YTDLP_USERNAME}")
+  [[ -n "${YTDLP_PASSWORD:-}" ]] && args+=(--password "${YTDLP_PASSWORD}")
+  [[ -n "${YTDLP_NETRC:-}" ]] && args+=(--netrc)
+  [[ -n "${YTDLP_EXTRA_ARGS:-}" ]] && args+=(${YTDLP_EXTRA_ARGS})
+  printf '%s\n' "${args[@]}"
+}
+
+build_ytdlp_pipeline() {
+  local fmt_args=()
+  [[ -n "${SOURCE_FMT}" ]] && fmt_args=(-f "${SOURCE_FMT}")
+  mapfile -t auth < <(ytdlp_auth_args)
+
+  printf '%s\0' "${YTDLP_BIN}" \
+    -o - --no-part --no-live-from-start --no-write-subs --no-embed-subs \
+    --remux-video mkv --merge-output-format mkv --no-audio-multistreams \
+    --geo-bypass --ignore-errors --hls-use-mpegts \
+    "${auth[@]}" \
+    "${fmt_args[@]}" \
+    "${SOURCE_URL}"
+}
+
+# Returns ffmpeg input argv: either direct -i URL or pipe from yt-dlp.
+build_primary_input() {
+  read_rate_flag
+  if [[ "${USE_YTDLP}" == "1" ]]; then
+    require_cmd "${YTDLP_BIN}"
+    printf '%s\0' pipe
+    printf '%s\0' -thread_queue_size "${THREAD_QUEUE_SIZE}" -i pipe:0
+    return
+  fi
+
+  local url="${SOURCE_URL}"
+  if [[ "${url}" == *.jpg || "${url}" == *.jpeg || "${url}" == *.png ]]; then
+    printf '%s\0' direct
+    printf '%s\0' "${RE_FLAG[@]}" -thread_queue_size "${THREAD_QUEUE_SIZE}" \
+      -loop 1 -i "${url}" -f lavfi -i anullsrc=channel_layout=stereo:sample_rate=48000
+    return
+  fi
+
+  printf '%s\0' direct
+  printf '%s\0' "${RE_FLAG[@]}" -thread_queue_size "${THREAD_QUEUE_SIZE}" \
+    -reconnect_on_network_error 1 -reconnect_streamed 1 -reconnect_delay_max 5 \
+    -i "${url}"
+}