<commit_before>#include "StreamerLTE.h"
#include "IConnection.h"
#include "LMS64CProtocol.h" //TODO remove when reset usb is abstracted
#include <fstream>
#include <iostream>
#include <ciso646>
#include "fifo.h"

#include "kiss_fft.h"

using namespace std;
using namespace lime;

IConnection* gDataPort;

StreamerLTE::StreamerLTE(IConnection* dataPort)
{
    mDataPort = dataPort;
    mRxFIFO = new LMS_SamplesFIFO(1, 1);
    mTxFIFO = new LMS_SamplesFIFO(1, 1);
    mStreamRunning.store(false);
}

StreamerLTE::~StreamerLTE()
{
    StopStreaming();
    delete mRxFIFO;
    delete mTxFIFO;
}

/** @brief Crates thread for packets processing
*/
StreamerLTE::STATUS StreamerLTE::StartStreaming(const int fftSize, const int channelsCount, const StreamDataFormat format)
{
    if (mStreamRunning.load() == true)
        return FAILURE;
    if (mDataPort->IsOpen() == false)
        return FAILURE;

    stopProcessing.store(false);
    threadProcessing = std::thread(ProcessPackets, this, fftSize, channelsCount, format);
    mStreamRunning.store(true);
    return SUCCESS;
}

/** @brief Stops receiving, processing and transmitting threads
*/
StreamerLTE::STATUS StreamerLTE::StopStreaming()
{
    if (mStreamRunning.load() == false)
        return FAILURE;
    stopProcessing.store(true);
    threadProcessing.join();
    mStreamRunning.store(false);
    return SUCCESS;
}

