Skip to content

Commit 9a9c906

Browse files
djthorpedjthorpe
committed
Updated decoder
1 parent 6b2b307 commit 9a9c906

File tree

7 files changed

+474
-24
lines changed

7 files changed

+474
-24
lines changed

‎pkg/ffmpeg/decoder.go

Lines changed: 172 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,172 @@
1+
package ffmpeg
2+
3+
import (
4+
"errors"
5+
"fmt"
6+
"io"
7+
"syscall"
8+
9+
// Packages
10+
ff "github.com/mutablelogic/go-media/sys/ffmpeg61"
11+
)
12+
13+
////////////////////////////////////////////////////////////////////////////////
14+
// TYPES
15+
16+
type Decoder struct {
17+
stream int
18+
codec *ff.AVCodecContext
19+
dest *Par // Destination parameters
20+
timeBase ff.AVRational // Timebase for the stream
21+
frame *ff.AVFrame // Destination frame
22+
}
23+
24+
// DecoderFrameFn is a function which is called to send a frame after decoding. It should
25+
// return nil to continue decoding or io.EOF to stop.
26+
type DecoderFrameFn func(int, *Frame) error
27+
28+
////////////////////////////////////////////////////////////////////////////////
29+
// LIFECYCLE
30+
31+
// Create a stream decoder which can decode packets from the input stream
32+
// TODO: resample and resize frames to the destination parameters
33+
func NewDecoder(stream *ff.AVStream, dest *Par, force bool) (*Decoder, error) {
34+
decoder := new(Decoder)
35+
decoder.stream = stream.Id()
36+
decoder.dest = dest
37+
decoder.timeBase = stream.TimeBase()
38+
39+
// Create a codec context for the decoder
40+
codec := ff.AVCodec_find_decoder(stream.CodecPar().CodecID())
41+
if codec == nil {
42+
return nil, fmt.Errorf("failed to find decoder for codec %q", stream.CodecPar().CodecID())
43+
} else if ctx := ff.AVCodec_alloc_context(codec); ctx == nil {
44+
return nil, fmt.Errorf("failed to allocate codec context for codec %q", codec.Name())
45+
} else {
46+
decoder.codec = ctx
47+
}
48+
49+
// Copy codec parameters from input stream to output codec context
50+
if err := ff.AVCodec_parameters_to_context(decoder.codec, stream.CodecPar()); err != nil {
51+
return nil, errors.Join(decoder.Close(), fmt.Errorf("failed to copy codec parameters to decoder context for codec %q", codec.Name()))
52+
}
53+
54+
// Init the decoder
55+
if err := ff.AVCodec_open(decoder.codec, codec, nil); err != nil {
56+
return nil, errors.Join(decoder.Close(), err)
57+
}
58+
59+
// Create a frame for decoder output - before resize/resample
60+
if frame := ff.AVUtil_frame_alloc(); frame == nil {
61+
return nil, errors.Join(decoder.Close(), errors.New("failed to allocate frame"))
62+
} else {
63+
decoder.frame = frame
64+
}
65+
66+
// Return success
67+
return decoder, nil
68+
}
69+
70+
// Close the decoder and free any resources
71+
func (d *Decoder) Close() error {
72+
var result error
73+
74+
// Free the codec context
75+
if d.codec != nil {
76+
ff.AVCodec_free_context(d.codec)
77+
}
78+
79+
// Free destination frame
80+
if d.frame != nil {
81+
ff.AVUtil_frame_free(d.frame)
82+
}
83+
84+
// Return any errors
85+
return result
86+
}
87+
88+
////////////////////////////////////////////////////////////////////////////////
89+
// PUBLIC METHODS
90+
91+
func (d *Decoder) decode(packet *ff.AVPacket, fn DecoderFrameFn) error {
92+
if fn == nil {
93+
return errors.New("DecoderFrameFn is nil")
94+
}
95+
96+
//if demuxfn != nil {
97+
// Send the packet (or a nil to flush) to the user defined packet function
98+
// return demuxfn(newPacket(packet, d.stream, d.codec.Codec().Type(), d.timeBase))
99+
//}
100+
101+
// Submit the packet to the decoder (nil packet will flush the decoder)
102+
if err := ff.AVCodec_send_packet(d.codec, packet); err != nil {
103+
return err
104+
}
105+
106+
// get all the available frames from the decoder
107+
var result error
108+
for {
109+
// End early if we've received an EOF
110+
if result != nil {
111+
break
112+
}
113+
114+
// Receive the next frame from the decoder
115+
if err := ff.AVCodec_receive_frame(d.codec, d.frame); errors.Is(err, syscall.EAGAIN) || errors.Is(err, io.EOF) {
116+
// Finished decoding packet or EOF
117+
break
118+
} else if err != nil {
119+
return err
120+
}
121+
122+
// Resample or resize the frame, then pass to the frame function
123+
//frame, err := d.re(d.frame)
124+
//if err != nil {
125+
// return err
126+
//}
127+
128+
// Copy over the timebase and ptr from the stream
129+
d.frame.SetTimeBase(d.timeBase)
130+
d.frame.SetPts(d.frame.Pts())
131+
132+
// Pass back to the caller
133+
if err := fn(d.stream, (*Frame)(d.frame)); errors.Is(err, io.EOF) {
134+
// End early, return EOF
135+
result = io.EOF
136+
} else if err != nil {
137+
return err
138+
}
139+
140+
// Re-allocate frames for next iteration
141+
ff.AVUtil_frame_unref(d.frame)
142+
// ff.AVUtil_frame_unref(d.reframe)
143+
}
144+
145+
// Flush the resizer or resampler if we haven't received an EOF
146+
/*
147+
if result == nil {
148+
finished := false
149+
for {
150+
if finished {
151+
break
152+
}
153+
if frame, err := d.reflush(d.frame); err != nil {
154+
return err
155+
} else if frame == nil {
156+
finished = true
157+
} else if err := framefn(newFrame(frame)); errors.Is(err, io.EOF) {
158+
finished = true
159+
} else if err != nil {
160+
return err
161+
}
162+
163+
// Re-allocate frames for next iteration
164+
ff.AVUtil_frame_unref(d.frame)
165+
ff.AVUtil_frame_unref(d.reframe)
166+
}
167+
}
168+
*/
169+
170+
// Return success or EOF
171+
return result
172+
}

