Software Público Brasileiro

#include <stdio.h>
#include <iostream>
#include "dprintf.h"
#include "synchronizer.h"
//#include "httpstreamer.h"
#include "jmutex.h"
#define VIDEO_PADRAO "vlibras_user/dicionario_libras/poseneutra.ts"
#define VIDEOTRANSP_PADRAO "vlibras_user/dicionarioTransp_libras/poseneutra.ts"

//DICIONARIO 45 fps
#define PCRPID 4097
#define PESPID 4113


//DICIONARIO 30 fps
//#define PCRPID 256
//#define PESPID 256

Mutex* mutexi = new Mutex();

Synchronizer::Synchronizer() {
    baseDir = (char*) "vlibras_user/dicionario_libras/";
    extensao = (char*) ".ts";
    filefd = 0;
    videos = new vector <char*> ();
    vector_pts = new vector <int64_t > ();

    this->baseDir = new char[strlen(baseDir) + 1];
    strcpy(this->baseDir, baseDir);

    this->extensao = new char[strlen(extensao) + 1];
    strcpy(this->extensao, extensao);
    flagPrimeiroArquivo = 1;
    ouvintes = new list<OuvinteSynchronizer*>();
    this->path_libras = path_libras;
    DPRINTF("Done!\n")
}

Synchronizer::Synchronizer(/*int64_t pcr_inicial, */char* baseDir, char* extensao, char* path_libras, int transparency) {
    
    hastext = true;
    count_tasks = 0; // Usado no teste de Eduardo!
    service_running = true;

    this->pcr_inicial = pcr_inicial;
    this->pcr_inicial = -1;
    this->tempo_poseneutra = 120000;
	
    this->transparency = transparency;

    //cout << "\n\nTransparencia: " << this->transparency << "\n";


    this->baseDir = new char[strlen(baseDir) + 1];
    strcpy(this->baseDir, baseDir);

    //cout << "\n\n#######BASEDIR: " << this->baseDir << "\n";

    this->extensao = new char[strlen(extensao) + 1];
    strcpy(this->extensao, extensao);

    filefd = 0;
    videos = new vector <char*> ();
    pcrBaseGeral = 0;
    ultimoPCRBase = 0;
    mediaDiferencaPCR = 0;
    contagemPCR = 0;
    vector_pts = new vector <int64_t > ();
    flagPrimeiroArquivo = 1;
    ouvintes = new list<OuvinteSynchronizer*>();
    this->path_libras = new char[strlen(path_libras) + 1];
    strcpy(this->path_libras, path_libras);
    //cout << "-> path libras sync: " << path_libras << endl;
    this->mutex_sync = new Mutex();

    DPRINTF("Done!\n")

}

Synchronizer::~Synchronizer() {
    free(videos);
    ouvintes->clear();
    delete ouvintes;
    if (escreveAux.is_open()) escreveAux.close();
    DDDPRINTF("Synchronized finalized!\n");

}

void Synchronizer::notifyPCRBase(uint64_t pcr_base){
    this->pcr_inicial =  pcr_base;
}

string toLower(string strr) {
    char str[100];
    string ret;
    strcpy(str, strr.c_str());
    int differ = 'A' - 'a';
    char ch;
    int ii = strlen(str);
    for (int i = 0; i < ii; i++) {
        strncpy(&ch, str + i, 1);
        if (ch >= 'A' && ch <= 'Z') {
            ch = ch - differ;
            memcpy(str + i, &ch, 1);
        }
    }
    ret = str;
    return ret;
}


void Synchronizer::setPCRBase(uint64_t pcr_base){
    this->pcr_inicial = pcr_base;
}


char* converteString(string palavra) {

    const char* aux = new char[palavra.size()];
    aux = palavra.c_str();
    char * aux2 = (char*) aux;

    return aux2;
}




/* Método utilizado pelos serviços (Services Generation) para informar que 
   este processo pode ser finalizado após a demanda atual, isto é, depois
   que terminar de gerar os vídeos que estão na fila.
 */