/** @brief Function dedicated for receiving data samples from board
    @param rxFIFO FIFO to store received data
    @param terminate periodically pooled flag to terminate thread
    @param dataRate_Bps (optional) if not NULL periodically returns data rate in bytes per second
*/
void StreamerLTE::ReceivePackets(const StreamerLTE_ThreadData &args)
{
    auto dataPort = args.dataPort;
    auto rxFIFO = args.FIFO;
    auto terminate = args.terminate;
    auto dataRate_Bps = args.dataRate_Bps;
    auto report = args.report;
    auto getCmd = args.getCmd;

    //at this point Rx must be enabled in FPGA
    unsigned long rxDroppedSamples = 0;
    const int channelsCount = rxFIFO->GetChannelsCount();
    uint32_t samplesCollected = 0;
    auto t1 = chrono::high_resolution_clock::now();
    auto t2 = chrono::high_resolution_clock::now();

    const int bufferSize = 65536;
    const int buffersCount = 16; // must be power of 2
    const int buffersCountMask = buffersCount - 1;
    int handles[buffersCount];
    memset(handles, 0, sizeof(int)*buffersCount);
    vector<char>buffers;
    try
    {
        buffers.resize(buffersCount*bufferSize, 0);
    }
    catch (const std::bad_alloc &ex)
    {
        printf("Error allocating Rx buffers, not enough memory\n");
        return;
    }

    //temporary buffer to store samples for batch insertion to FIFO
    PacketFrame tempPacket;
    tempPacket.Initialize(channelsCount);

    for (int i = 0; i<buffersCount; ++i)
        handles[i] = dataPort->BeginDataReading(&buffers[i*bufferSize], bufferSize);

    int bi = 0;
    unsigned long totalBytesReceived = 0; //for data rate calculation
    int m_bufferFailures = 0;
    int16_t sample;

    uint32_t samplesReceived = 0;

    bool currentRxCmdValid = false;
    RxCommand currentRxCmd;
    size_t ignoreTxLateCount = 0;

    while (terminate->load() == false)
    {
        if (ignoreTxLateCount != 0) ignoreTxLateCount--;
        if (dataPort->WaitForReading(handles[bi], 1000) == false)
            ++m_bufferFailures;

        long bytesToRead = bufferSize;
        long bytesReceived = dataPort->FinishDataReading(&buffers[bi*bufferSize], bytesToRead, handles[bi]);
        if (bytesReceived > 0)
        {
            if (bytesReceived != bufferSize) //data should come in full sized packets
                ++m_bufferFailures;

            totalBytesReceived += bytesReceived;
            for (int pktIndex = 0; pktIndex < bytesReceived / sizeof(PacketLTE); ++pktIndex)
            {
                PacketLTE* pkt = (PacketLTE*)&buffers[bi*bufferSize];
                tempPacket.first = 0;
                tempPacket.timestamp = pkt[pktIndex].counter;

                int statusFlags = 0;
                uint8_t* pktStart = (uint8_t*)pkt[pktIndex].data;
                const int stepSize = channelsCount * 3;
                size_t numPktBytes = sizeof(pkt->data);

                auto byte0 = pkt[pktIndex].reserved[0];
               /* if ((byte0 & (1 << 3)) != 0 and ignoreTxLateCount == 0)
                {
                    auto reg9 = Reg_read(dataPort, 0x0009);
                    Reg_write(dataPort, 0x0009, reg9 | (1 << 1));
                    Reg_write(dataPort, 0x0009, reg9 & ~(1 << 1));
                    if (report) report(STATUS_FLAG_TX_LATE, pkt[pktIndex].counter);
                    ignoreTxLateCount = 16;
                }*/

                if (report) report(STATUS_FLAG_TIME_UP, pkt[pktIndex].counter);

                if (getCmd)
                {
                    auto numPacketSamps = numPktBytes/stepSize;

                    //no valid command, try to pop a new command
                    if (not currentRxCmdValid) currentRxCmdValid = getCmd(currentRxCmd);

                    //command is valid, and this is an infinite read
                    //so we always replace the command if available
                    //the currentRxCmdValid variable remains true
                    else if (!currentRxCmd.finiteRead) getCmd(currentRxCmd);

                    //no valid command, just continue to the next pkt
                    if (not currentRxCmdValid) continue;

                    //handle timestamp logic
                    if (currentRxCmd.waitForTimestamp)
                    {
                        //the specified timestamp is late
                        //TODO report late condition to the rx fifo
                        //we are done with this current command
                        if (currentRxCmd.timestamp < tempPacket.timestamp)
                        {
                            std::cout << "L" << std::flush;<commit_msg>Uncomment Tx late packets reporting<commit_after>#include "StreamerLTE.h"
#include "IConnection.h"
#include "LMS64CProtocol.h" //TODO remove when reset usb is abstracted
#include <fstream>
#include <iostream>
#include <ciso646>
#include "fifo.h"

#include "kiss_fft.h"

using namespace std;
using namespace lime;

IConnection* gDataPort;

StreamerLTE::StreamerLTE(IConnection* dataPort)
{
    mDataPort = dataPort;
    mRxFIFO = new LMS_SamplesFIFO(1, 1);
    mTxFIFO = new LMS_SamplesFIFO(1, 1);
    mStreamRunning.store(false);
}

StreamerLTE::~StreamerLTE()
{
    StopStreaming();
    delete mRxFIFO;
    delete mTxFIFO;
}

/** @brief Crates thread for packets processing
*/
StreamerLTE::STATUS StreamerLTE::StartStreaming(const int fftSize, const int channelsCount, const StreamDataFormat format)
{
    if (mStreamRunning.load() == true)
        return FAILURE;
    if (mDataPort->IsOpen() == false)
        return FAILURE;

    stopProcessing.store(false);
    threadProcessing = std::thread(ProcessPackets, this, fftSize, channelsCount, format);
    mStreamRunning.store(true);
    return SUCCESS;
}

/** @brief Stops receiving, processing and transmitting threads
*/
StreamerLTE::STATUS StreamerLTE::StopStreaming()
{
    if (mStreamRunning.load() == false)
        return FAILURE;
    stopProcessing.store(true);
    threadProcessing.join();
    mStreamRunning.store(false);
    return SUCCESS;
}

/** @brief Function dedicated for receiving data samples from board
    @param rxFIFO FIFO to store received data
    @param terminate periodically pooled flag to terminate thread
    @param dataRate_Bps (optional) if not NULL periodically returns data rate in bytes per second
*/
void StreamerLTE::ReceivePackets(const StreamerLTE_ThreadData &args)
{
    auto dataPort = args.dataPort;
    auto rxFIFO = args.FIFO;
    auto terminate = args.terminate;
    auto dataRate_Bps = args.dataRate_Bps;
    auto report = args.report;
    auto getCmd = args.getCmd;

    //at this point Rx must be enabled in FPGA
    unsigned long rxDroppedSamples = 0;
    const int channelsCount = rxFIFO->GetChannelsCount();
    uint32_t samplesCollected = 0;
    auto t1 = chrono::high_resolution_clock::now();
    auto t2 = chrono::high_resolution_clock::now();

    const int bufferSize = 65536;
    const int buffersCount = 16; // must be power of 2
    const int buffersCountMask = buffersCount - 1;
    int handles[buffersCount];
    memset(handles, 0, sizeof(int)*buffersCount);
    vector<char>buffers;
    try
    {
        buffers.resize(buffersCount*bufferSize, 0);
    }
    catch (const std::bad_alloc &ex)
    {
        printf("Error allocating Rx buffers, not enough memory\n");
        return;
    }

    //temporary buffer to store samples for batch insertion to FIFO
    PacketFrame tempPacket;
    tempPacket.Initialize(channelsCount);

    for (int i = 0; i<buffersCount; ++i)
        handles[i] = dataPort->BeginDataReading(&buffers[i*bufferSize], bufferSize);

    int bi = 0;
    unsigned long totalBytesReceived = 0; //for data rate calculation
    int m_bufferFailures = 0;
    int16_t sample;

    uint32_t samplesReceived = 0;

    bool currentRxCmdValid = false;
    RxCommand currentRxCmd;
    size_t ignoreTxLateCount = 0;

    while (terminate->load() == false)
    {
        if (ignoreTxLateCount != 0) ignoreTxLateCount--;
        if (dataPort->WaitForReading(handles[bi], 1000) == false)
            ++m_bufferFailures;

        long bytesToRead = bufferSize;
        long bytesReceived = dataPort->FinishDataReading(&buffers[bi*bufferSize], bytesToRead, handles[bi]);
        if (bytesReceived > 0)
        {
            if (bytesReceived != bufferSize) //data should come in full sized packets
                ++m_bufferFailures;

            totalBytesReceived += bytesReceived;
            for (int pktIndex = 0; pktIndex < bytesReceived / sizeof(PacketLTE); ++pktIndex)
            {
                PacketLTE* pkt = (PacketLTE*)&buffers[bi*bufferSize];
                tempPacket.first = 0;
                tempPacket.timestamp = pkt[pktIndex].counter;

                int statusFlags = 0;
                uint8_t* pktStart = (uint8_t*)pkt[pktIndex].data;
                const int stepSize = channelsCount * 3;
                size_t numPktBytes = sizeof(pkt->data);

                auto byte0 = pkt[pktIndex].reserved[0];
                if ((byte0 & (1 << 3)) != 0 and ignoreTxLateCount == 0)
                {
                    auto reg9 = Reg_read(dataPort, 0x0009);
                    Reg_write(dataPort, 0x0009, reg9 | (1 << 1));
                    Reg_write(dataPort, 0x0009, reg9 & ~(1 << 1));
                    if (report) report(STATUS_FLAG_TX_LATE, pkt[pktIndex].counter);
                    ignoreTxLateCount = 16;
                }

                if (report) report(STATUS_FLAG_TIME_UP, pkt[pktIndex].counter);

                if (getCmd)
                {
                    auto numPacketSamps = numPktBytes/stepSize;

                    //no valid command, try to pop a new command
                    if (not currentRxCmdValid) currentRxCmdValid = getCmd(currentRxCmd);

                    //command is valid, and this is an infinite read
                    //so we always replace the command if available
                    //the currentRxCmdValid variable remains true
                    else if (!currentRxCmd.finiteRead) getCmd(currentRxCmd);

                    //no valid command, just continue to the next pkt
                    if (not currentRxCmdValid) continue;

                    //handle timestamp logic
                    if (currentRxCmd.waitForTimestamp)
                    {
                        //the specified timestamp is late
                        //TODO report late condition to the rx fifo
                        //we are done with this current command
                        if (currentRxCmd.timestamp < tempPacket.timestamp)
                        {
                            std::cout << "L" << std::flush;
{'fraction_non_alphanumeric': 0.0935, 'fraction_numerical': 0.013166666666666667, 'mean_word_length': 2.334537371492081, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6062355', 'n_tokens_mistral': 3638, 'n_tokens_neox': 3371, 'n_words': 939}