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/
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

data.c 10KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423
  1. /*
  2. * mptsd data
  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 <stdlib.h>
  19. #include <math.h>
  20. #include <ctype.h>
  21. #include <unistd.h>
  22. #include <string.h>
  23. #include <sys/time.h>
  24. #include <regex.h>
  25. #include <sys/types.h>
  26. #include <sys/stat.h>
  27. #include <fcntl.h>
  28. #include "libfuncs/io.h"
  29. #include "libfuncs/log.h"
  30. #include "libfuncs/list.h"
  31. #include "libfuncs/asyncdns.h"
  32. #include "libtsfuncs/tsfuncs.h"
  33. #include "data.h"
  34. #include "config.h"
  35. #include "output.h"
  36. extern CONFIG *config;
  37. channel_source get_sproto(char *url) {
  38. return strncmp(url, "http", 4)==0 ? tcp_sock : udp_sock;
  39. }
  40. int is_rtp(char *url) {
  41. return strncmp(url, "rtp", 3) == 0;
  42. }
  43. CHANSRC *chansrc_init(char *url) {
  44. if (!url)
  45. return NULL;
  46. regex_t re;
  47. regmatch_t res[5];
  48. regcomp(&re, "^([a-z]+)://([^:/?]+):?([0-9]*)/?(.*)", REG_EXTENDED);
  49. if (regexec(&re,url,5,res,0)==0) {
  50. char *data = strdup(url);
  51. char *proto, *host, *port, *path;
  52. int iport;
  53. proto= data+res[1].rm_so; data[res[1].rm_eo]=0;
  54. host = data+res[2].rm_so; data[res[2].rm_eo]=0;
  55. port = data+res[3].rm_so; data[res[3].rm_eo]=0;
  56. path = data+res[4].rm_so; data[res[4].rm_eo]=0;
  57. iport = atoi(port);
  58. /* Setup */
  59. CHANSRC *src = calloc(1, sizeof(CHANSRC));
  60. src->proto = strdup(proto);
  61. src->sproto= get_sproto(url);
  62. src->host = strdup(host);
  63. src->port = iport ? iport : 80;
  64. src->path = strdup(path);
  65. src->rtp = strcmp(proto, "rtp") == 0;
  66. FREE(data);
  67. regfree(&re);
  68. return src;
  69. }
  70. regfree(&re);
  71. return NULL;
  72. }
  73. void chansrc_free(CHANSRC **purl) {
  74. CHANSRC *url = *purl;
  75. if (url) {
  76. FREE(url->proto);
  77. FREE(url->host);
  78. FREE(url->path);
  79. FREE(*purl);
  80. }
  81. };
  82. void chansrc_add(CHANNEL *c, const char *src) {
  83. if (c->num_src >= MAX_CHANNEL_SOURCES-1)
  84. return;
  85. c->sources[c->num_src] = strdup(src);
  86. if (c->num_src == 0) /* Set default source to first one */
  87. c->source = c->sources[c->num_src];
  88. c->num_src++;
  89. }
  90. void chansrc_next(CHANNEL *c) {
  91. if (c->num_src <= 1)
  92. return;
  93. // uint8_t old_src = c->curr_src;
  94. c->curr_src++;
  95. if (c->curr_src >= MAX_CHANNEL_SOURCES-1 || c->sources[c->curr_src] == NULL)
  96. c->curr_src = 0;
  97. c->source = c->sources[c->curr_src];
  98. // LOGf("CHAN : Switch source | Channel: %s OldSrc: %d %s NewSrc: %d %s\n", c->name, old_src, c->sources[old_src], c->curr_src, c->source);
  99. }
  100. void chansrc_set(CHANNEL *c, uint8_t src_id) {
  101. if (src_id >= MAX_CHANNEL_SOURCES-1 || c->sources[src_id] == NULL)
  102. return;
  103. // uint8_t old_src = c->curr_src;
  104. c->curr_src = src_id;
  105. c->source = c->sources[c->curr_src];
  106. // LOGf("CHAN : Set source | Channel: %s OldSrc: %d %s NewSrc: %d %s\n", c->name, old_src, c->sources[old_src], c->curr_src, c->source);
  107. }
  108. CHANNEL *channel_new(int service_id, int is_radio, const char *id, const char *name, const char *source, int channel_index){
  109. if (channel_index<=0 || channel_index>=256)
  110. {
  111. LOGf("CONFIG: Error channel_new invalid index %d\n", channel_index);
  112. return NULL;
  113. }
  114. //LOGf("CONFIG: ------------------channel_new() serviceid %d id %s name %s source %s index %d\n", service_id, id, name , source , channel_index);
  115. CHANNEL *c = calloc(1, sizeof(CHANNEL));
  116. c->service_id = service_id;
  117. c->radio = is_radio;
  118. c->index = channel_index;
  119. c->base_pid = c->index * 32; // The first pid is saved for PMT , channel_index must > 0
  120. c->pmt_pid = c->base_pid; // The first pid is saved for PMT
  121. c->id = strdup(id);
  122. c->name = strdup(name);
  123. chansrc_add(c, source);
  124. return c;
  125. }
  126. void channel_free_epg(CHANNEL *c) {
  127. epg_free(&c->epg_now);
  128. epg_free(&c->epg_next);
  129. ts_eit_free(&c->eit_now);
  130. ts_eit_free(&c->eit_next);
  131. }
  132. void channel_free(CHANNEL **pc) {
  133. CHANNEL *c = *pc;
  134. if (c) {
  135. channel_free_epg(c);
  136. FREE(c->id);
  137. FREE(c->name);
  138. int i;
  139. for (i=c->num_src-1; i>=0; i--) {
  140. FREE(c->sources[i]);
  141. }
  142. c->source = NULL;
  143. FREE(*pc);
  144. }
  145. }
  146. EPG_ENTRY *epg_new(time_t start, int duration, char *encoding, char *event, char *short_desc, char *long_desc) {
  147. EPG_ENTRY *e;
  148. if (!event)
  149. return NULL;
  150. e = calloc(1, sizeof(EPG_ENTRY));
  151. e->event_id = (start / 60) &~ 0xffff0000;
  152. e->start = start;
  153. e->duration = duration;
  154. if (encoding && strcmp(encoding, "iso-8859-5")==0) {
  155. e->event = init_dvb_string_iso_8859_5(event);
  156. e->short_desc = init_dvb_string_iso_8859_5(short_desc);
  157. e->long_desc = init_dvb_string_iso_8859_5(long_desc);
  158. } else { // Default is utf-8
  159. e->event = init_dvb_string_utf8(event);
  160. e->short_desc = init_dvb_string_utf8(short_desc);
  161. e->long_desc = init_dvb_string_utf8(long_desc);
  162. }
  163. return e;
  164. }
  165. void epg_free(EPG_ENTRY **pe) {
  166. EPG_ENTRY *e = *pe;
  167. if (e) {
  168. FREE(e->event);
  169. FREE(e->short_desc);
  170. FREE(e->long_desc);
  171. FREE(*pe);
  172. }
  173. }
  174. // Return 1 if they are different
  175. // Return 0 if they are the same
  176. int epg_changed(EPG_ENTRY *a, EPG_ENTRY *b) {
  177. if (!a && b) return 1;
  178. if (!b && a) return 1;
  179. if (!a && !b) return 0;
  180. if (a->event_id != b->event_id) return 1;
  181. if (a->start != b->start) return 1;
  182. if (a->duration != b->duration) return 1;
  183. if (xstrcmp(a->event, b->event) != 0) return 1;
  184. if (xstrcmp(a->short_desc, b->short_desc) != 0) return 1;
  185. if (xstrcmp(a->long_desc, b->long_desc) != 0) return 1;
  186. return 0;
  187. }
  188. void input_stream_alloc(INPUT *input) {
  189. input->stream.pidref = pidref_init(64, input->channel->base_pid);
  190. input->stream.pat = ts_pat_alloc();
  191. input->stream.pmt = ts_pmt_alloc();
  192. input->stream.last_pat = ts_pat_alloc();
  193. input->stream.last_pmt = ts_pmt_alloc();
  194. }
  195. void input_stream_free(INPUT *input) {
  196. ts_pmt_free(&input->stream.pmt);
  197. ts_pmt_free(&input->stream.pmt_rewritten);
  198. ts_pmt_free(&input->stream.last_pmt);
  199. ts_pat_free(&input->stream.pat);
  200. ts_pat_free(&input->stream.pat_rewritten);
  201. ts_pat_free(&input->stream.last_pat);
  202. pidref_free(&input->stream.pidref);
  203. input->stream.nit_pid = 0;
  204. input->stream.pmt_pid = 0;
  205. input->stream.pcr_pid = 0;
  206. input->stream.input_pcr = 0;
  207. }
  208. void input_stream_reset(INPUT *input) {
  209. input_stream_free(input);
  210. input_stream_alloc(input);
  211. }
  212. INPUT * input_new(const char *name, CHANNEL *channel) {
  213. char *tmp;
  214. INPUT *r = calloc(1, sizeof(INPUT));
  215. r->name = strdup(name);
  216. r->sock = -1;
  217. r->channel = channel;
  218. if (config->write_input_file) {
  219. asprintf(&tmp, "mptsd-input-%s.ts", channel->id);
  220. r->ifd = open(tmp, O_CREAT | O_WRONLY | O_TRUNC, 0644);
  221. FREE(tmp);
  222. }
  223. r->buf = cbuf_init(1428 * 1316, channel->id); // ~ 10000 x 188
  224. input_stream_alloc(r);
  225. return r;
  226. }
  227. void input_free(INPUT **pinput) {
  228. INPUT *r = *pinput;
  229. if (!r)
  230. return;
  231. if (r->sock > -1)
  232. shutdown_fd(&(r->sock));
  233. if (r->freechannel)
  234. channel_free(&r->channel);
  235. if (r->ifd)
  236. close(r->ifd);
  237. input_stream_free(r);
  238. cbuf_free(&r->buf);
  239. FREE(r->name);
  240. FREE(*pinput);
  241. }
  242. OUTPUT *output_new() {
  243. OUTPUT *o = calloc(1, sizeof(OUTPUT));
  244. o->obuf_ms = 100;
  245. o->psibuf = cbuf_init(50 * 1316, "psi");
  246. if (!o->psibuf) {
  247. LOGf("ERROR: Can't allocate PSI input buffer\n");
  248. exit(1);
  249. }
  250. cbuf_poison(o->psibuf, 'Y');
  251. return o;
  252. }
  253. void output_open_file(OUTPUT *o) {
  254. o->ofd = open("mptsd-output.ts", O_CREAT | O_WRONLY | O_TRUNC, 0644);
  255. }
  256. void obuf_reset(OBUF *ob) {
  257. int i;
  258. memset(ob->buf, 0xff, ob->size);
  259. for (i=0; i<ob->size; i+=188) {
  260. ob->buf[i+0] = 0x47;
  261. ob->buf[i+1] = 0x1f;
  262. ob->buf[i+2] = 0xff;
  263. ob->buf[i+3] = 0x00;
  264. }
  265. ob->written = 0;
  266. ob->status = obuf_empty;
  267. }
  268. void output_buffer_alloc(OUTPUT *o, double output_bitrate) {
  269. if (!output_bitrate) {
  270. LOGf("No output bitrate, can't determine buffer!\n");
  271. exit(1);
  272. }
  273. o->output_bitrate = output_bitrate;
  274. long pps = ceil((double)output_bitrate / (FRAME_PACKET_SIZE * 8)); // Packets per second
  275. long ppms = ceil((double)pps / ((double)1000 / o->obuf_ms)); // Packets per o->buffer_ms miliseconds
  276. long obuf_size = ppms * 1316;
  277. o->obuf[0].size = obuf_size;
  278. o->obuf[0].status = obuf_empty;
  279. o->obuf[0].buf = malloc(o->obuf[0].size);
  280. obuf_reset(&o->obuf[0]);
  281. o->obuf[1].size = obuf_size;
  282. o->obuf[1].status = obuf_empty;
  283. o->obuf[1].buf = malloc(o->obuf[0].size);
  284. obuf_reset(&o->obuf[1]);
  285. LOGf("\tOutput buf size : %ld * 2 = %ld\n", obuf_size, obuf_size * 2);
  286. LOGf("\tOutput buf packets: %ld (188 bytes)\n", obuf_size / 188);
  287. LOGf("\tOutput buf frames : %ld (1316 bytes)\n", obuf_size / 1316);
  288. LOGf("\tOutput buf ms : %u ms\n", o->obuf_ms);
  289. }
  290. void output_free(OUTPUT **po) {
  291. OUTPUT *o = *po;
  292. if (!o)
  293. return;
  294. if (o->out_sock > -1)
  295. shutdown_fd(&(o->out_sock));
  296. if (o->ofd)
  297. close(o->ofd);
  298. cbuf_free(&o->psibuf);
  299. FREE(o->obuf[0].buf);
  300. FREE(o->obuf[1].buf);
  301. output_psi_free(o);
  302. FREE(*po);
  303. }
  304. NIT *nit_new(uint16_t ts_id, char *freq, char *mod, char *symbol_rate) {
  305. char tmp[9];
  306. unsigned i, pos;
  307. if (strlen(freq) != 9 || strlen(symbol_rate) != 8)
  308. return NULL;
  309. NIT *n = calloc(1, sizeof(NIT));
  310. n->freq = strdup(freq);
  311. n->modulation = strdup(mod);
  312. n->symbol_rate = strdup(symbol_rate);
  313. n->ts_id = ts_id;
  314. n->_modulation =
  315. strcmp(mod, "16-QAM") == 0 ? 0x01 :
  316. strcmp(mod, "32-QAM") == 0 ? 0x02 :
  317. strcmp(mod, "64-QAM") == 0 ? 0x03 :
  318. strcmp(mod, "128-QAM") == 0 ? 0x04 :
  319. strcmp(mod, "256-QAM") == 0 ? 0x05 : 0x00;
  320. memset(tmp, 0, sizeof(tmp));
  321. pos = 0;
  322. for (i=0;i<strlen(freq);i++) {
  323. if (isdigit(freq[i])) {
  324. tmp[pos] = freq[i];
  325. pos++;
  326. }
  327. }
  328. n->_freq = strtol(tmp, NULL, 16);
  329. memset(tmp, 0, sizeof(tmp));
  330. pos = 0;
  331. for (i=0;i<strlen(symbol_rate);i++) {
  332. if (isdigit(symbol_rate[i])) {
  333. tmp[pos] = symbol_rate[i];
  334. pos++;
  335. }
  336. }
  337. n->_symbol_rate = strtol(tmp, NULL, 16);
  338. return n;
  339. }
  340. void nit_free(NIT **pn) {
  341. NIT *n = *pn;
  342. if (n) {
  343. FREE(n->freq);
  344. FREE(n->modulation);
  345. FREE(n->symbol_rate);
  346. FREE(*pn);
  347. }
  348. }
  349. void proxy_log(INPUT *r, char *msg) {
  350. LOGf("INPUT : [%-12s] %s fd: %d src: %s\n", r->channel->id, msg, r->sock, r->channel->source);
  351. }
  352. void proxy_close(LIST *inputs, INPUT **input) {
  353. proxy_log(*input, "Stop");
  354. // If there are no clients left, no "Timeout" messages will be logged
  355. list_del_entry(inputs, *input);
  356. input_free(input);
  357. }