void Synchronizer::stop() {
    //printf("> O Synchronizer recebeu a mensagem de que pode terminar a execução\n");
    service_running = false;
}

int Synchronizer::gettaxa() {
    return taxa;
}

void Synchronizer::settaxa(int taxa) {
    this->taxa = taxa;
}

vector <char*> * Synchronizer::getVideos() {
    return videos;
}

/* Trata a glosa de modo single, ou seja, havendo sinal ou soletrando,
   este método é chamado individualmente em cada sinal correspondente.
 */
void Synchronizer::recebeglosa(char * video, int tam, int64_t pts) {
    mutexi->Lock();
    char *video_copy = new char [tam + 1];
    strcpy(video_copy, video);
    videos->push_back(video_copy);
    count_tasks++;
    vector_pts->push_back(pts);
    mutexi->Unlock();
}

int64_t Synchronizer::getMilisegundos() {
    struct timeval tempo;
    gettimeofday(&tempo, NULL);
    return (int64_t) tempo.tv_sec * 1000 + tempo.tv_usec / 1000;
}

void Synchronizer::msleep(long tsleep, long *desconto) {
    struct timeval tempo;
    long deltaTAux = 0;
    long tusleep = 0;
    int64_t tAux = 0, tAux2 = 0;

    tsleep -= *desconto;

    tusleep = tsleep / 1000;

    gettimeofday(&tempo, NULL);
    tAux = tempo.tv_sec * 1000 + tempo.tv_usec / 1000;

    while ((tsleep > 0) && (deltaTAux <= tsleep)) {
        usleep(tusleep);

        gettimeofday(&tempo, NULL);
        tAux2 = tempo.tv_sec * 1000 + tempo.tv_usec / 1000;

        deltaTAux = (tAux2 - tAux)*1000;
    }

    //desconto = (deltaTAux - (tsleep/1000))*1000;
    *desconto = (deltaTAux - tsleep);
}

char* proximo = new char[1000];
char* proximo2 = new char[1000];

char * Synchronizer::getproximoarquivo() {
    char * video = new char[1000];
    string s1, s2, s3;
    video = NULL;
    list<char *>::iterator it;
    //proximo = NULL;

    if (videos->size() == 0) {
	if(this->transparency == 0)
        	return VIDEO_PADRAO;
	else
		return VIDEOTRANSP_PADRAO;
    }

    mutexi->Lock();
    video = videos->front();
    videos->erase(videos->begin());
    mutexi->Unlock();

    if (video == NULL) {
        if(this->transparency == 0)
        	return VIDEO_PADRAO;
	else
		return VIDEOTRANSP_PADRAO;

    } else {	
        return video;
    }

}


void Synchronizer::analisaPCR(unsigned char *pacote) {
    // preciso ter o PID de PCR
    if ((getPid(pacote) != PCRPID) || (!contemAF(pacote))
            //if ((!contemAF(pacote))
            || (getTamanhoAF(pacote) == 0) || (!contemPCR(pacote))) {

      
      if (contemPCR(pacote)){
	  	//printf("Encontrei PCR - PID:%d\n", getPid(pacote));

      }
        return;
    }

    uint64_t pcrBase = retornaPCRBase(pacote);

    if (mediaDiferencaPCR != 0 && contagemPCR == 0) { // nesse caso esse eh o primeiro PCR de um novo video
	//printf("Encontrei o primeiro PCR de um video!\n");
        setPCRBase(pacote, pcrBaseGeral + mediaDiferencaPCR); // tem que multiplicar por 2 o mediaDiferencaPCR?
        ultimoPCRBaseMod = pcrBaseGeral + mediaDiferencaPCR;
        mediaDiferencaPCR = 0;
    } else if (mediaDiferencaPCR == 0 && contagemPCR == 0 && flagPrimeiroArquivo == 1) {
        flagPrimeiroArquivo = 0;
        setPCRBase(pacote, pcr_inicial + mediaDiferencaPCR); // tem que multiplicar por 2 o mediaDiferencaPCR?
        contagemPCR++;
        sleep(1);
        ultimoPCRBaseMod = pcr_inicial + mediaDiferencaPCR;

    }
    else {

        mediaDiferencaPCR = (ultimoPCRBase > 0) ? pcrBase - ultimoPCRBase : 0;
        contagemPCR++;

        //if (pcrBaseGeral != 0) { // se nao for o primeiro video  //2 - Mudei para o serviço
        setPCRBase(pacote, ultimoPCRBaseMod + mediaDiferencaPCR);
        ultimoPCRBaseMod += mediaDiferencaPCR;
        //}
    }

    ultimoPCRBase = pcrBase;

}

