?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
Commentary tracks: Remove support for old Opus format (use OggOpus)
[lsnes.git] / src / core / inthread.cpp
blob681af3e0a8ea346e83b046c8a49d61e62ec81e87
1 #include <lsnes.hpp>
2 #include <cstdint>
3 #ifdef WITH_OPUS_CODEC
4 #include "library/filesys.hpp"
5 #include "library/minmax.hpp"
6 #include "library/workthread.hpp"
7 #include "library/serialization.hpp"
8 #include "library/string.hpp"
9 #include "library/ogg.hpp"
10 #include "library/opus.hpp"
11 #include "core/audioapi.hpp"
12 #include "core/command.hpp"
13 #include "core/dispatch.hpp"
14 #include "core/framerate.hpp"
15 #include "core/inthread.hpp"
16 #include "core/keymapper.hpp"
17 #include "core/settings.hpp"
18 #include "core/misc.hpp"
19 #include <cmath>
20 #include <list>
21 #include <iostream>
22 #include <fstream>
23 #include <cstring>
24 #include <unistd.h>
25 #include <sys/time.h>
26 #include <zlib.h>
27
28 //Farther than this, packets can be fastskipped.
29 #define OPUS_CONVERGE_MAX 5760
30 //Maximum size of PCM output for one packet.
31 #define OPUS_MAX_OUT 5760
32 //Output block size.
33 #define OUTPUT_BLOCK 1440
34 //Main sampling rate.
35 #define OPUS_SAMPLERATE 48000
36 //Opus block size
37 #define OPUS_BLOCK_SIZE 960
38 //Threshold for decoding additional block
39 #define BLOCK_THRESHOLD 1200
40 //Maximum output block size.
41 #define OUTPUT_SIZE (BLOCK_THRESHOLD + OUTPUT_BLOCK)
42 //Amount of microseconds per interation.
43 #define ITERATION_TIME 15000
44 //Opus bitrate to use.
45 #define OPUS_BITRATE 48000
46 //Opus min bitrate to use.
47 #define OPUS_MIN_BITRATE 8000
48 //Opus max bitrate to use.
49 #define OPUS_MAX_BITRATE 255000
50 //Ogg Opus granule rate.
51 #define OGGOPUS_GRANULERATE 48000
52 //Record buffer size threshold divider.
53 #define REC_THRESHOLD_DIV 40
54 //Playback buffer size threshold divider.
55 #define PLAY_THRESHOLD_DIV 30
56 //Special granule position: None.
57 #define GRANULEPOS_NONE 0xFFFFFFFFFFFFFFFFULL
58
59 namespace
60 {
61 class opus_playback_stream;
62 class opus_stream;
63 class stream_collection;
64
65 setting_var<setting_var_model_int<OPUS_MIN_BITRATE,OPUS_MAX_BITRATE>> opus_bitrate(lsnes_vset, "opus-bitrate",
66 "commentary‣Bitrate", OPUS_BITRATE);
67 setting_var<setting_var_model_int<OPUS_MIN_BITRATE,OPUS_MAX_BITRATE>> opus_max_bitrate(lsnes_vset,
68 "opus-max-bitrate", "commentary‣Max bitrate", OPUS_MAX_BITRATE);
69
70 //Recording active flag.
71 volatile bool active_flag = false;
72 //Last seen frame number.
73 uint64_t last_frame_number = 0;
74 //Last seen rate.
75 double last_rate = 0;
76 //Mutex protecting current_time and time_jump.
77 mutex_class time_mutex;
78 //The current time.
79 uint64_t current_time;
80 //Time jump flag. Set if time jump is detected.
81 //If time jump is detected, all current playing streams are stopped, stream locks are cleared and
82 //apropriate streams are restarted. If time jump is false, all unlocked streams coming into range
83 //are started.
84 bool time_jump;
85 //Lock protecting active_playback_streams.
86 mutex_class active_playback_streams_lock;
87 //List of streams currently playing.
88 std::list<opus_playback_stream*> active_playback_streams;
89 //The collection of streams.
90 stream_collection* current_collection;
91 //Lock protecting current collection.
92 mutex_class current_collection_lock;
93
94 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
95 //Bitrate tracker.
96 struct bitrate_tracker
97 {
98 bitrate_tracker() throw();
99 void reset() throw();
100 double get_min() throw();
101 double get_avg() throw();
102 double get_max() throw();
103 double get_length() throw();
104 uint64_t get_bytes() throw();
105 uint64_t get_blocks() throw();
106 void submit(uint32_t bytes, uint32_t samples) throw();
107 private:
108 uint64_t blocks;
109 uint64_t samples;
110 uint64_t bytes;
111 uint32_t minrate;
112 uint32_t maxrate;
113 };
114
115 bitrate_tracker::bitrate_tracker() throw()
116 {
117 reset();
118 }
119
120 void bitrate_tracker::reset() throw()
121 {
122 blocks = 0;
123 samples = 0;
124 bytes = 0;
125 minrate = std::numeric_limits<uint32_t>::max();
126 maxrate = 0;
127 }
128
129 double bitrate_tracker::get_min() throw()
130 {
131 return blocks ? minrate / 1000.0 : 0.0;
132 }
133
134 double bitrate_tracker::get_avg() throw()
135 {
136 return samples ? bytes / (125.0 * samples / OPUS_SAMPLERATE) : 0.0;
137 }
138
139 double bitrate_tracker::get_max() throw()
140 {
141 return blocks ? maxrate / 1000.0 : 0.0;
142 }
143
144 double bitrate_tracker::get_length() throw()
145 {
146 return 1.0 * samples / OPUS_SAMPLERATE;
147 }
148
149 uint64_t bitrate_tracker::get_bytes() throw()
150 {
151 return bytes;
152 }
153
154 uint64_t bitrate_tracker::get_blocks() throw()
155 {
156 return blocks;
157 }
158
159 void bitrate_tracker::submit(uint32_t _bytes, uint32_t _samples) throw()
160 {
161 blocks++;
162 samples += _samples;
163 bytes += _bytes;
164 uint32_t irate = _bytes * 8 * OPUS_SAMPLERATE / OPUS_BLOCK_SIZE;
165 minrate = min(minrate, irate);
166 maxrate = max(maxrate, irate);
167 }
168
169 std::ostream& operator<<(std::ostream& s, bitrate_tracker& t)
170 {
171 s << t.get_bytes() << " bytes for " << t.get_length() << "s (" << t.get_blocks() << " blocks)"
172 << std::endl << "Bitrate (kbps): min: " << t.get_min() << " avg: " << t.get_avg() << " max:"
173 << t.get_max() << std::endl;
174 return s;
175 }
176
177 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
178 //Information about individual opus packet in stream.
179 struct opus_packetinfo
180 {
181 //Length is in units of 1/400th of a second.
182 opus_packetinfo(uint16_t datasize, uint8_t length, uint64_t offset)
183 {
184 descriptor = (offset & 0xFFFFFFFFFFULL) | (static_cast<uint64_t>(length) << 40) |
185 (static_cast<uint64_t>(datasize) << 48);
186 }
187 //Get the data size of the packet.
188 uint16_t size() { return descriptor >> 48; }
189 //Calculate the length of packet in samples.
190 uint16_t length() { return 120 * ((descriptor >> 40) & 0xFF); }
191 //Calculate the true offset.
192 uint64_t offset() { return descriptor & 0xFFFFFFFFFFULL; }
193 //Read the packet.
194 //Can throw.
195 std::vector<unsigned char> packet(filesystem_ref from_sys);
196 private:
197 uint64_t descriptor;
198 };
199
200 std::vector<unsigned char> opus_packetinfo::packet(filesystem_ref from_sys)
201 {
202 std::vector<unsigned char> ret;
203 uint64_t off = offset();
204 uint32_t sz = size();
205 uint32_t cluster = off / CLUSTER_SIZE;
206 uint32_t coff = off % CLUSTER_SIZE;
207 ret.resize(sz);
208 size_t r = from_sys.read_data(cluster, coff, &ret[0], sz);
209 if(r != sz)
210 throw std::runtime_error("Incomplete read");
211 return ret;
212 }
213
214 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
215 //Information about opus stream.
216 struct opus_stream
217 {
218 //Create new empty stream with specified base time.
219 opus_stream(uint64_t base, filesystem_ref filesys);
220 //Read stream with specified base time and specified start clusters.
221 //Can throw.
222 opus_stream(uint64_t base, filesystem_ref filesys, uint32_t ctrl_cluster, uint32_t data_cluster);
223 //Import a stream with specified base time.
224 //Can throw.
225 opus_stream(uint64_t base, filesystem_ref filesys, std::ifstream& data,
226 external_stream_format extfmt);
227 //Delete this stream (also puts a ref)
228 void delete_stream() { deleting = true; put_ref(); }
229 //Export a stream.
230 //Can throw.
231 void export_stream(std::ofstream& data, external_stream_format extfmt);
232 //Get length of specified packet in samples.
233 uint16_t packet_length(uint32_t seqno)
234 {
235 return (seqno < packets.size()) ? packets[seqno].length() : 0;
236 }
237 //Get data of specified packet.
238 //Can throw.
239 std::vector<unsigned char> packet(uint32_t seqno)
240 {
241 return (seqno < packets.size()) ? packets[seqno].packet(fs) : std::vector<unsigned char>();
242 }
243 //Get base time in samples for stream.
244 uint64_t timebase() { return s_timebase; }
245 //Set base time in samples for stream.
246 void timebase(uint64_t ts) { s_timebase = ts; }
247 //Get length of stream in samples.
248 uint64_t length()
249 {
250 if(pregap_length + postgap_length > total_len)
251 return 0;
252 else
253 return total_len - pregap_length - postgap_length;
254 }
255 //Set the pregap length.
256 void set_pregap(uint32_t p) { pregap_length = p; }
257 //Get the pregap length.
258 uint32_t get_pregap() { return pregap_length; }
259 //Set the postgap length.
260 void set_potsgap(uint32_t p) { postgap_length = p; }
261 //Get the postgap length.
262 uint32_t get_postgap() { return postgap_length; }
263 //Set gain.
264 void set_gain(int16_t g) { gain = g; }
265 //Get gain.
266 int16_t get_gain() { return gain; }
267 //Get linear gain.
268 float get_gain_linear() { return pow(10, gain / 20); }
269 //Get number of packets in stream.
270 uint32_t blocks() { return packets.size(); }
271 //Is this stream locked?
272 bool islocked() { return locked; }
273 //Lock a stream.
274 void lock() { locked = true; }
275 //Unlock a stream.
276 void unlock() { locked = false; }
277 //Increment reference count.
278 void get_ref() { umutex_class m(reflock); refcount++; }
279 //Decrement reference count, destroying object if it hits zero.
280 void put_ref() { umutex_class m(reflock); refcount--; if(!refcount) destroy(); }
281 //Add new packet into stream.
282 //Not safe to call simultaneously with packet_length() or packet().
283 //Can throw.
284 void write(uint8_t len, const unsigned char* payload, size_t payload_len);
285 //Write stream trailer.
286 void write_trailier();
287 //Get clusters.
288 std::pair<uint32_t, uint32_t> get_clusters() { return std::make_pair(ctrl_cluster, data_cluster); }
289 private:
290 void export_stream_sox(std::ofstream& data);
291 void export_stream_oggopus(std::ofstream& data);
292 void import_stream_sox(std::ifstream& data);
293 void import_stream_oggopus(std::ifstream& data);
294
295 opus_stream(const opus_stream&);
296 opus_stream& operator=(const opus_stream&);
297 void destroy();
298 filesystem_ref fs;
299 std::vector<opus_packetinfo> packets;
300 uint64_t total_len;
301 uint64_t s_timebase;
302 uint32_t next_cluster;
303 uint32_t next_offset;
304 uint32_t next_mcluster;
305 uint32_t next_moffset;
306 uint32_t ctrl_cluster;
307 uint32_t data_cluster;
308 uint32_t pregap_length;
309 uint32_t postgap_length;
310 int16_t gain;
311 bool locked;
312 mutex_class reflock;
313 unsigned refcount;
314 bool deleting;
315 };
316
317 opus_stream::opus_stream(uint64_t base, filesystem_ref filesys)
318 : fs(filesys)
319 {
320 refcount = 1;
321 deleting = false;
322 total_len = 0;
323 s_timebase = base;
324 locked = false;
325 next_cluster = 0;
326 next_mcluster = 0;
327 next_offset = 0;
328 next_moffset = 0;
329 ctrl_cluster = 0;
330 data_cluster = 0;
331 pregap_length = 0;
332 postgap_length = 0;
333 gain = 0;
334 }
335
336 opus_stream::opus_stream(uint64_t base, filesystem_ref filesys, uint32_t _ctrl_cluster,
337 uint32_t _data_cluster)
338 : fs(filesys)
339 {
340 refcount = 1;
341 deleting = false;
342 total_len = 0;
343 s_timebase = base;
344 locked = false;
345 next_cluster = data_cluster = _data_cluster;
346 next_mcluster = ctrl_cluster = _ctrl_cluster;
347 next_offset = 0;
348 next_moffset = 0;
349 pregap_length = 0;
350 postgap_length = 0;
351 gain = 0;
352 //Read the data buffers.
353 char buf[CLUSTER_SIZE];
354 uint32_t last_cluster_seen = next_mcluster;
355 uint64_t total_size = 0;
356 uint64_t total_frames = 0;
357 bool trailers = false;
358 bool saved_pointer_valid = false;
359 uint32_t saved_next_mcluster = 0;
360 uint32_t saved_next_moffset = 0;
361 while(true) {
362 last_cluster_seen = next_mcluster;
363 size_t r = fs.read_data(next_mcluster, next_moffset, buf, CLUSTER_SIZE);
364 if(!r) {
365 //The stream ends here.
366 break;
367 }
368 //Find the first unused entry if any.
369 for(unsigned i = 0; i < CLUSTER_SIZE; i += 4)
370 if(!buf[i + 3] || trailers) {
371 //This entry is unused. If the next entry is also unused, that is the end.
372 //Otherwise, there might be stream trailers.
373 if(trailers && !buf[i + 3]) {
374 goto out_parsing; //Ends for real.
375 }
376 if(!trailers) {
377 //Set the trailer flag and continue parsing.
378 //The saved offset must be placed here.
379 saved_next_mcluster = last_cluster_seen;
380 saved_next_moffset = i;
381 saved_pointer_valid = true;
382 trailers = true;
383 continue;
384 }
385 //This is a trailer entry.
386 if(buf[i + 3] == 2) {
387 //Pregap.
388 pregap_length = read32ube(buf + i) >> 8;
389 } else if(buf[i + 3] == 3) {
390 //Postgap.
391 postgap_length = read32ube(buf + i) >> 8;
392 } else if(buf[i + 3] == 4) {
393 //Gain.
394 gain = read16sbe(buf + i);
395 }
396 } else {
397 uint16_t psize = read16ube(buf + i);
398 uint8_t plen = read8ube(buf + i + 2);
399 total_size += psize;
400 total_len += 120 * plen;
401 opus_packetinfo p(psize, plen, 1ULL * next_cluster * CLUSTER_SIZE +
402 next_offset);
403 size_t r2 = fs.skip_data(next_cluster, next_offset, psize);
404 if(r2 < psize)
405 throw std::runtime_error("Incomplete data stream");
406 packets.push_back(p);
407 total_frames++;
408 }
409 }
410 out_parsing:
411 //If saved pointer is valid, restore to that.
412 if(saved_pointer_valid) {
413 next_mcluster = saved_next_mcluster;
414 next_moffset = saved_next_moffset;
415 }
416 }
417
418 opus_stream::opus_stream(uint64_t base, filesystem_ref filesys, std::ifstream& data,
419 external_stream_format extfmt)
420 : fs(filesys)
421 {
422 refcount = 1;
423 deleting = false;
424 total_len = 0;
425 s_timebase = base;
426 locked = false;
427 next_cluster = 0;
428 next_mcluster = 0;
429 next_offset = 0;
430 next_moffset = 0;
431 ctrl_cluster = 0;
432 data_cluster = 0;
433 pregap_length = 0;
434 postgap_length = 0;
435 gain = 0;
436 if(extfmt == EXTFMT_OGGOPUS)
437 import_stream_oggopus(data);
438 else if(extfmt == EXTFMT_SOX)
439 import_stream_sox(data);
440 }
441
442 void opus_stream::import_stream_oggopus(std::ifstream& data)
443 {
444 ogg_stream_reader_iostreams reader(data);
445 reader.set_errors_to(messages);
446 ogg_page page;
447 ogg_demuxer d(messages);
448 int state = 0;
449 postgap_length = 0;
450 uint64_t datalen = 0;
451 uint64_t last_datalen = 0;
452 uint64_t last_granulepos = 0;
453 try {
454 while(reader.get_page(page)) {
455 bool c = d.page_in(page);
456 struct oggopus_header h;
457 struct oggopus_tags t;
458 switch(state) {
459 case 0: //Not locked.
460 h = parse_oggopus_header(page);
461 if(h.streams != 1)
462 throw std::runtime_error("Multistream OggOpus streams are not "
463 "supported");
464 state = 1; //Expecting comment.
465 pregap_length = h.preskip;
466 gain = h.gain;
467 while(d.wants_packet_out())
468 d.discard_packet();
469 break;
470 case 1: //Expecting comment.
471 t = parse_oggopus_tags(page);
472 state = 2; //Data page.
473 if(page.get_eos())
474 throw std::runtime_error("Empty OggOpus stream");
475 //We don't do anything with this.
476 while(d.wants_packet_out())
477 d.discard_packet();
478 break;
479 case 2: //Data page.
480 case 3: //Data page.
481 while(d.wants_packet_out()) {
482 ogg_packet p;
483 d.packet_out(p);
484 std::vector<uint8_t> pkt = p.get_vector();
485 uint8_t tcnt = opus_packet_tick_count(&pkt[0], pkt.size());
486 if(tcnt) {
487 write(tcnt, &pkt[0], pkt.size());
488 datalen += tcnt * 120;
489 }
490 if(p.get_last_page()) {
491 uint64_t samples = p.get_granulepos() - last_granulepos;
492 if(samples > p.get_granulepos())
493 samples = 0;
494 uint64_t rsamples = datalen - last_datalen;
495 if((samples > rsamples && state == 3) || (samples <
496 rsamples && !p.get_on_eos_page()))
497 messages << "Warning: Granulepos says there are "
498 << samples << " samples, found " << rsamples
499 << std::endl;
500 last_datalen = datalen;
501 last_granulepos = p.get_granulepos();
502 if(p.get_on_eos_page()) {
503 if(samples < rsamples)
504 postgap_length = rsamples - samples;
505 state = 4;
506 goto out;
507 }
508 }
509 }
510 state = 3;
511 break;
512 }
513 }
514 out:
515 if(state == 0)
516 throw std::runtime_error("No OggOpus stream found");
517 if(state == 1)
518 throw std::runtime_error("Oggopus stream missing required tags page");
519 if(state == 2 || state == 3)
520 messages << "Warning: Incomplete Oggopus stream." << std::endl;
521 if(datalen <= pregap_length)
522 throw std::runtime_error("Stream too short (entiere pregap not present)");
523 write_trailier();
524 } catch(...) {
525 if(ctrl_cluster) fs.free_cluster_chain(ctrl_cluster);
526 if(data_cluster) fs.free_cluster_chain(data_cluster);
527 throw;
528 }
529 }
530
531 void opus_stream::import_stream_sox(std::ifstream& data)
532 {
533 bitrate_tracker brtrack;
534 int err;
535 unsigned char tmpi[65536];
536 float tmp[OPUS_MAX_OUT];
537 char header[260];
538 data.read(header, 32);
539 if(!data)
540 throw std::runtime_error("Can't read .sox header");
541 if(read32ule(header + 0) != 0x586F532EULL)
542 throw std::runtime_error("Bad .sox header magic");
543 if(read8ube(header + 4) > 28)
544 data.read(header + 32, read8ube(header + 4) - 28);
545 if(!data)
546 throw std::runtime_error("Can't read .sox header");
547 if(read64ule(header + 16) != 4676829883349860352ULL)
548 throw std::runtime_error("Bad .sox sampling rate");
549 if(read32ule(header + 24) != 1)
550 throw std::runtime_error("Only mono streams are supported");
551 uint64_t samples = read64ule(header + 8);
552 opus::encoder enc(opus::samplerate::r48k, false, opus::application::voice);
553 enc.ctl(opus::bitrate(opus_bitrate.get()));
554 int32_t pregap = enc.ctl(opus::lookahead);
555 pregap_length = pregap;
556 for(uint64_t i = 0; i < samples + pregap; i += OPUS_BLOCK_SIZE) {
557 size_t bs = OPUS_BLOCK_SIZE;
558 if(i + bs > samples + pregap)
559 bs = samples + pregap - i;
560 //We have to read zero bytes after the end of stream.
561 size_t readable = bs;
562 if(readable + i > samples)
563 readable = max(samples, i) - i;
564 if(readable > 0)
565 data.read(reinterpret_cast<char*>(tmpi), 4 * readable);
566 if(readable < bs)
567 memset(tmpi + 4 * readable, 0, 4 * (bs - readable));
568 if(!data) {
569 if(ctrl_cluster) fs.free_cluster_chain(ctrl_cluster);
570 if(data_cluster) fs.free_cluster_chain(data_cluster);
571 throw std::runtime_error("Can't read .sox data");
572 }
573 for(size_t j = 0; j < bs; j++)
574 tmp[j] = static_cast<float>(read32sle(tmpi + 4 * j)) / 268435456;
575 if(bs < OPUS_BLOCK_SIZE)
576 postgap_length = OPUS_BLOCK_SIZE - bs;
577 for(size_t j = bs; j < OPUS_BLOCK_SIZE; j++)
578 tmp[j] = 0;
579 try {
580 const size_t opus_out_max2 = opus_max_bitrate.get() * OPUS_BLOCK_SIZE / 384000;
581 size_t r = enc.encode(tmp, OPUS_BLOCK_SIZE, tmpi, opus_out_max2);
582 write(OPUS_BLOCK_SIZE / 120, tmpi, r);
583 brtrack.submit(r, bs);
584 } catch(std::exception& e) {
585 if(ctrl_cluster) fs.free_cluster_chain(ctrl_cluster);
586 if(data_cluster) fs.free_cluster_chain(data_cluster);
587 (stringfmt() << "Error encoding opus packet: " << e.what()).throwex();
588 }
589 }
590 messages << "Imported stream: " << brtrack;
591 try {
592 write_trailier();
593 } catch(...) {
594 if(ctrl_cluster) fs.free_cluster_chain(ctrl_cluster);
595 if(data_cluster) fs.free_cluster_chain(data_cluster);
596 throw;
597 }
598 }
599
600 void opus_stream::destroy()
601 {
602 if(deleting) {
603 //We catch the errors and print em, because otherwise put_ref could throw, which would
604 //be too much.
605 try {
606 fs.free_cluster_chain(ctrl_cluster);
607 } catch(std::exception& e) {
608 messages << "Failed to delete stream control file: " << e.what();
609 }
610 try {
611 fs.free_cluster_chain(data_cluster);
612 } catch(std::exception& e) {
613 messages << "Failed to delete stream data file: " << e.what();
614 }
615 }
616 delete this;
617 }
618
619 void opus_stream::export_stream_oggopus(std::ofstream& data)
620 {
621 oggopus_header header;
622 oggopus_tags tags;
623 ogg_stream_writer_iostreams writer(data);
624 unsigned stream_id = 1;
625 uint64_t true_granule = 0;
626 uint32_t seq = 2;
627 //Headers / Tags.
628 header.version = 1;
629 header.channels = 1;
630 header.preskip = pregap_length;
631 header.rate = OPUS_SAMPLERATE;
632 header.gain = 0;
633 header.map_family = 0;
634 header.streams = 1;
635 header.coupled = 0;
636 header.chanmap[0] = 0;
637 memset(header.chanmap + 1, 255, 254);
638 tags.vendor = "lsnes rr" + lsnes_version;
639 tags.comments.push_back((stringfmt() << "LSNES_STREAM_TS=" << s_timebase).str());
640 struct ogg_page hpage = serialize_oggopus_header(header);
641 struct ogg_page tpage = serialize_oggopus_tags(tags);
642 struct ogg_page ppage;
643 hpage.set_stream(stream_id);
644 tpage.set_stream(stream_id);
645 //Empty stream?
646 if(!packets.size())
647 tpage.set_eos(true);
648 writer.put_page(hpage);
649 writer.put_page(tpage);
650 ogg_muxer mux(stream_id, 2);
651 for(size_t i = 0; i < packets.size(); i++) {
652 std::vector<unsigned char> p;
653 try {
654 p = packet(i);
655 } catch(std::exception& e) {
656 (stringfmt() << "Error reading opus packet: " << e.what()).throwex();
657 }
658 if(!p.size())
659 (stringfmt() << "Empty Opus packet is not valid").throwex();
660 uint32_t frames = opus::packet_get_nb_frames(&p[0], p.size());
661 uint32_t samples_pf = opus::packet_get_samples_per_frame(&p[0], opus::samplerate::r48k);
662 uint32_t samples = frames * samples_pf;
663 if(i + 1 < packets.size())
664 true_granule += samples;
665 else
666 true_granule = max(true_granule, true_granule + samples - postgap_length);
667 if(!mux.wants_packet_in() || !mux.packet_fits(p.size()))
668 while(mux.has_page_out()) {
669 mux.page_out(ppage);
670 writer.put_page(ppage);
671 }
672 mux.packet_in(p, true_granule);
673 }
674 mux.signal_eos();
675 while(mux.has_page_out()) {
676 mux.page_out(ppage);
677 writer.put_page(ppage);
678 }
679 }
680
681 void opus_stream::export_stream_sox(std::ofstream& data)
682 {
683 int err;
684 opus::decoder dec(opus::samplerate::r48k, false);
685 std::vector<unsigned char> p;
686 float tmp[OPUS_MAX_OUT];
687 char header[32];
688 write64ule(header, 0x1C586F532EULL); //Magic and header size.
689 write64ule(header + 16, 4676829883349860352ULL); //Sampling rate.
690 write32ule(header + 24, 1);
691 uint64_t tlen = 0;
692 uint32_t lookahead_thrown = 0;
693 data.write(header, 32);
694 if(!data)
695 throw std::runtime_error("Error writing PCM data.");
696 float lgain = get_gain_linear();
697 for(size_t i = 0; i < packets.size(); i++) {
698 char blank[4] = {0, 0, 0, 0};
699 try {
700 uint32_t pregap_throw = 0;
701 uint32_t postgap_throw = 0;
702 std::vector<unsigned char> p = packet(i);
703 uint32_t len = packet_length(i);
704 size_t r = dec.decode(&p[0], p.size(), tmp, OPUS_MAX_OUT);
705 bool is_last = (i == packets.size() - 1);
706 if(lookahead_thrown < pregap_length) {
707 //We haven't yet thrown the full pregap. Throw some.
708 uint32_t maxthrow = pregap_length - lookahead_thrown;
709 pregap_throw = min(len, maxthrow);
710 lookahead_thrown += pregap_length;
711 }
712 if(is_last)
713 postgap_throw = min(len - pregap_throw, postgap_length);
714 tlen += (len - pregap_throw - postgap_throw);
715 for(uint32_t j = pregap_throw; j < len - postgap_throw; j++) {
716 int32_t s = (int32_t)(tmp[j] * lgain * 268435456.0);
717 write32sle(blank, s);
718 data.write(blank, 4);
719 if(!data)
720 throw std::runtime_error("Error writing PCM data.");
721 }
722 } catch(std::exception& e) {
723 (stringfmt() << "Error decoding opus packet: " << e.what()).throwex();
724 }
725 }
726 data.seekp(0, std::ios_base::beg);
727 write64ule(header + 8, tlen);
728 data.write(header, 32);
729 if(!data) {
730 throw std::runtime_error("Error writing PCM data.");
731 }
732 }
733
734 void opus_stream::export_stream(std::ofstream& data, external_stream_format extfmt)
735 {
736 if(extfmt == EXTFMT_OGGOPUS)
737 export_stream_oggopus(data);
738 else if(extfmt == EXTFMT_SOX)
739 export_stream_sox(data);
740 }
741
742 void opus_stream::write(uint8_t len, const unsigned char* payload, size_t payload_len)
743 {
744 try {
745 char descriptor[4];
746 uint32_t used_cluster, used_offset;
747 uint32_t used_mcluster, used_moffset;
748 if(!next_cluster)
749 next_cluster = data_cluster = fs.allocate_cluster();
750 if(!next_mcluster)
751 next_mcluster = ctrl_cluster = fs.allocate_cluster();
752 write16ube(descriptor, payload_len);
753 write8ube(descriptor + 2, len);
754 write8ube(descriptor + 3, 1);
755 fs.write_data(next_cluster, next_offset, payload, payload_len, used_cluster, used_offset);
756 fs.write_data(next_mcluster, next_moffset, descriptor, 4, used_mcluster, used_moffset);
757 uint64_t off = static_cast<uint64_t>(used_cluster) * CLUSTER_SIZE + used_offset;
758 opus_packetinfo p(payload_len, len, off);
759 total_len += p.length();
760 packets.push_back(p);
761 } catch(std::exception& e) {
762 (stringfmt() << "Can't write opus packet: " << e.what()).throwex();
763 }
764 }
765
766 void opus_stream::write_trailier()
767 {
768 try {
769 char descriptor[16];
770 uint32_t used_mcluster, used_moffset;
771 //The allocation must be done for real.
772 if(!next_mcluster)
773 next_mcluster = ctrl_cluster = fs.allocate_cluster();
774 //But the write must not update the pointers..
775 uint32_t tmp_mcluster = next_mcluster;
776 uint32_t tmp_moffset = next_moffset;
777 write32ube(descriptor, 0);
778 write32ube(descriptor + 4, (pregap_length << 8) | 0x02);
779 write32ube(descriptor + 8, (postgap_length << 8) | 0x03);
780 write16sbe(descriptor + 12, gain);
781 write16ube(descriptor + 14, 0x0004);
782 fs.write_data(tmp_mcluster, tmp_moffset, descriptor, 16, used_mcluster, used_moffset);
783 } catch(std::exception& e) {
784 (stringfmt() << "Can't write stream trailer: " << e.what()).throwex();
785 }
786 }
787
788
789 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
790 //Playing opus stream.
791 struct opus_playback_stream
792 {
793 //Create a new playing stream from given opus stream.
794 opus_playback_stream(opus_stream& data);
795 //Destroy playing opus stream.
796 ~opus_playback_stream();
797 //Read samples from stream.
798 //Can throw.
799 void read(float* data, size_t samples);
800 //Skip samples from stream.
801 //Can throw.
802 void skip(uint64_t samples);
803 //Has the stream already ended?
804 bool eof();
805 private:
806 opus_playback_stream(const opus_playback_stream&);
807 opus_playback_stream& operator=(const opus_playback_stream&);
808 //Can throw.
809 void decode_block();
810 float output[OPUS_MAX_OUT];
811 unsigned output_left;
812 uint32_t pregap_thrown;
813 bool postgap_thrown;
814 opus::decoder* decoder;
815 opus_stream& stream;
816 uint32_t next_block;
817 uint32_t blocks;
818 };
819
820 opus_playback_stream::opus_playback_stream(opus_stream& data)
821 : stream(data)
822 {
823 int err;
824 stream.get_ref();
825 stream.lock();
826 next_block = 0;
827 output_left = 0;
828 pregap_thrown = 0;
829 postgap_thrown = false;
830 blocks = stream.blocks();
831 decoder = new opus::decoder(opus::samplerate::r48k, false);
832 if(!decoder)
833 throw std::bad_alloc();
834 }
835
836 opus_playback_stream::~opus_playback_stream()
837 {
838 //No, we don't unlock the stream.
839 stream.put_ref();
840 delete decoder;
841 }
842
843 bool opus_playback_stream::eof()
844 {
845 return (next_block >= blocks && !output_left);
846 }
847
848 void opus_playback_stream::decode_block()
849 {
850 if(next_block >= blocks)
851 return;
852 if(output_left >= OPUS_MAX_OUT)
853 return;
854 unsigned plen = stream.packet_length(next_block);
855 if(plen + output_left > OPUS_MAX_OUT)
856 return;
857 std::vector<unsigned char> pdata = stream.packet(next_block);
858 try {
859 size_t c = decoder->decode(&pdata[0], pdata.size(), output + output_left,
860 OPUS_MAX_OUT - output_left);
861 output_left = min(output_left + c, static_cast<size_t>(OPUS_MAX_OUT));
862 } catch(...) {
863 //Bad packet, insert silence.
864 for(unsigned i = 0; i < plen; i++)
865 output[output_left++] = 0;
866 }
867 //Throw the pregap away if needed.
868 if(pregap_thrown < stream.get_pregap()) {
869 uint32_t throw_amt = min(stream.get_pregap() - pregap_thrown, (uint32_t)output_left);
870 if(throw_amt && throw_amt < output_left)
871 memmove(output, output + throw_amt, (output_left - throw_amt) * sizeof(float));
872 output_left -= throw_amt;
873 pregap_thrown += throw_amt;
874 }
875 next_block++;
876 }
877
878 void opus_playback_stream::read(float* data, size_t samples)
879 {
880 float lgain = stream.get_gain_linear();
881 while(samples > 0) {
882 decode_block();
883 if(next_block >= blocks && !postgap_thrown) {
884 //This is the final packet. Throw away postgap samples at the end.
885 uint32_t thrown = min(stream.get_postgap(), (uint32_t)output_left);
886 output_left -= thrown;
887 postgap_thrown = true;
888 }
889 if(next_block >= blocks && !output_left) {
890 //Zerofill remainder.
891 for(size_t i = 0; i < samples; i++)
892 data[i] = 0;
893 return;
894 }
895 unsigned maxcopy = min(static_cast<unsigned>(samples), output_left);
896 if(maxcopy) {
897 memcpy(data, output, maxcopy * sizeof(float));
898 for(size_t i = 0; i < maxcopy; i++)
899 data[i] *= lgain;
900 }
901 if(maxcopy < output_left && maxcopy)
902 memmove(output, output + maxcopy, (output_left - maxcopy) * sizeof(float));
903 output_left -= maxcopy;
904 samples -= maxcopy;
905 data += maxcopy;
906 }
907 }
908
909 void opus_playback_stream::skip(uint64_t samples)
910 {
911 //Adjust for preskip and declare all preskip already thrown away.
912 pregap_thrown = stream.get_pregap();
913 samples += pregap_thrown;
914 postgap_thrown = false;
915 //First, skip inside decoded samples.
916 if(samples < output_left) {
917 //Skipping less than amount in output buffer. Just discard from output buffer and try
918 //to decode a new block.
919 memmove(output, output + samples, (output_left - samples) * sizeof(float));
920 output_left -= samples;
921 decode_block();
922 return;
923 } else {
924 //Skipping at least the amount of samples in output buffer. First, blank the output buffer
925 //and count those towards samples discarded.
926 samples -= output_left;
927 output_left = 0;
928 }
929 //While number of samples is so great that adequate convergence period can be ensured without
930 //decoding this packet, just skip the samples from the packet.
931 while(samples > OPUS_CONVERGE_MAX) {
932 samples -= stream.packet_length(next_block++);
933 //Did we hit EOF?
934 if(next_block >= blocks)
935 return;
936 }
937 //Okay, we are near the point. Start decoding packets.
938 while(samples > 0) {
939 decode_block();
940 //Did we hit EOF?
941 if(next_block >= blocks && !output_left)
942 return;
943 //Skip as many samples as possible.
944 unsigned maxskip = min(static_cast<unsigned>(samples), output_left);
945 if(maxskip < output_left)
946 memmove(output, output + maxskip, (output_left - maxskip) * sizeof(float));
947 output_left -= maxskip;
948 samples -= maxskip;
949 }
950 //Just to be nice, decode a extra block.
951 decode_block();
952 }
953
954
955 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
956 //Collection of streams.
957 struct stream_collection
958 {
959 public:
960 //Create a new collection.
961 //Can throw.
962 stream_collection(filesystem_ref filesys);
963 //Destroy a collection. All streams are destroyed but not deleted.
964 ~stream_collection();
965 //Get list of streams active at given point.
966 std::list<uint64_t> streams_at(uint64_t point);
967 //Add a stream into collection.
968 //Can throw.
969 uint64_t add_stream(opus_stream& stream);
970 //Get the filesystem this collection is for.
971 filesystem_ref get_filesystem() { return fs; }
972 //Unlock all streams in collection.
973 void unlock_all();
974 //Get stream with given index (NULL if not found).
975 opus_stream* get_stream(uint64_t index)
976 {
977 umutex_class m(mutex);
978 if(streams.count(index)) {
979 streams[index]->get_ref();
980 return streams[index];
981 }
982 return NULL;
983 }
984 //Delete a stream.
985 //Can throw.
986 void delete_stream(uint64_t index);
987 //Alter stream timebase.
988 //Can throw.
989 void alter_stream_timebase(uint64_t index, uint64_t newts);
990 //Alter stream gain.
991 void alter_stream_gain(uint64_t index, uint16_t newgain);
992 //Enumerate all valid stream indices, in time order.
993 std::list<uint64_t> all_streams();
994 //Export the entiere superstream.
995 //Can throw.
996 void export_superstream(std::ofstream& out);
997 private:
998 filesystem_ref fs;
999 uint64_t next_index;
1000 unsigned next_stream;
1001 mutex_class mutex;
1002 std::set<uint64_t> free_indices;
1003 std::map<uint64_t, uint64_t> entries;
1004 std::multimap<uint64_t, uint64_t> streams_by_time;
1005 //FIXME: Something more efficient.
1006 std::map<uint64_t, opus_stream*> streams;
1007 };
1008
1009 stream_collection::stream_collection(filesystem_ref filesys)
1010 : fs(filesys)
1011 {
1012 next_stream = 0;
1013 next_index = 0;
1014 //The stream index table is in cluster 2.
1015 uint32_t next_cluster = 2;
1016 uint32_t next_offset = 0;
1017 uint32_t i = 0;
1018 try {
1019 while(true) {
1020 char buffer[16];
1021 size_t r = fs.read_data(next_cluster, next_offset, buffer, 16);
1022 if(r < 16)
1023 break;
1024 uint64_t timebase = read64ube(buffer);
1025 uint32_t ctrl_cluster = read32ube(buffer + 8);
1026 uint32_t data_cluster = read32ube(buffer + 12);
1027 if(ctrl_cluster) {
1028 opus_stream* x = new opus_stream(timebase, fs, ctrl_cluster, data_cluster);
1029 entries[next_index] = i;
1030 streams_by_time.insert(std::make_pair(timebase, next_index));
1031 streams[next_index++] = x;
1032 } else
1033 free_indices.insert(i);
1034 next_stream = ++i;
1035 }
1036 } catch(std::exception& e) {
1037 for(auto i : streams)
1038 i.second->put_ref();
1039 (stringfmt() << "Failed to parse LSVS: " << e.what()).throwex();
1040 }
1041 }
1042
1043 stream_collection::~stream_collection()
1044 {
1045 umutex_class m(mutex);
1046 for(auto i : streams)
1047 i.second->put_ref();
1048 streams.clear();
1049 }
1050
1051 std::list<uint64_t> stream_collection::streams_at(uint64_t point)
1052 {
1053 umutex_class m(mutex);
1054 std::list<uint64_t> s;
1055 for(auto i : streams) {
1056 uint64_t start = i.second->timebase();
1057 uint64_t end = start + i.second->length();
1058 if(point >= start && point < end) {
1059 i.second->get_ref();
1060 s.push_back(i.first);
1061 }
1062 }
1063 return s;
1064 }
1065
1066 uint64_t stream_collection::add_stream(opus_stream& stream)
1067 {
1068 uint64_t idx;
1069 try {
1070 umutex_class m(mutex);
1071 //Lock the added stream so it doesn't start playing back immediately.
1072 stream.lock();
1073 idx = next_index++;
1074 streams[idx] = &stream;
1075 char buffer[16];
1076 write64ube(buffer, stream.timebase());
1077 auto r = stream.get_clusters();
1078 write32ube(buffer + 8, r.first);
1079 write32ube(buffer + 12, r.second);
1080 uint64_t entry_number = 0;
1081 if(free_indices.empty())
1082 entry_number = next_stream++;
1083 else {
1084 entry_number = *free_indices.begin();
1085 free_indices.erase(entry_number);
1086 }
1087 uint32_t write_cluster = 2;
1088 uint32_t write_offset = 0;
1089 uint32_t dummy1, dummy2;
1090 fs.skip_data(write_cluster, write_offset, 16 * entry_number);
1091 fs.write_data(write_cluster, write_offset, buffer, 16, dummy1, dummy2);
1092 streams_by_time.insert(std::make_pair(stream.timebase(), idx));
1093 entries[idx] = entry_number;
1094 return idx;
1095 } catch(std::exception& e) {
1096 (stringfmt() << "Failed to add stream: " << e.what()).throwex();
1097 }
1098 return idx;
1099 }
1100
1101 void stream_collection::unlock_all()
1102 {
1103 umutex_class m(mutex);
1104 for(auto i : streams)
1105 i.second->unlock();
1106 }
1107
1108 void stream_collection::delete_stream(uint64_t index)
1109 {
1110 umutex_class m(mutex);
1111 if(!entries.count(index))
1112 return;
1113 uint64_t entry_number = entries[index];
1114 uint32_t write_cluster = 2;
1115 uint32_t write_offset = 0;
1116 uint32_t dummy1, dummy2;
1117 char buffer[16] = {0};
1118 fs.skip_data(write_cluster, write_offset, 16 * entry_number);
1119 fs.write_data(write_cluster, write_offset, buffer, 16, dummy1, dummy2);
1120 auto itr = streams_by_time.lower_bound(streams[index]->timebase());
1121 auto itr2 = streams_by_time.upper_bound(streams[index]->timebase());
1122 for(auto x = itr; x != itr2; x++)
1123 if(x->second == index) {
1124 streams_by_time.erase(x);
1125 break;
1126 }
1127 streams[index]->delete_stream();
1128 streams.erase(index);
1129 }
1130
1131 void stream_collection::alter_stream_timebase(uint64_t index, uint64_t newts)
1132 {
1133 try {
1134 umutex_class m(mutex);
1135 if(!streams.count(index))
1136 return;
1137 if(entries.count(index)) {
1138 char buffer[8];
1139 uint32_t write_cluster = 2;
1140 uint32_t write_offset = 0;
1141 uint32_t dummy1, dummy2;
1142 write64ube(buffer, newts);
1143 fs.skip_data(write_cluster, write_offset, 16 * entries[index]);
1144 fs.write_data(write_cluster, write_offset, buffer, 8, dummy1, dummy2);
1145 }
1146 auto itr = streams_by_time.lower_bound(streams[index]->timebase());
1147 auto itr2 = streams_by_time.upper_bound(streams[index]->timebase());
1148 for(auto x = itr; x != itr2; x++)
1149 if(x->second == index) {
1150 streams_by_time.erase(x);
1151 break;
1152 }
1153 streams[index]->timebase(newts);
1154 streams_by_time.insert(std::make_pair(newts, index));
1155 } catch(std::exception& e) {
1156 (stringfmt() << "Failed to alter stream timebase: " << e.what()).throwex();
1157 }
1158 }
1159
1160 void stream_collection::alter_stream_gain(uint64_t index, uint16_t newgain)
1161 {
1162 try {
1163 umutex_class m(mutex);
1164 if(!streams.count(index))
1165 return;
1166 streams[index]->set_gain(newgain);
1167 streams[index]->write_trailier();
1168 } catch(std::exception& e) {
1169 (stringfmt() << "Failed to alter stream gain: " << e.what()).throwex();
1170 }
1171 }
1172
1173 std::list<uint64_t> stream_collection::all_streams()
1174 {
1175 umutex_class m(mutex);
1176 std::list<uint64_t> s;
1177 for(auto i : streams_by_time)
1178 s.push_back(i.second);
1179 return s;
1180 }
1181
1182 void stream_collection::export_superstream(std::ofstream& out)
1183 {
1184 std::list<uint64_t> slist = all_streams();
1185 //Find the total length of superstream.
1186 uint64_t len = 0;
1187 for(auto i : slist) {
1188 opus_stream* s = get_stream(i);
1189 if(s) {
1190 len = max(len, s->timebase() + s->length());
1191 s->put_ref();
1192 }
1193 }
1194 char header[32];
1195 write64ule(header, 0x1C586F532EULL); //Magic and header size.
1196 write64ule(header + 8, len);
1197 write64ule(header + 16, 4676829883349860352ULL); //Sampling rate.
1198 write64ule(header + 24, 1);
1199 out.write(header, 32);
1200 if(!out)
1201 throw std::runtime_error("Error writing PCM output");
1202
1203 //Find the first valid stream.
1204 auto next_i = slist.begin();
1205 opus_stream* next_stream = NULL;
1206 while(next_i != slist.end()) {
1207 next_stream = get_stream(*next_i);
1208 next_i++;
1209 if(next_stream)
1210 break;
1211 }
1212 uint64_t next_ts;
1213 next_ts = next_stream ? next_stream->timebase() : len;
1214
1215 std::list<opus_playback_stream*> active;
1216 try {
1217 for(uint64_t s = 0; s < len;) {
1218 if(s == next_ts) {
1219 active.push_back(new opus_playback_stream(*next_stream));
1220 next_stream->put_ref();
1221 next_stream = NULL;
1222 while(next_i != slist.end()) {
1223 next_stream = get_stream(*next_i);
1224 next_i++;
1225 if(!next_stream)
1226 continue;
1227 uint64_t next_ts = next_stream->timebase();
1228 if(next_ts > s)
1229 break;
1230 //Okay, this starts too...
1231 active.push_back(new opus_playback_stream(*next_stream));
1232 next_stream->put_ref();
1233 next_stream = NULL;
1234 };
1235 next_ts = next_stream ? next_stream->timebase() : len;
1236 }
1237 uint64_t maxsamples = min(next_ts - s, static_cast<uint64_t>(OUTPUT_BLOCK));
1238 maxsamples = min(maxsamples, len - s);
1239 char outbuf[4 * OUTPUT_BLOCK];
1240 float buf1[OUTPUT_BLOCK];
1241 float buf2[OUTPUT_BLOCK];
1242 for(size_t t = 0; t < maxsamples; t++)
1243 buf1[t] = 0;
1244 for(auto t : active) {
1245 t->read(buf2, maxsamples);
1246 for(size_t u = 0; u < maxsamples; u++)
1247 buf1[u] += buf2[u];
1248 }
1249 for(auto t = active.begin(); t != active.end();) {
1250 if((*t)->eof()) {
1251 auto todel = t;
1252 t++;
1253 delete *todel;
1254 active.erase(todel);
1255 } else
1256 t++;
1257 }
1258 for(size_t t = 0; t < maxsamples; t++)
1259 write32sle(outbuf + 4 * t, buf1[t] * 268435456);
1260 out.write(outbuf, 4 * maxsamples);
1261 if(!out)
1262 throw std::runtime_error("Failed to write PCM");
1263 s += maxsamples;
1264 }
1265 } catch(std::exception& e) {
1266 (stringfmt() << "Failed to export PCM: " << e.what()).throwex();
1267 }
1268 for(auto t = active.begin(); t != active.end();) {
1269 if((*t)->eof()) {
1270 auto todelete = t;
1271 t++;
1272 delete *todelete;
1273 active.erase(todelete);
1274 } else
1275 t++;
1276 }
1277 }
1278
1279 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
1280 void start_management_stream(opus_stream& s)
1281 {
1282 opus_playback_stream* p = new opus_playback_stream(s);
1283 umutex_class m(active_playback_streams_lock);
1284 active_playback_streams.push_back(p);
1285 }
1286
1287 void advance_time(uint64_t newtime)
1288 {
1289 umutex_class m2(current_collection_lock);
1290 if(!current_collection) {
1291 //Clear all.
1292 umutex_class m(active_playback_streams_lock);
1293 for(auto i : active_playback_streams)
1294 delete i;
1295 active_playback_streams.clear();
1296 return;
1297 }
1298 std::list<uint64_t> sactive = current_collection->streams_at(newtime);
1299 for(auto j : sactive) {
1300 opus_stream* i = current_collection->get_stream(j);
1301 if(!i)
1302 continue;
1303 //Don't play locked streams in order to avoid double playing.
1304 umutex_class m(active_playback_streams_lock);
1305 try {
1306 if(!i->islocked())
1307 active_playback_streams.push_back(new opus_playback_stream(*i));
1308 } catch(std::exception& e) {
1309 messages << "Can't start stream: " << e.what() << std::endl;
1310 }
1311 i->put_ref();
1312 }
1313 }
1314
1315 void jump_time(uint64_t newtime)
1316 {
1317 umutex_class m2(current_collection_lock);
1318 if(!current_collection) {
1319 //Clear all.
1320 umutex_class m(active_playback_streams_lock);
1321 for(auto i : active_playback_streams)
1322 delete i;
1323 active_playback_streams.clear();
1324 return;
1325 }
1326 //Close all currently playing streams.
1327 {
1328 umutex_class m(active_playback_streams_lock);
1329 for(auto i : active_playback_streams)
1330 delete i;
1331 active_playback_streams.clear();
1332 }
1333 //Unlock all streams, so they will play.
1334 current_collection->unlock_all();
1335 //Reopen all streams that should be open (with seeking)
1336 std::list<uint64_t> sactive = current_collection->streams_at(newtime);
1337 for(auto j : sactive) {
1338 opus_stream* i = current_collection->get_stream(j);
1339 if(!i)
1340 continue;
1341 //No need to check for locks, because we just busted all of those.
1342 uint64_t p = newtime - i->timebase();
1343 opus_playback_stream* s;
1344 try {
1345 s = new opus_playback_stream(*i);
1346 } catch(std::exception& e) {
1347 messages << "Can't start stream: " << e.what() << std::endl;
1348 }
1349 i->put_ref();
1350 if(!s)
1351 continue;
1352 s->skip(p);
1353 umutex_class m(active_playback_streams_lock);
1354 active_playback_streams.push_back(s);
1355 }
1356 }
1357
1358 //Resample.
1359 void do_resample(audioapi_resampler& r, float* srcbuf, size_t& srcuse, float* dstbuf, size_t& dstuse,
1360 size_t dstmax, double ratio)
1361 {
1362 if(srcuse == 0 || dstuse >= dstmax)
1363 return;
1364 float* in = srcbuf;
1365 size_t in_u = srcuse;
1366 float* out = dstbuf + dstuse;
1367 size_t out_u = dstmax - dstuse;
1368 r.resample(in, in_u, out, out_u, ratio, false);
1369 size_t offset = in - srcbuf;
1370 if(offset < srcuse)
1371 memmove(srcbuf, srcbuf + offset, sizeof(float) * (srcuse - offset));
1372 srcuse -= offset;
1373 dstuse = dstmax - out_u;
1374 }
1375
1376 //Drain the input buffer.
1377 void drain_input()
1378 {
1379 while(audioapi_voice_r_status() > 0) {
1380 float buf[256];
1381 unsigned size = min(audioapi_voice_r_status(), 256u);
1382 audioapi_record_voice(buf, size);
1383 }
1384 }
1385
1386 //Read the input buffer.
1387 void read_input(float* buf, size_t& use, size_t maxuse)
1388 {
1389 size_t rleft = audioapi_voice_r_status();
1390 unsigned toread = min(rleft, max(maxuse, use) - use);
1391 if(toread > 0) {
1392 audioapi_record_voice(buf + use, toread);
1393 use += toread;
1394 }
1395 }
1396
1397 //Compress Opus block.
1398 void compress_opus_block(opus::encoder& e, float* buf, size_t& use, opus_stream& active_stream,
1399 bitrate_tracker& brtrack)
1400 {
1401 const size_t opus_out_max = 1276;
1402 unsigned char opus_output[opus_out_max];
1403 size_t cblock = 0;
1404 if(use >= 960)
1405 cblock = 960;
1406 else if(use >= 480)
1407 cblock = 480;
1408 else if(use >= 240)
1409 cblock = 240;
1410 else if(use >= 120)
1411 cblock = 120;
1412 else
1413 return; //No valid data to compress.
1414 const size_t opus_out_max2 = opus_max_bitrate.get() * cblock / 384000;
1415 try {
1416 size_t c = e.encode(buf, cblock, opus_output, opus_out_max2);
1417 //Successfully compressed a block.
1418 size_t opus_output_len = c;
1419 brtrack.submit(c, cblock);
1420 try {
1421 active_stream.write(cblock / 120, opus_output, opus_output_len);
1422 } catch(std::exception& e) {
1423 messages << "Error writing data: " << e.what() << std::endl;
1424 }
1425 } catch(std::exception& e) {
1426 messages << "Opus encoder error: " << e.what() << std::endl;
1427 }
1428 use -= cblock;
1429 }
1430
1431 void update_time()
1432 {
1433 uint64_t sampletime;
1434 bool jumping;
1435 {
1436 umutex_class m(time_mutex);
1437 sampletime = current_time;
1438 jumping = time_jump;
1439 time_jump = false;
1440 }
1441 if(jumping)
1442 jump_time(sampletime);
1443 else
1444 advance_time(sampletime);
1445 }
1446
1447 void decompress_active_streams(float* out, size_t& use)
1448 {
1449 size_t base = use;
1450 use += OUTPUT_BLOCK;
1451 for(unsigned i = 0; i < OUTPUT_BLOCK; i++)
1452 out[i + base] = 0;
1453 //Do it this way to minimize the amount of time playback streams lock
1454 //is held.
1455 std::list<opus_playback_stream*> stmp;
1456 {
1457 umutex_class m(active_playback_streams_lock);
1458 stmp = active_playback_streams;
1459 }
1460 std::set<opus_playback_stream*> toerase;
1461 for(auto i : stmp) {
1462 float tmp[OUTPUT_BLOCK];
1463 try {
1464 i->read(tmp, OUTPUT_BLOCK);
1465 } catch(std::exception& e) {
1466 messages << "Failed to decompress: " << e.what() << std::endl;
1467 for(unsigned j = 0; j < OUTPUT_BLOCK; j++)
1468 tmp[j] = 0;
1469 }
1470 for(unsigned j = 0; j < OUTPUT_BLOCK; j++)
1471 out[j + base] += tmp[j];
1472 if(i->eof())
1473 toerase.insert(i);
1474 }
1475 {
1476 umutex_class m(active_playback_streams_lock);
1477 for(auto i = active_playback_streams.begin(); i != active_playback_streams.end();) {
1478 if(toerase.count(*i)) {
1479 auto toerase = i;
1480 i++;
1481 delete *toerase;
1482 active_playback_streams.erase(toerase);
1483 } else
1484 i++;
1485 }
1486 }
1487 }
1488
1489 void handle_tangent_positive_edge(opus::encoder& e, opus_stream*& active_stream, bitrate_tracker& brtrack)
1490 {
1491 umutex_class m2(current_collection_lock);
1492 if(!current_collection)
1493 return;
1494 try {
1495 e.ctl(opus::reset);
1496 e.ctl(opus::bitrate(opus_bitrate.get()));
1497 brtrack.reset();
1498 uint64_t ctime;
1499 {
1500 umutex_class m(time_mutex);
1501 ctime = current_time;
1502 }
1503 active_stream = NULL;
1504 active_stream = new opus_stream(ctime, current_collection->get_filesystem());
1505 int32_t pregap = e.ctl(opus::lookahead);
1506 active_stream->set_pregap(pregap);
1507 } catch(std::exception& e) {
1508 messages << "Can't start stream: " << e.what() << std::endl;
1509 return;
1510 }
1511 messages << "Tangent enaged." << std::endl;
1512 }
1513
1514 void handle_tangent_negative_edge(opus_stream*& active_stream, bitrate_tracker& brtrack)
1515 {
1516 umutex_class m2(current_collection_lock);
1517 messages << "Tangent disenaged: " << brtrack;
1518 try {
1519 active_stream->write_trailier();
1520 } catch(std::exception& e) {
1521 messages << e.what() << std::endl;
1522 }
1523 if(current_collection) {
1524 try {
1525 current_collection->add_stream(*active_stream);
1526 } catch(std::exception& e) {
1527 messages << "Can't add stream: " << e.what() << std::endl;
1528 active_stream->put_ref();
1529 }
1530 information_dispatch::do_voice_stream_change();
1531 } else
1532 active_stream->put_ref();
1533 active_stream = NULL;
1534 }
1535
1536 class inthread_th : public worker_thread
1537 {
1538 public:
1539 inthread_th()
1540 {
1541 quit = false;
1542 quit_ack = false;
1543 rptr = 0;
1544 fire();
1545 }
1546 void kill()
1547 {
1548 quit = true;
1549 while(!quit_ack)
1550 usleep(100000);
1551 usleep(100000);
1552 }
1553 protected:
1554 void entry()
1555 {
1556 try {
1557 entry2();
1558 } catch(std::bad_alloc& e) {
1559 OOM_panic();
1560 } catch(std::exception& e) {
1561 messages << "AIEEE... Fatal exception in voice thread: " << e.what() << std::endl;
1562 }
1563 quit_ack = true;
1564 }
1565 void entry2()
1566 {
1567 int err;
1568 opus::encoder oenc(opus::samplerate::r48k, false, opus::application::voice);
1569 oenc.ctl(opus::bitrate(opus_bitrate.get()));
1570 audioapi_resampler rin;
1571 audioapi_resampler rout;
1572 const unsigned buf_max = 6144; //These buffers better be large.
1573 size_t buf_in_use = 0;
1574 size_t buf_inr_use = 0;
1575 size_t buf_outr_use = 0;
1576 size_t buf_out_use = 0;
1577 float buf_in[buf_max];
1578 float buf_inr[OPUS_BLOCK_SIZE];
1579 float buf_outr[OUTPUT_SIZE];
1580 float buf_out[buf_max];
1581 bitrate_tracker brtrack;
1582 opus_stream* active_stream = NULL;
1583
1584 drain_input();
1585 while(1) {
1586 if(clear_workflag(WORKFLAG_QUIT_REQUEST) & WORKFLAG_QUIT_REQUEST) {
1587 if(!active_flag && active_stream)
1588 handle_tangent_negative_edge(active_stream, brtrack);
1589 break;
1590 }
1591 uint64_t ticks = get_utime();
1592 //Handle tangent edgets.
1593 if(active_flag && !active_stream) {
1594 drain_input();
1595 buf_in_use = 0;
1596 buf_inr_use = 0;
1597 handle_tangent_positive_edge(oenc, active_stream, brtrack);
1598 }
1599 else if((!active_flag || quit) && active_stream)
1600 handle_tangent_negative_edge(active_stream, brtrack);
1601 if(quit)
1602 break;
1603
1604 //Read input, up to 25ms.
1605 unsigned rate_in = audioapi_voice_rate().first;
1606 unsigned rate_out = audioapi_voice_rate().second;
1607 size_t dbuf_max = min(buf_max, rate_in / REC_THRESHOLD_DIV);
1608 read_input(buf_in, buf_in_use, dbuf_max);
1609
1610 //Resample up to full opus block.
1611 do_resample(rin, buf_in, buf_in_use, buf_inr, buf_inr_use, OPUS_BLOCK_SIZE,
1612 1.0 * OPUS_SAMPLERATE / rate_in);
1613
1614 //If we have full opus block and recording is enabled, compress it.
1615 if(buf_inr_use >= OPUS_BLOCK_SIZE && active_stream)
1616 compress_opus_block(oenc, buf_inr, buf_inr_use, *active_stream, brtrack);
1617
1618 //Update time, starting/ending streams.
1619 update_time();
1620
1621 //Decompress active streams.
1622 if(buf_outr_use < BLOCK_THRESHOLD)
1623 decompress_active_streams(buf_outr, buf_outr_use);
1624
1625 //Resample to output rate.
1626 do_resample(rout, buf_outr, buf_outr_use, buf_out, buf_out_use, buf_max,
1627 1.0 * rate_out / OPUS_SAMPLERATE);
1628
1629 //Output stuff.
1630 if(buf_out_use > 0 && audioapi_voice_p_status2() < rate_out / PLAY_THRESHOLD_DIV) {
1631 audioapi_play_voice(buf_out, buf_out_use);
1632 buf_out_use = 0;
1633 }
1634
1635 //Sleep a bit to save CPU use.
1636 uint64_t ticks_spent = get_utime() - ticks;
1637 if(ticks_spent < ITERATION_TIME)
1638 usleep(ITERATION_TIME - ticks_spent);
1639 }
1640 delete current_collection;
1641 }
1642 private:
1643 size_t rptr;
1644 double position;
1645 volatile bool quit;
1646 volatile bool quit_ack;
1647 };
1648
1649 //The tangent function.
1650 function_ptr_command<> ptangent(lsnes_cmd, "+tangent", "Voice tangent",
1651 "Syntax: +tangent\nVoice tangent.\n",
1652 []() throw(std::bad_alloc, std::runtime_error) {
1653 active_flag = true;
1654 });
1655 function_ptr_command<> ntangent(lsnes_cmd, "-tangent", "Voice tangent",
1656 "Syntax: -tangent\nVoice tangent.\n",
1657 []() throw(std::bad_alloc, std::runtime_error) {
1658 active_flag = false;
1659 });
1660 inverse_bind itangent(lsnes_mapper, "+tangent", "Movie‣Voice tangent");
1661 inthread_th* int_task;
1662 }
1663
1664 //Rate is not sampling rate!
1665 void voice_frame_number(uint64_t newframe, double rate)
1666 {
1667 if(rate == last_rate && last_frame_number == newframe)
1668 return;
1669 umutex_class m(time_mutex);
1670 current_time = newframe / rate * OPUS_SAMPLERATE;
1671 if(fabs(rate - last_rate) > 1e-6 || last_frame_number + 1 != newframe)
1672 time_jump = true;
1673 last_frame_number = newframe;
1674 last_rate = rate;
1675 }
1676
1677 void voicethread_task()
1678 {
1679 int_task = new inthread_th;
1680 }
1681
1682 void voicethread_kill()
1683 {
1684 int_task->kill();
1685 int_task = NULL;
1686 }
1687
1688 uint64_t voicesub_parse_timebase(const std::string& n)
1689 {
1690 std::string x = n;
1691 if(x.length() > 0 && x[x.length() - 1] == 's') {
1692 x = x.substr(0, x.length() - 1);
1693 return 48000 * parse_value<double>(x);
1694 } else
1695 return parse_value<uint64_t>(x);
1696 }
1697
1698 bool voicesub_collection_loaded()
1699 {
1700 umutex_class m2(current_collection_lock);
1701 return (current_collection != NULL);
1702 }
1703
1704 std::list<playback_stream_info> voicesub_get_stream_info()
1705 {
1706 umutex_class m2(current_collection_lock);
1707 std::list<playback_stream_info> in;
1708 if(!current_collection)
1709 return in;
1710 for(auto i : current_collection->all_streams()) {
1711 opus_stream* s = current_collection->get_stream(i);
1712 playback_stream_info pi;
1713 if(!s)
1714 continue;
1715 pi.id = i;
1716 pi.base = s->timebase();
1717 pi.length = s->length();
1718 try {
1719 in.push_back(pi);
1720 } catch(...) {
1721 }
1722 s->put_ref();
1723 }
1724 return in;
1725 }
1726
1727 void voicesub_play_stream(uint64_t id)
1728 {
1729 umutex_class m2(current_collection_lock);
1730 if(!current_collection)
1731 throw std::runtime_error("No collection loaded");
1732 opus_stream* s = current_collection->get_stream(id);
1733 if(!s)
1734 return;
1735 try {
1736 start_management_stream(*s);
1737 } catch(...) {
1738 s->put_ref();
1739 throw;
1740 }
1741 s->put_ref();
1742 }
1743
1744 void voicesub_export_stream(uint64_t id, const std::string& filename, external_stream_format fmt)
1745 {
1746 umutex_class m2(current_collection_lock);
1747 if(!current_collection)
1748 throw std::runtime_error("No collection loaded");
1749 opus_stream* st = current_collection->get_stream(id);
1750 if(!st)
1751 return;
1752 std::ofstream s(filename, std::ios_base::out | std::ios_base::binary);
1753 if(!s) {
1754 st->put_ref();
1755 throw std::runtime_error("Can't open output file");
1756 }
1757 try {
1758 st->export_stream(s, fmt);
1759 } catch(std::exception& e) {
1760 st->put_ref();
1761 (stringfmt() << "Export failed: " << e.what()).throwex();
1762 }
1763 st->put_ref();
1764 }
1765
1766 uint64_t voicesub_import_stream(uint64_t ts, const std::string& filename, external_stream_format fmt)
1767 {
1768 umutex_class m2(current_collection_lock);
1769 if(!current_collection)
1770 throw std::runtime_error("No collection loaded");
1771
1772 std::ifstream s(filename, std::ios_base::in | std::ios_base::binary);
1773 if(!s)
1774 throw std::runtime_error("Can't open input file");
1775 opus_stream* st = new opus_stream(ts, current_collection->get_filesystem(), s, fmt);
1776 uint64_t id;
1777 try {
1778 id = current_collection->add_stream(*st);
1779 } catch(...) {
1780 st->delete_stream();
1781 throw;
1782 }
1783 st->unlock(); //Not locked.
1784 information_dispatch::do_voice_stream_change();
1785 return id;
1786 }
1787
1788 void voicesub_delete_stream(uint64_t id)
1789 {
1790 umutex_class m2(current_collection_lock);
1791 if(!current_collection)
1792 throw std::runtime_error("No collection loaded");
1793 current_collection->delete_stream(id);
1794 information_dispatch::do_voice_stream_change();
1795 }
1796
1797 void voicesub_export_superstream(const std::string& filename)
1798 {
1799 umutex_class m2(current_collection_lock);
1800 if(!current_collection)
1801 throw std::runtime_error("No collection loaded");
1802 std::ofstream s(filename, std::ios_base::out | std::ios_base::binary);
1803 if(!s)
1804 throw std::runtime_error("Can't open output file");
1805 current_collection->export_superstream(s);
1806 }
1807
1808 void voicesub_load_collection(const std::string& filename)
1809 {
1810 umutex_class m2(current_collection_lock);
1811 filesystem_ref newfs;
1812 stream_collection* newc;
1813 newfs = filesystem_ref(filename);
1814 newc = new stream_collection(newfs);
1815 if(current_collection)
1816 delete current_collection;
1817 current_collection = newc;
1818 information_dispatch::do_voice_stream_change();
1819 }
1820
1821 void voicesub_unload_collection()
1822 {
1823 umutex_class m2(current_collection_lock);
1824 if(current_collection)
1825 delete current_collection;
1826 current_collection = NULL;
1827 information_dispatch::do_voice_stream_change();
1828 }
1829
1830 void voicesub_alter_timebase(uint64_t id, uint64_t ts)
1831 {
1832 umutex_class m2(current_collection_lock);
1833 if(!current_collection)
1834 throw std::runtime_error("No collection loaded");
1835 current_collection->alter_stream_timebase(id, ts);
1836 information_dispatch::do_voice_stream_change();
1837 }
1838
1839 float voicesub_get_gain(uint64_t id)
1840 {
1841 umutex_class m2(current_collection_lock);
1842 if(!current_collection)
1843 throw std::runtime_error("No collection loaded");
1844 return current_collection->get_stream(id)->get_gain() / 256.0;
1845 }
1846
1847 void voicesub_set_gain(uint64_t id, float gain)
1848 {
1849 umutex_class m2(current_collection_lock);
1850 if(!current_collection)
1851 throw std::runtime_error("No collection loaded");
1852 int64_t _gain = gain * 256;
1853 if(_gain < -32768 || _gain > 32767)
1854 throw std::runtime_error("Gain out of range (+-128dB)");
1855 current_collection->alter_stream_gain(id, _gain);
1856 information_dispatch::do_voice_stream_change();
1857 }
1858
1859 double voicesub_ts_seconds(uint64_t ts)
1860 {
1861 return ts / 48000.0;
1862 }
1863 #else
1864 void voicethread_task() {}
1865 void voice_frame_number(uint64_t newframe, double rate) {}
1866 void voicethread_kill() {}
1867 #endif
Rerecording SNES/GB(C) emulator based on bsnes / gambatte core