‎pkg/ffmpeg/frame.go

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@ import (
1212
// TYPES
1313

1414
type Frame ff.AVFrame
15-
type Type ff.AVMediaType
1615

1716
///////////////////////////////////////////////////////////////////////////////
1817
// GLOBALS
@@ -21,14 +20,6 @@ const (
2120
PTS_UNDEFINED = ff.AV_NOPTS_VALUE
2221
)
2322

24-
const (
25-
UNKNOWN Type = Type(ff.AVMEDIA_TYPE_UNKNOWN)
26-
VIDEO Type = Type(ff.AVMEDIA_TYPE_VIDEO)
27-
AUDIO Type = Type(ff.AVMEDIA_TYPE_AUDIO)
28-
DATA Type = Type(ff.AVMEDIA_TYPE_DATA)
29-
SUBTITLE Type = Type(ff.AVMEDIA_TYPE_SUBTITLE)
30-
)
31-
3223
///////////////////////////////////////////////////////////////////////////////
3324
// LIFECYCLE
3425

‎pkg/ffmpeg/reader.go

Lines changed: 121 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,9 @@ import (
1111

1212
// Packages
1313
ff "github.com/mutablelogic/go-media/sys/ffmpeg61"
14+
15+
// Namespace imports
16+
. "github.com/djthorpe/go-errors"
1417
)
1518

1619
////////////////////////////////////////////////////////////////////////////////
@@ -20,6 +23,7 @@ import (
2023
type Reader struct {
2124
input *ff.AVFormatContext
2225
avio *ff.AVIOContextEx
26+
force bool
2327
}
2428

2529
type reader_callback struct {
@@ -105,14 +109,17 @@ func NewReader(r io.Reader, opt ...Opt) (*Reader, error) {
105109
return reader.open(options)
106110
}
107111

108-
func (r *Reader) open(_ *opts) (*Reader, error) {
112+
func (r *Reader) open(options *opts) (*Reader, error) {
109113
// Find stream information
110114
if err := ff.AVFormat_find_stream_info(r.input, nil); err != nil {
111115
ff.AVFormat_free_context(r.input)
112116
ff.AVFormat_avio_context_free(r.avio)
113117
return nil, err
114118
}
115119

120+
// Set force flag
121+
r.force = options.force
122+
116123
// Return success
117124
return r, nil
118125
}
@@ -185,13 +192,122 @@ func (r *Reader) Metadata(keys ...string) []*Metadata {
185192
return result
186193
}
187194

188-
// TODO Decode the media stream into packets and frames
189-
func (r *Reader) Decode(ctx context.Context, fn DecoderMapFunc) error {
190-
return errors.New("not implemented yet")
195+
// Decode the media stream into frames. The decodefn is called for each
196+
// frame decoded from the stream. The map function is called for each stream
197+
// and should return the parameters for the destination frame. If the map
198+
// function returns nil, then the stream is ignored.
199+
//
200+
// The decoding can be interrupted by cancelling the context, or by the decodefn
201+
// returning an error or io.EOF. The latter will end the decoding process early but
202+
// will not return an error.
203+
func (r *Reader) Decode(ctx context.Context, decodefn DecoderFrameFn, mapfn DecoderMapFunc) error {
204+
decoders := make(map[int]*Decoder, r.input.NumStreams())
205+
206+
// Standard decoder map function copies all streams
207+
if mapfn == nil {
208+
mapfn = func(_ int, par *Par) (*Par, error) {
209+
return par, nil
210+
}
211+
}
212+
213+
// Create a decoder for each stream
214+
// The decoder map function should be returning the parameters for the
215+
// destination frame.
216+
var result error
217+
for _, stream := range r.input.Streams() {
218+
stream_index := stream.Index()
219+
220+
// Get decoder parameters and map to a decoder
221+
par, err := mapfn(stream.Id(), &Par{
222+
AVCodecParameters: *stream.CodecPar(),
223+
})
224+
if err != nil {
225+
result = errors.Join(result, err)
226+
} else if par == nil {
227+
continue
228+
} else if decoder, err := NewDecoder(stream, par, r.force); err != nil {
229+
result = errors.Join(result, err)
230+
} else if _, exists := decoders[stream_index]; exists {
231+
result = errors.Join(result, ErrDuplicateEntry.Withf("stream index %d", stream_index))
232+
} else {
233+
decoders[stream_index] = decoder
234+
}
235+
}
236+
237+
// Check to see if we have to do something
238+
if len(decoders) == 0 {
239+
result = errors.Join(result, ErrBadParameter.With("no streams to decode"))
240+
}
241+
242+
// Now we have a map of decoders, we can start decoding
243+
if result == nil {
244+
result = r.decode(ctx, decoders, decodefn)
245+
}
246+
247+
// Release resources
248+
for _, decoder := range decoders {
249+
if err := decoder.Close(); err != nil {
250+
result = errors.Join(result, err)
251+
}
252+
}
253+
254+
// Return any errors
255+
return result
256+
}
257+
258+
////////////////////////////////////////////////////////////////////////////////
259+
// PRIVATE METHODS - DECODE
260+
261+
func (r *Reader) decode(ctx context.Context, decoders map[int]*Decoder, fn DecoderFrameFn) error {
262+
// Allocate a packet
263+
packet := ff.AVCodec_packet_alloc()
264+
if packet == nil {
265+
return errors.New("failed to allocate packet")
266+
}
267+
defer ff.AVCodec_packet_free(packet)
268+
269+
// Read packets
270+
FOR_LOOP:
271+
for {
272+
select {
273+
case <-ctx.Done():
274+
break FOR_LOOP
275+
default:
276+
if err := ff.AVFormat_read_frame(r.input, packet); errors.Is(err, io.EOF) {
277+
break FOR_LOOP
278+
} else if err != nil {
279+
return err
280+
}
281+
stream_index := packet.StreamIndex()
282+
if decoder := decoders[stream_index]; decoder != nil {
283+
if err := decoder.decode(packet, fn); errors.Is(err, io.EOF) {
284+
break FOR_LOOP
285+
} else if err != nil {
286+
return err
287+
}
288+
}
289+
}
290+
291+
// Unreference the packet
292+
ff.AVCodec_packet_unref(packet)
293+
}
294+
295+
// Flush the decoders
296+
for _, decoder := range decoders {
297+
if err := decoder.decode(nil, fn); errors.Is(err, io.EOF) {
298+
// no-op
299+
} else if err != nil {
300+
return err
301+
}
302+
}
303+
304+
// Return the context error - will be cancelled, perhaps, or nil if the
305+
// demuxer finished successfully without cancellation
306+
return ctx.Err()
191307
}
192308

193309
////////////////////////////////////////////////////////////////////////////////
194-
// PRIVATE METHODS
310+
// PRIVATE METHODS - CALLBACK
195311

196312
func (r *reader_callback) Reader(buf []byte) int {
197313
n, err := r.r.Read(buf)

0 commit comments

Comments
 (0)