void Synchronizer::analisaPTSDTS(unsigned char *pacote) {


    int inicioPESData = 4;
    if (contemAF(pacote)) inicioPESData += getTamanhoAF(pacote) + 1;

    if ((!verificaPESStartCode(pacote + inicioPESData)) ||
            (!verificaPESStreamID(pacote + inicioPESData)) ||
            (!contemPTS(pacote + inicioPESData))) {

        return;
    }

    uint64_t PTS = retornaPTS(pacote + inicioPESData);
    int diferenca = PTS - ultimoPCRBase;
    // confirmar se isso aqui SEMPRE eh chamado depois de pelo menos um PCR ser encontrado, principalmente
    // no inicio do video
    if (ultimoPCRBaseMod != 0)
        setNovoPTS(pacote + inicioPESData, ultimoPCRBaseMod + diferenca);
    else
        setNovoPTS(pacote + inicioPESData, pcrBaseGeral + diferenca);

    if (!contemDTS(pacote + inicioPESData)) return;

    uint64_t DTS = retornaDTS(pacote + inicioPESData);
    diferenca = DTS - ultimoPCRBase;

    if (ultimoPCRBaseMod != 0)
        setNovoDTS(pacote + inicioPESData, ultimoPCRBaseMod + diferenca);
    else
        setNovoDTS(pacote + inicioPESData, pcrBaseGeral + diferenca);
}

int Synchronizer::getPid(unsigned char *pacote) {
    return ((pacote[1] & 0x1F) << 8) | pacote[2];
}

bool Synchronizer::contemAF(unsigned char *pacote) {
    return pacote[3] & 0x20;
}

int Synchronizer::getTamanhoAF(unsigned char *pacote) {
    return pacote[4];
}

bool Synchronizer::contemPCR(unsigned char *pacote) {
    return pacote[5] & 0x10;
}

uint64_t Synchronizer::retornaPCRBase(unsigned char *pacote) {
    uint64_t pcrbase = 0;

    pcrbase = (uint64_t) pacote[6] << 25 | pacote[7] << 17 | pacote[8] << 9
            | pacote[9] << 1 | (pacote[10] & 0x80) >> 7;

    return pcrbase;
}

void Synchronizer::setPCRBase(unsigned char *pacote, uint64_t pcrbase) {
    pacote[6] = pcrbase >> 25;
    pacote[7] = pcrbase >> 17;
    pacote[8] = pcrbase >> 9;
    pacote[9] = pcrbase >> 1;
    if ((pcrbase << 63) != 0)
        pacote[10] = pacote[10] | 0x80;
    else
        pacote[10] = pacote[10] & 0x7F;
}

bool Synchronizer::contemPayloadStartIndicator(unsigned char *pacote) {
    return pacote[1] & 0x40;
}

bool Synchronizer::contemPayloadData(unsigned char *pacote) {
    return pacote[3] & 0x10;
}

bool Synchronizer::verificaPESStartCode(unsigned char *PESPacket) {
    return (((PESPacket[0] << 16) | (PESPacket[1] << 8) | (PESPacket[2]))
            == 0x000001) ? 1 : 0;
}

bool Synchronizer::verificaPESStreamID(unsigned char *PESPacket) {
    int stream_id = PESPacket[3];

    return ((stream_id != 0xBC) && (stream_id != 0xBE) && (stream_id != 0xBF)
            && (stream_id != 0xF0) && (stream_id != 0xF1) && (stream_id != 0xFF)
            && (stream_id != 0xF2) && (stream_id != 0xF8));
}

