michal/jellyfin-plugin-enigma2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Retain selectable audio languages and DVB subtitles through live continuity

Browse Source
This commit is contained in:
michal
2026-10-23 23:50:46 +02:00
parent 9a056b3070
commit bd7395f045
17 changed files with 666 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Jellyfin.Plugin.Enigma2/ContinuityStreams.cs
+58 -33
View File
@@ -53,42 +53,54 @@ public sealed class ContinuityStreams : IDisposable
public string Id = Guid.NewGuid().ToString("N");
public Source Source;
public Dictionary<string, byte[]> Slates;
public DvbSubtitleVersions SubtitleVersions;
public DateTime LastAccess = DateTime.UtcNow;
}
public async Task<MediaSourceInfo> Open(string relayUrl, string language, CancellationToken token)
public async Task<MediaSourceInfo> Open(string relayUrl, string language, CancellationToken token, BroadcastTracks tracks = null, bool useSlates = true)
{
language = StartupMessages.ResolveLanguage(language, null, "en");
var clips = await CachedSlates(language, token);
tracks ??= BroadcastTracks.Fallback(preserveResolution);
if (tracks.Normalize != useSlates) throw new ArgumentException("Track format does not match continuity mode");
var clips = useSlates ? await CachedSlates(language, token, tracks) : null;
var sourceKey = relayUrl + "|" + useSlates + "|" + tracks.Key;
token.ThrowIfCancellationRequested();
Viewer viewer;
lock (gate)
{
ObjectDisposedException.ThrowIf(disposed, this);
if (viewers.Count >= 64) throw new InvalidOperationException("Continuity viewer limit reached");
if (!sources.TryGetValue(relayUrl, out var source))
if (!sources.TryGetValue(sourceKey, out var source))
{
if (sources.Count >= 4) throw new InvalidOperationException("Continuity channel limit reached");
source = new(this, relayUrl); sources.Add(relayUrl, source);
source = new(this, relayUrl, sourceKey, tracks, useSlates); sources.Add(sourceKey, source);
}
viewer = new() { Source = source, Slates = clips }; viewers.Add(viewer.Id, viewer);
viewer = new() { Source = source, Slates = clips, SubtitleVersions = new(tracks) }; viewers.Add(viewer.Id, viewer);
}
return new MediaSourceInfo {
Id = viewer.Id, Path = $"http://127.0.0.1:{port}/DecoderIntegration/Continuity/{viewer.Id}/index.m3u8",
Protocol = MediaProtocol.Http, Container = "hls", IsInfiniteStream = true, SupportsProbing = false,
SupportsDirectPlay = false, SupportsDirectStream = true, SupportsTranscoding = true, BufferMs = 1000,
MediaStreams = new[] {
new MediaStream { Type = MediaStreamType.Video, Index = -1, Codec = "h264", Profile = "Constrained Baseline", Level = preserveResolution ? null : 42,
Width = preserveResolution ? null : Width, Height = preserveResolution ? null : Height, AverageFrameRate = FrameRate, RealFrameRate = FrameRate, IsInterlaced = false, BitRate = preserveResolution ? null : 12000000, PixelFormat = "yuv420p" },
new MediaStream { Type = MediaStreamType.Audio, Index = -1, Codec = "aac", Channels = 2, SampleRate = 48000, BitRate = 128000, IsDefault = true }
}
};
try
{
if (!useSlates) await viewer.Source.Ready.Task.WaitAsync(TimeSpan.FromSeconds(35), token);
return new MediaSourceInfo {
Id = viewer.Id, Path = $"http://127.0.0.1:{port}/DecoderIntegration/Continuity/{viewer.Id}/index.m3u8",
Protocol = MediaProtocol.Http, Container = "hls", IsInfiniteStream = true, SupportsProbing = false,
SupportsDirectPlay = false, SupportsDirectStream = true, SupportsTranscoding = true, BufferMs = 1000,
MediaStreams = tracks.Output,
DefaultAudioStreamIndex = tracks.Output.FirstOrDefault(s => s.Type == MediaStreamType.Audio && s.IsDefault)?.Index
?? tracks.Output.FirstOrDefault(s => s.Type == MediaStreamType.Audio)?.Index,
DefaultSubtitleStreamIndex = tracks.Output.FirstOrDefault(s => s.Type == MediaStreamType.Subtitle && s.IsDefault)?.Index
};
}
catch { Close(viewer.Id); throw; }
}
private async Task<Dictionary<string, byte[]>> CachedSlates(string language, CancellationToken token)
private async Task<Dictionary<string, byte[]>> CachedSlates(string language, CancellationToken token, BroadcastTracks tracks = null)
{
var cached = slates.GetOrAdd(language, lang => new(() => GenerateSlates(lang), LazyThreadSafetyMode.ExecutionAndPublication));
tracks ??= BroadcastTracks.Fallback(preserveResolution);
var key = language + "-" + tracks.Key;
if (slates.Count >= 64)
foreach (var stale in slates.Keys.Where(k => k != key).Take(slates.Count - 63)) slates.TryRemove(stale, out _);
var cached = slates.GetOrAdd(key, _ => new(() => GenerateSlates(language, tracks), LazyThreadSafetyMode.ExecutionAndPublication));
Dictionary<string, byte[]> clips;
try { clips = await cached.Value.WaitAsync(token); }
catch { if (cached.IsValueCreated && cached.Value.IsFaulted) slates.TryRemove(new KeyValuePair<string, Lazy<Task<Dictionary<string, byte[]>>>>(language, cached)); throw; }
catch { if (cached.IsValueCreated && cached.Value.IsFaulted) slates.TryRemove(new KeyValuePair<string, Lazy<Task<Dictionary<string, byte[]>>>>(key, cached)); throw; }
return clips;
}
public async Task<byte[]> SlatePreview(string language, string phase, CancellationToken token)
@@ -103,7 +115,7 @@ public sealed class ContinuityStreams : IDisposable
{
if (!viewers.Remove(id, out var viewer)) return;
if (!viewers.Values.Any(v => ReferenceEquals(v.Source, viewer.Source)))
{ sources.Remove(viewer.Source.Url); viewer.Source.Stop(); }
{ sources.Remove(viewer.Source.Key); viewer.Source.Stop(); }
}
}
public string Playlist(string id)
@@ -122,7 +134,9 @@ public sealed class ContinuityStreams : IDisposable
viewer.LastAccess = DateTime.UtcNow;
var segment = viewer.Source.Timeline.Find(sequence);
if (segment == null) return null;
return TransportStreamClock.Rebase(segment.Slate is { } slate ? viewer.Slates[slate] : segment.Video, segment.Start90k);
var transport = TransportStreamClock.Rebase(segment.Slate is { } slate ? viewer.Slates[slate] : segment.Video, segment.Start90k, alignAudio: viewer.Source.UseSlates);
viewer.SubtitleVersions.Apply(transport, sequence);
return transport;
}
}
public object Status() { lock (gate) return new { Sources = sources.Count, Viewers = viewers.Count, Retrying = sources.Values.Count(s => s.Retrying), ProducerAttempts = Interlocked.Read(ref attempts), LiveSegmentsProduced = Interlocked.Read(ref liveSegments) }; }
@@ -136,7 +150,7 @@ public sealed class ContinuityStreams : IDisposable
lock (gate)
{
foreach (var id in viewers.Values.Where(v => DateTime.UtcNow - v.LastAccess > TimeSpan.FromSeconds(60)).Select(v => v.Id).ToArray()) Close(id);
foreach (var source in sources.Values) source.Timeline.Tick(DateTime.UtcNow, source.Retrying);
foreach (var source in sources.Values) if (source.UseSlates) source.Timeline.Tick(DateTime.UtcNow, source.Retrying);
}
}
}
@@ -164,7 +178,7 @@ public sealed class ContinuityStreams : IDisposable
process.ErrorDataReceived += (_, _) => { }; process.OutputDataReceived += (_, _) => { };
process.BeginErrorReadLine(); process.BeginOutputReadLine(); return process;
}
private async Task<Dictionary<string, byte[]>> GenerateSlates(string language)
private async Task<Dictionary<string, byte[]>> GenerateSlates(string language, BroadcastTracks tracks)
{
await slateGate.WaitAsync(lifetime.Token);
var folder = Path.Combine(root, "slates-" + language + "-" + Guid.NewGuid().ToString("N"));
@@ -180,13 +194,13 @@ public sealed class ContinuityStreams : IDisposable
var filter = $"drawbox=x=(iw-160)/2:y=260:w=160:h=100:color=0x58bfee:t=6,drawtext=fontfile={font}:text=TV:fontsize=42:fontcolor=0x58bfee:x=(w-tw)/2:y=285," +
$"drawtext=fontfile={font}:textfile=message.txt:fontsize=42:fontcolor=white:line_spacing=22:x=(w-tw)/2:y=490";
using var process = Start(new[] { "-f", "lavfi", "-i", $"color=c=0x101723:s={Width}x{Height}:r={FrameRate}", "-f", "lavfi", "-i", "anullsrc=r=48000:cl=stereo", "-vf", filter,
"-map", "0:v:0", "-map", "1:a:0", "-t", "6.72" }.Concat(CodecArguments(slate: true)).Concat(new[] { "-f", "hls", "-hls_time", "2.24", "-hls_list_size", "0", "-hls_segment_filename", "slate-%d.ts", "slate.m3u8" }), folder);
"-map", "0:v:0" }.Concat(Enumerable.Range(0, tracks.AudioCount).SelectMany(_ => new[] { "-map", "1:a:0" })).Concat(new[] { "-t", "6.72" }).Concat(CodecArguments(slate: true)).Concat(tracks.MetadataArguments(audioVideoOnly: true)).Concat(new[] { "-f", "hls", "-hls_time", "2.24", "-hls_list_size", "0", "-hls_segment_filename", "slate-%d.ts", "slate.m3u8" }), folder);
try { await process.WaitForExitAsync(lifetime.Token).WaitAsync(TimeSpan.FromSeconds(20), lifetime.Token); }
finally { if (!process.HasExited) process.Kill(entireProcessTree: true); }
if (process.ExitCode != 0) throw new IOException("FFmpeg slate generation failed");
// Use an interior HLS segment: it has neither AAC encoder priming
// nor the final padding of a newly encoded standalone file.
clips[phase] = await File.ReadAllBytesAsync(Path.Combine(folder, "slate-1.ts"), lifetime.Token);
clips[phase] = BroadcastSlateProgram.Apply(await File.ReadAllBytesAsync(Path.Combine(folder, "slate-1.ts"), lifetime.Token), tracks);
foreach (var file in Directory.GetFiles(folder, "slate*")) File.Delete(file);
}
return clips;
@@ -198,11 +212,14 @@ public sealed class ContinuityStreams : IDisposable
private readonly ContinuityStreams owner;
private readonly CancellationTokenSource lifetime;
private readonly Task task;
public readonly string Url;
public ContinuityTimeline Timeline { get; } = new(DateTime.UtcNow);
public readonly string Url, Key;
public readonly BroadcastTracks Tracks;
public readonly bool UseSlates;
public TaskCompletionSource Ready { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
public ContinuityTimeline Timeline { get; }
public volatile bool Retrying;
public Source(ContinuityStreams owner, string url)
{ this.owner = owner; Url = url; lifetime = CancellationTokenSource.CreateLinkedTokenSource(owner.lifetime.Token); task = Task.Run(Run); }
public Source(ContinuityStreams owner, string url, string key, BroadcastTracks tracks, bool useSlates)
{ this.owner = owner; Url = url; Key = key; Tracks = tracks; UseSlates = useSlates; Timeline = new(DateTime.UtcNow, useSlates); lifetime = CancellationTokenSource.CreateLinkedTokenSource(owner.lifetime.Token); task = Task.Run(Run); }
public void Stop() { try { lifetime.Cancel(); } catch (ObjectDisposedException) { } }
private async Task Run()
{
@@ -216,10 +233,15 @@ public sealed class ContinuityStreams : IDisposable
try
{
Directory.CreateDirectory(folder);
var args = new[] { "-rw_timeout", "10000000", "-analyzeduration", "5000000", "-probesize", "4000000", "-i", Url,
"-map", "0:v:0", "-map", "0:a:0", "-sn", "-dn", "-vf",
owner.preserveResolution ? "bwdif=mode=send_field:parity=auto:deint=interlaced" : $"bwdif=mode=send_field:parity=auto:deint=interlaced,scale=w={Width}:h={Height}:force_original_aspect_ratio=decrease,pad={Width}:{Height}:(ow-iw)/2:(oh-ih)/2,setsar=1" }
.Concat(owner.CodecArguments()).Concat(new[] { "-f", "hls", "-hls_time", "2.24", "-hls_list_size", "6", "-hls_delete_threshold", "8", "-hls_flags", "delete_segments+temp_file+omit_endlist", "-hls_segment_filename", "segment-%09d.ts", "live.m3u8" });
var args = new[] { "-rw_timeout", "10000000", "-analyzeduration", "5000000", "-probesize", "4000000", "-i", Url }
.Concat(Tracks.MapArguments()).Concat(new[] { "-dn" });
if (UseSlates)
args = args.Concat(new[] { "-vf", owner.preserveResolution ? "bwdif=mode=send_field:parity=auto:deint=interlaced" : $"bwdif=mode=send_field:parity=auto:deint=interlaced,scale=w={Width}:h={Height}:force_original_aspect_ratio=decrease,pad={Width}:{Height}:(ow-iw)/2:(oh-ih)/2,setsar=1" })
.Concat(owner.CodecArguments()).Concat(new[] { "-c:s", "copy" });
else args = args.Concat(new[] { "-c", "copy" });
// MPEG-TS segmenter preserves DVB/teletext subtitles, unlike the HLS muxer's WebVTT-only subtitle output.
args = args.Concat(Tracks.MetadataArguments()).Concat(new[] { "-max_interleave_delta", "1000000", "-f", "segment", "-segment_format", "mpegts", "-segment_time", "2.24",
"-segment_list", "live.m3u8", "-segment_list_type", "m3u8", "-segment_list_size", "6", "-segment_list_flags", "+live", "segment-%09d.ts" });
Interlocked.Increment(ref owner.attempts);
using var process = owner.Start(args, folder);
try
@@ -240,7 +262,10 @@ public sealed class ContinuityStreams : IDisposable
var path = Path.Combine(folder, line);
if (!File.Exists(path) || new FileInfo(path).Length > 32 * 1024 * 1024) continue;
Timeline.Live(await File.ReadAllBytesAsync(path, lifetime.Token), epoch, duration, DateTime.UtcNow);
highest = index; lastOutput = DateTime.UtcNow; delivered = true; Retrying = false; failures = 0; Interlocked.Increment(ref owner.liveSegments);
highest = index;
foreach (var obsolete in Directory.EnumerateFiles(folder, "segment-*.ts"))
if (int.TryParse(Path.GetFileNameWithoutExtension(obsolete).AsSpan(8), out var number) && number < highest - 14) File.Delete(obsolete);
lastOutput = DateTime.UtcNow; delivered = true; Ready.TrySetResult(); Retrying = false; failures = 0; Interlocked.Increment(ref owner.liveSegments);
}
}
if ((DateTime.UtcNow - lastOutput).TotalSeconds > (delivered ? 12 : 35)) throw new IOException("No usable encoded segments");