mptsd reads mpegts streams from udp/multicast or http and combines them into one multiple program stream that is suitable for outputting to DVB-C modulator. Tested with Dektec DTE-3114 Quad QAM Modulator and used in production in small DVB-C networks. https://georgi.unixsol.org/programs/mptsd/

output_mix.c 6.7KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233
  1. /*
  2. * mptsd output mix packets
  3. * Copyright (C) 2010-2011 Unix Solutions Ltd.
  4. *
  5. * This program is free software; you can redistribute it and/or modify
  6. * it under the terms of the GNU General Public License version 2
  7. * as published by the Free Software Foundation.
  8. *
  9. * This program is distributed in the hope that it will be useful,
  10. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. * GNU General Public License for more details.
  13. *
  14. * You should have received a copy of the GNU General Public License
  15. * along with this program; if not, write to the Free Software
  16. * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston MA 02110-1301, USA.
  17. */
  18. #include <math.h>
  19. #include <stdlib.h>
  20. #include <unistd.h>
  21. #include <string.h>
  22. #include <signal.h>
  23. #include <errno.h>
  24. #include "libfuncs/libfuncs.h"
  25. #include "libtsfuncs/tsfuncs.h"
  26. #include "data.h"
  27. #include "config.h"
  28. #include "input.h"
  29. void output_show_programs(CONFIG *conf) {
  30. LNODE *lr, *lrtmp;
  31. list_for_each(conf->inputs, lr, lrtmp) {
  32. INPUT *r = lr->data;
  33. if (r->input_ready == 1) {
  34. LOGf("OUTPUT: [%-12s] Service %d appeared.\n", r->channel->id, r->channel->service_id);
  35. r->input_ready++;
  36. }
  37. }
  38. }
  39. void * output_handle_mix(void *_config) {
  40. LNODE *lr, *lrtmp;
  41. LNODE *inpt; // Track last used input
  42. CONFIG *conf = _config;
  43. OUTPUT *o = conf->output;
  44. int buf_in_use = 0;
  45. unsigned int o_datasize, o_packets, packets;
  46. unsigned int o_maxpackets = o->obuf[0].size / TS_PACKET_SIZE;
  47. signal(SIGPIPE, SIG_IGN);
  48. inpt = conf->inputs->tail; // Next is the first one
  49. while (!o->dienow) {
  50. OBUF *curbuf = &o->obuf[buf_in_use];
  51. usleep(o->obuf_ms); // Fill interval
  52. output_show_programs(conf);
  53. while (curbuf->status != obuf_empty) {
  54. if (o->dienow)
  55. goto OUT;
  56. //LOGf("MIX: Waiting for obuf %d\n", buf_in_use);
  57. usleep(1);
  58. }
  59. list_lock(conf->inputs);
  60. o_datasize = o->psibuf->input - o->psibuf->output; // PSI data
  61. list_for_each(conf->inputs, lr, lrtmp) { // INPUT data
  62. INPUT *r = lr->data;
  63. o_datasize += r->buf->input - r->buf->output;
  64. }
  65. o_packets = o_datasize / TS_PACKET_SIZE;
  66. packets = min(o_packets, o_maxpackets);
  67. double null_per_data = 1;
  68. double data_per_null = 0;
  69. if (o_maxpackets - packets) {
  70. data_per_null = (double)packets / (o_maxpackets-packets);
  71. if (data_per_null < 1) {
  72. null_per_data = (double)(o_maxpackets-packets) / packets;
  73. data_per_null = 1;
  74. }
  75. }
  76. curbuf->status = obuf_filling; // Mark buffer as being filled
  77. if (conf->debug) {
  78. LOGf("MIX[%2d]: Data:%6u | Bufsz:%6d | Packs:%4u | D/N:%5.2f/%5.2f\n",
  79. buf_in_use,
  80. o_datasize,
  81. curbuf->size,
  82. packets,
  83. ((double)packets / o_maxpackets) * 100,
  84. (double)100 - ((double)packets / o_maxpackets) * 100
  85. );
  86. LOGf("datapacks:%5d maxpacks:%5d null:%5d (%5.2f) | null_per_data:%5.2f data_per_null:%5.2f\n",
  87. packets,
  88. o_maxpackets,
  89. o_maxpackets-packets,
  90. 100-((double)packets / o_maxpackets)*100,
  91. null_per_data,
  92. data_per_null
  93. );
  94. }
  95. unsigned int nulls=0, null_packets_count = o_maxpackets - packets;
  96. // The is no data in the input buffer, send only NULLs
  97. if (null_packets_count == o_maxpackets) {
  98. // Increase sended packets
  99. list_for_each(conf->inputs, lr, lrtmp) {
  100. INPUT *r = lr->data;
  101. r->outputed_packets += o_maxpackets;
  102. }
  103. goto NEXT_BUFFER;
  104. }
  105. unsigned int data_packets;
  106. int data_size;
  107. uint8_t *data;
  108. for (data_packets=0;data_packets<packets;data_packets++) {
  109. if (o->dienow)
  110. break;
  111. // Try the PSI data first
  112. data = cbuf_get(o->psibuf, TS_PACKET_SIZE, &data_size);
  113. if (data && data_size == TS_PACKET_SIZE)
  114. goto SEND_PACKET;
  115. // Loop over inputs
  116. int inputs_left = conf->inputs->items;
  117. while (inputs_left--) {
  118. inpt = inpt->next;
  119. INPUT *r = inpt->data;
  120. if (!r || !r->buf)
  121. continue;
  122. // Move pcrs || Move & rewrite prcs
  123. if (conf->pcr_mode == 1 || conf->pcr_mode == 3) {
  124. // Is there any data in this input?
  125. data = cbuf_peek(r->buf, TS_PACKET_SIZE, &data_size);
  126. if (data_size == TS_PACKET_SIZE) {
  127. uint16_t pid = ts_packet_get_pid(data);
  128. // Do we have PCR packet?
  129. if (pid == r->output_pcr_pid && ts_packet_has_pcr(data)) {
  130. if (r->output_pcr_packets_needed > 0 && r->outputed_packets < r->output_pcr_packets_needed) {
  131. data = NULL;
  132. data_size = 0;
  133. continue;
  134. }
  135. /*
  136. LOGf("%10s | pcr:%15llu last_pcr:%15llu diff:%10lld packets:%5d needed_packs:%d diff:%d\n",
  137. r->channel->id,
  138. r->output_pcr,
  139. r->output_last_pcr,
  140. r->output_pcr - r->output_last_pcr,
  141. r->outputed_packets,
  142. r->output_pcr_packets_needed,
  143. r->outputed_packets - r->output_pcr_packets_needed
  144. );
  145. */
  146. uint64_t last_last_pcr = r->output_last_pcr;
  147. r->output_last_pcr = r->output_pcr;
  148. r->output_pcr = ts_packet_get_pcr(data);
  149. if (last_last_pcr)
  150. r->output_pcr_packets_needed = round(conf->output_bitrate / 8 * (r->output_pcr - r->output_last_pcr) / 27000000 / 188);
  151. r->outputed_packets = 0;
  152. }
  153. data = cbuf_get(r->buf, TS_PACKET_SIZE, &data_size);
  154. if (data_size == TS_PACKET_SIZE) // We have our data, no need to look at other inputs
  155. break;
  156. }
  157. // Do not move PCRs
  158. } else {
  159. data = cbuf_get(r->buf, TS_PACKET_SIZE, &data_size);
  160. if (data_size == TS_PACKET_SIZE) // We have our data, no need to look at other inputs
  161. break;
  162. }
  163. } // while (inputs_left--)
  164. // We have data. Mix it with NULLs and stuff it in the output buffer
  165. // If the have no data, the output buffer will automaticaly be left
  166. // with NULL packets
  167. SEND_PACKET:
  168. if (data && data_size == TS_PACKET_SIZE) {
  169. // Mix data with NULLs
  170. if (nulls < null_packets_count) {
  171. if (round(nulls * data_per_null) < round(data_packets * null_per_data)) {
  172. nulls += round(data_packets * null_per_data) - round(nulls * data_per_null);
  173. }
  174. if (nulls > null_packets_count)
  175. nulls = null_packets_count;
  176. }
  177. if (data_packets+nulls >= o_maxpackets) { // Can't happen
  178. LOGf("wtf: %d packets:%d\n", data_packets+nulls, o_maxpackets);
  179. break;
  180. }
  181. uint8_t *bufptr = curbuf->buf + ((data_packets + nulls) * TS_PACKET_SIZE);
  182. memcpy(bufptr, data, TS_PACKET_SIZE);
  183. }
  184. // Increase sended packets
  185. list_for_each(conf->inputs, lr, lrtmp) {
  186. INPUT *r = lr->data;
  187. r->outputed_packets++;
  188. }
  189. }
  190. NEXT_BUFFER:
  191. list_unlock(conf->inputs);
  192. curbuf->status = obuf_full; // Mark buffer as full
  193. buf_in_use = buf_in_use ? 0 : 1; // Switch buffer
  194. }
  195. OUT:
  196. LOG("OUTPUT: MIX thread stopped.\n");
  197. o->dienow++;
  198. LNODE *l, *tmp;
  199. list_for_each(conf->inputs, l, tmp) {
  200. INPUT *r = l->data;
  201. r->dienow = 1;
  202. }
  203. return 0;
  204. }