bool Synchronizer::contemPTS(unsigned char *PESPacket) {
    //printf("\nTem PTS: %d\n", PESPacket[7] & 0x80);
    return PESPacket[7] & 0x80;
}

bool Synchronizer::contemDTS(unsigned char *PESPacket) {
    return PESPacket[7] & 0x40;
}

uint64_t Synchronizer::retornaPTS(unsigned char *PESPacket) {
    uint64_t PTS = (uint64_t) ((PESPacket[9] & 0x0E) >> 1) << 30 |
            (PESPacket[10] << 22) |
            (((PESPacket[11] & 0xFE) >> 1) << 15) |
            (PESPacket[12] << 7) |
            ((PESPacket[13] & 0xFE) >> 1);
    return PTS;
}

uint64_t Synchronizer::retornaDTS(unsigned char *PESPacket) {
    uint64_t DTS = (uint64_t) ((PESPacket[14] & 0x0E) >> 1) << 30 |
            (PESPacket[15] << 22) |
            (((PESPacket[16] & 0xFE) >> 1) << 15) |
            (PESPacket[17] << 7) |
            ((PESPacket[18] & 0xFE) >> 1);
    return DTS;
}

void Synchronizer::setNovoPTS(unsigned char *pacote, uint64_t PTS) {
    pacote[9] = pacote[9] & 0xF1;
    pacote[9] = pacote[9] | ((PTS >> 30) << 1);
    pacote[10] = PTS >> 22;
    pacote[11] = pacote[11] & 0x1;
    pacote[11] = pacote[11] | ((PTS >> 15) << 1);
    pacote[12] = PTS >> 7;
    pacote[13] = pacote[13] & 0x1;
    pacote[13] = pacote[13] | PTS << 1;
}

void Synchronizer::setNovoDTS(unsigned char *pacote, uint64_t DTS) {
    pacote[14] = pacote[14] & 0xF1;
    pacote[14] = pacote[14] | ((DTS >> 30) << 1);
    pacote[15] = DTS >> 22;
    pacote[16] = pacote[16] & 0x1;
    pacote[16] = pacote[16] | ((DTS >> 15) << 1);
    pacote[17] = DTS >> 7;
    pacote[18] = pacote[18] & 0x1;
    pacote[18] = pacote[18] | DTS << 1;
}

void Synchronizer::escrevePacote(char *pacote) {
    if (!escreveAux.is_open()) {
        escreveAux.open(this->path_libras, ios::out | ios::binary);

    }
    //printf("tamanho do pacote = %d\n\n", strlen(pacote));
    escreveAux.write(pacote, 188*70);
    //printf(" OK!\n\n");

}

int Synchronizer::escrevePacote(int * descritor, char *pacote, int size) {
    if (*descritor < 0) {// ainda não abri
        *descritor = open(this->path_libras, O_WRONLY | O_CREAT, 0660);
        if (descritor < 0) {
            printf("Can't open output file\n");
            return -1;
        }   
    }
    return write(*descritor, pacote, size);
}

void ouch(int sig) {
    printf("\nSinal Capturado %d", sig);
    (void) signal(SIGPIPE, SIG_DFL);

}

void Synchronizer::soletraGlosa(string glosa, int64_t pts) {

    //string path = (string) baseDir + glosa + (string) extensao;
    //cout << "\n***Glosa NAO ENCONTRADA: " << glosa << endl;
    /*mutexi->Lock();
    int file = open(converteString(path), O_RDONLY);
    mutexi->Unlock();*/
    //if (file != NULL) {
        for (int i = 0; i < glosa.length(); i++) {
            if (glosa[i] == '[')
                return;
            char c = glosa[i];
            if (c >= 97 && c <= 122) {
                //cout << "soletrando a letra: " << glosa[i] << endl;
                string path2 = (string) baseDir + glosa[i] + (string) extensao;
                char* path2_char = converteString(path2);
                recebeglosa(converteString(path2), strlen(path2_char), pts);
            } 
        }
        //close(file);
    //} else {
    //    printf("*** ERRO ao abrir o arquivo.");
    //}

}

/* Utilizado no teste de Eduardo */
void Synchronizer::recebeglosa(string glosa, int64_t pts) {
    glosa = toLower(glosa);
    
    size_t found;
    found=glosa.rfind("dl");
    if (found == 0)
	glosa = glosa.substr(2, glosa.length());

    string path = (string) baseDir + glosa.c_str() + (string) extensao;
    char* urlchar = converteString(path);
    mutexi->Lock();
    FILE *f = fopen(urlchar, "r");
    mutexi->Unlock();
    if (f == NULL) {
        soletraGlosa(glosa, pts);
    } else {
        fclose(f);
        recebeglosa(urlchar, strlen(urlchar), pts);
    }

}


void Synchronizer::registraOuvinte(OuvinteSynchronizer * ouvinte) {
    ouvintes->push_back(ouvinte);
}

void Synchronizer::removeOuvinte(OuvinteSynchronizer * ouvinte) {
    ouvintes->remove(ouvinte);
}

void Synchronizer::notificaOuvintes() {
    for (list<OuvinteSynchronizer*>::iterator i = ouvintes->begin(); i != ouvintes->end(); i++) {
        (*i)->finalizouSincronizacao();
    }
}

void Synchronizer::Run() {

    pcrBaseGeral = ultimoPCRBase = mediaDiferencaPCR = contagemPCR = ultimoPCRBaseMod = 0;
    int64_t t0 = 0, tf = 0;
    int64_t pts_run;
    int cnt, finalcnt, bitrate = 1121600, deltaT = 0;
    int bufferSize = 7 * 188;
    char * buff = new char[bufferSize];

    char * arquivo;

    while (videos->size() == 0)
        usleep(50000); //50ms

    if(this->transparency == 0)
        arquivo = VIDEO_PADRAO;
    else
        arquivo = VIDEOTRANSP_PADRAO;

    mutexi->Lock();
    filefd = open(arquivo, O_RDONLY);
    mutexi->Unlock();
    
    if (filefd < 0)
        cout << "NAO CONSEGUI ABRIR O ARQUIVO2 - " << arquivo << endl;

    (void) signal(SIGPIPE, ouch);

    int videos_processed = 0;    

    int desc_out = -1;   

    while (videos_processed <= count_tasks || service_running) {
        cnt = read(filefd, buff, bufferSize);    

        // atualiza os valores de PCR e PTS/DTS para cada video
        for (int i = 0; i < cnt / 188; i++) {
            analisaPCR((unsigned char*) buff + (188 * i));
            analisaPTSDTS((unsigned char*) buff + (188 * i));
        }

        escrevePacote(&desc_out,buff,cnt);
        free(buff);
        buff = new char[bufferSize]; 

        finalcnt = cnt;
        if (finalcnt == 0) {   

            while (videos->size() == 0 && service_running) { // esperando arquivos do serviço
                usleep(10000);            
            }

            pts_run = vector_pts->front();
            if (ultimoPCRBaseMod + tempo_poseneutra < pts_run) {
                if(this->transparency == 0){
                    arquivo = VIDEO_PADRAO;
                }else{
                    arquivo = VIDEOTRANSP_PADRAO;
                }

             }else {
                if (vector_pts->size() > 0) {
                    vector_pts->erase(vector_pts->begin());
                    arquivo = getproximoarquivo();
                }
                videos_processed++;
            }

            close(filefd);
            filefd = open(arquivo, O_RDONLY);
            if (ultimoPCRBase != 0) {
                if (ultimoPCRBaseMod != 0) {
                    pcrBaseGeral = ultimoPCRBaseMod;
                } else {
                    pcrBaseGeral = ultimoPCRBase;
                }
                contagemPCR = 0;
                ultimoPCRBaseMod = 0;
            }          
            
        
        } else if (finalcnt == -1) {
            printf("Erro: Finalizando Sincronizador!\n");
            this->stop();
        }     
    } // Fim do while!

    //fsync(desc_out);
    close(filefd);
    close(desc_out);
    notificaOuvintes();
    
}