Write the error message to log.

[ruwai.git] / software / c++ / ruwai / src / recorder.cpp

/*--------------------------------------------------------------------------*/
// LICENSE
//
// This file is part of ruwai.
//
// If you use ruwai_parser in any program or publication, please inform and
// acknowledge its author Stefan Mertl (stefan@mertl-research.at).
//
// ruwai is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.
/*--------------------------------------------------------------------------*/

#include "recorder.h"
#include <bitset>


Recorder::Recorder(std::string serial_number, unsigned int n_channels, unsigned int sps, std::string sd_mnt, unsigned int file_length)
{
    this->serial_number = serial_number;
    this->sps = sps;
    this->n_channels = n_channels;
    this->write_limit = sps;
    this->sd_mnt_point = sd_mnt;
    tow_sub_ms_scale = std::pow(2, -32);


    // Create one vector for each channel.
    for (unsigned int i = 0; i < n_channels; i++)
    {
        sample_buffer.push_back(std::vector<std::int32_t>());
        timestamp_pos.push_back(std::vector<int>());
        timestamps.push_back(std::vector<timestamp_t>());

        MSTrace* mst_template = mst_init(NULL);
        strcpy(mst_template->network, "XX");
        strcpy(mst_template->station, this->serial_number.c_str());
        strcpy(mst_template->location, "00");
        mst_template->dataquality = 'D';
        mst_template->sampletype = 'i';
        mst_template->samprate = this->sps;
        mst_template->samplecnt = 0;
        mst_template->numsamples = 0;
        strcpy(mst_template->channel, (str(boost::format("%02d") % (i + 1)).c_str()));

        // Initialize the MSRecord template and save it in the prvtptr.
        MSRecord *msr_template = msr_init(NULL);        // The MiniSeed Record used for adding sample data to the trace group.
        init_msr(msr_template, i + 1);                  // Create the blockettes and set header fields.  
        mst_template->prvtptr = msr_template;

        traces.push_back(mst_template);
    }

    // Initialize the miniseed trace group.
    //mstg = mst_initgroup(NULL);

    // Create the output directory string.
    data_dir = sd_mnt_point + "/" + "mseed";

    // Initialize the recorder state.
    this->file_length = file_length;

    state_gps_fix_init = false;
    //state_gps_fix = FIX_NO;
    //state_gps_fix_ok = false;
    sec_slts_cnt = 0;
    last_timestamp.time = 0;
    last_timestamp.sec_slts = 0;
    last_timestamp.utc_available = false;
    last_timestamp.utc_timebase = false;
    last_timestamp.gps_fix_ok = false;
    last_timestamp.gps_fix = FIX_NO;
}


bool
Recorder::check_output_dir_structure(void)
{
    // Check for the miniseed output folder. Create it if it doesn't exist.
    if (!boost::filesystem::exists(data_dir))
    {
        boost::filesystem::create_directory(data_dir);

    }

    /*
    std::string recorder_dir = data_dir + "/" + this->serial_number;
    if (!boost::filesystem::exists(recorder_dir))
    {
        boost::filesystem::create_directory(recorder_dir);
    }
    */
    return true;
}

void
Recorder::add_sample(rw_smp_24bit_t paket)
{
    mut_sample_buffer.lock();
    for (unsigned int i = 0; i < sample_buffer.size(); i++)
    {
        sample_buffer[i].push_back(paket.samples[i]);

    }
    mut_sample_buffer.unlock();
}


void
Recorder::add_timestamp_sample(rw_smp_24bit_tsp_t paket)
{
    syslog(LOG_DEBUG, "[add_timestamp_sample] Adding a timestamp sample.");
    bool set_system_time = false;

    // Add the samples to the sample buffer.
    mut_sample_buffer.lock();
    mut_timestamps.lock();
    for (unsigned int i = 0; i < sample_buffer.size(); i++)
    {
        sample_buffer[i].push_back(paket.samples[i]);
        timestamp_pos[i].push_back(sample_buffer[i].size() - 1);

    }
    mut_timestamps.unlock();
    mut_sample_buffer.unlock();

    // Convert the GPS timestamp to hptime used by libmseed.
    // GPS time starts on 1980-01-06 UTC.
    // Unix epoch time starts on 1970-01-01 UTC.
    // The difference between GPS time and Unix epoch time is 315964800
    // seconds.
    timestamp_t cur_timestamp;
    cur_timestamp.utc_available = (bool)paket.flags.utc;
    cur_timestamp.utc_timebase = (bool)paket.flags.timebase;
    cur_timestamp.gps_fix_ok = (bool)paket.flags.gps_fix_ok;
    cur_timestamp.gps_fix = (gpsfix_t)paket.flags.gps_fix;
    cur_timestamp.sec_slts = paket.sec_slts;

    //std::cout << "week: " << paket.week << "," << "tow_ms: " << paket.tow_ms << "," << "tow_sub_ms: " << paket.tow_sub_ms  << ", sec_slts: " << paket.sec_slts << std::endl;
    syslog(LOG_INFO, "[add_timestamp_sample] paket time info:    week: %d, tow_ms: %d, tow_sub_ms: %d, sec_slts: %d", paket.week, paket.tow_ms, paket.tow_sub_ms, paket.sec_slts);
    double gps_timestamp = (double)paket.week * (double)604800 + (double)paket.tow_ms / (double)1000 + (double)paket.tow_sub_ms/(double)1000 * tow_sub_ms_scale;
    double tow_sub_ms = (double)paket.tow_sub_ms/(double)1000 * tow_sub_ms_scale;

    // Add the seconds since the last timestamp.
    gps_timestamp += paket.sec_slts;

    syslog(LOG_DEBUG, "[add_timestamp_sample] scaled tow_sub_ms: %f", tow_sub_ms);
    syslog(LOG_DEBUG, "[add_timestamp_sample] gps_timestamp: %f", gps_timestamp);
    //std::cout << "tow_sub_ms: " << str(boost::format("%f") % tow_sub_ms) << std::endl;
    //std::cout << "gps_timestamp: " << str(boost::format("%f") % gps_timestamp) << std::endl;


    // Correct the timestamp with the ADC group delay.
    // TODO: Handle the group delay for high-resolution mode (39/f_data).
    gps_timestamp -= 38/(float)sps;

    double epoch_timestamp;
    if(cur_timestamp.utc_timebase)
    { 
        epoch_timestamp = gps_timestamp + 315964800;
    }
    else 
    {
        //TODO: Handle the GPS timebase.
        epoch_timestamp = 0;
    }


    if (epoch_timestamp >= 0)
    {
        cur_timestamp.time = MS_EPOCH2HPTIME(epoch_timestamp);
        mut_timestamps.lock();
        for (unsigned int k = 0; k < timestamps.size(); k++)
        {
            timestamps[k].push_back(cur_timestamp);
            syslog(LOG_DEBUG, "[add_timestamp_sample] sample_buffer[%d].size(): %d", k, (int)sample_buffer[k].size());
            syslog(LOG_DEBUG, "[add_timestamp_sample] Added timestamp_pos: %d", (int)timestamp_pos[k].back());
            //std::cout << "sample_buffer[" << k << "].size(): " << sample_buffer[k].size() << "\n";
        }
        mut_timestamps.unlock();
    }

    char time_string[27];
    syslog(LOG_INFO, "[add_timestamp_sample] timestamp: %s", ms_hptime2isotimestr(cur_timestamp.time, time_string, true));
    syslog(LOG_INFO, "[add_timestamp_sample] hp_time: %lld", (long long int)cur_timestamp.time);
    syslog(LOG_INFO, "[add_timestamp_sample] gps_fix_ok: %d", cur_timestamp.gps_fix_ok);
    syslog(LOG_INFO, "[add_timestamp_sample] gps_fix: %d", cur_timestamp.gps_fix);
    syslog(LOG_INFO, "[add_timestamp_sample] utc_available: %d", cur_timestamp.utc_available);
    syslog(LOG_INFO, "[add_timestamp_sample] utc_timebase: %d", cur_timestamp.utc_timebase);

    // Check the time fix state and log any changes.
    if (!state_gps_fix_init)
    {
        syslog(LOG_NOTICE, "[add_timestamp_sample] GPS_FIX state initialized: timestamp = %s, gps_fix = %d, gps_fix_ok = %d, utc_available = %d", ms_hptime2isotimestr(cur_timestamp.time, time_string, true), cur_timestamp.gps_fix, cur_timestamp.gps_fix_ok, cur_timestamp.utc_available);
        state_gps_fix_init = true;
    }
    else
    {
        if (cur_timestamp.gps_fix != last_timestamp.gps_fix)
        {
            syslog(LOG_NOTICE, "[add_timestamp_sample] GPS_FIX changed: timestamp = %s, gps_fix = %d, gps_fix_ok = %d, utc_available = %d", ms_hptime2isotimestr(cur_timestamp.time, time_string, true), cur_timestamp.gps_fix, cur_timestamp.gps_fix_ok, cur_timestamp.utc_available);
        }

        if (cur_timestamp.gps_fix_ok != last_timestamp.gps_fix_ok)
        {
            syslog(LOG_NOTICE, "[add_timestamp_sample] GPS_FIX_OK changed: timestamp = %s, gps_fix = %d, gps_fix_ok = %d, utc_available = %d", ms_hptime2isotimestr(cur_timestamp.time, time_string, true), cur_timestamp.gps_fix, cur_timestamp.gps_fix_ok, cur_timestamp.utc_available);
        }

        if (cur_timestamp.utc_available != last_timestamp.utc_available)
        {
            syslog(LOG_NOTICE, "[add_timestamp_sample] UTC_AVAILABLE changed: timestamp = %s, gps_fix = %d, gps_fix_ok = %d, utc_available = %d", ms_hptime2isotimestr(cur_timestamp.time, time_string, true), cur_timestamp.gps_fix, cur_timestamp.gps_fix_ok, cur_timestamp.utc_available);
        }

        hptime_t ts_diff = cur_timestamp.time - last_timestamp.time;
        if (ts_diff < 0)
        {
            syslog(LOG_WARNING, "[WARNING][add_timestamp_sample] Negative timestamp difference of %lld us detected. current: %s, last: %s", (long long int)ts_diff, ms_hptime2isotimestr(cur_timestamp.time, time_string, true), ms_hptime2isotimestr(last_timestamp.time, time_string, true));
            set_system_time = true;
        }
        else if (ts_diff > 1000000)
        {
            syslog(LOG_WARNING, "[WARNING][add_timestamp_sample] Positive timestamp difference of %lld us detected. current: %s, last: %s", (long long int)ts_diff, ms_hptime2isotimestr(cur_timestamp.time, time_string, true), ms_hptime2isotimestr(last_timestamp.time, time_string, true));
            set_system_time = true;
        }
        else if (ts_diff == 0) 
        {
            syslog(LOG_WARNING, "[WARNING][add_timestamp_sample] Equal timestamp difference of %lld us detected. current: %s, last: %s", (long long int)ts_diff, ms_hptime2isotimestr(cur_timestamp.time, time_string, true), ms_hptime2isotimestr(last_timestamp.time, time_string, true));
            set_system_time = true;
        }
        else if (ts_diff > 0 && ts_diff < 1000000) 
        {
            syslog(LOG_WARNING, "[WARNING][add_timestamp_sample] Timestamp difference between 0 and 1 of %lld us detected. current: %s, last: %s", (long long int)ts_diff, ms_hptime2isotimestr(cur_timestamp.time, time_string, true), ms_hptime2isotimestr(last_timestamp.time, time_string, true));
            set_system_time = true;
        }
        else 
        {
            syslog(LOG_INFO, "[add_timestamp_sample] Correct timestamp difference of %lld us detected. current: %s, last: %s", (long long int)ts_diff, ms_hptime2isotimestr(cur_timestamp.time, time_string, true), ms_hptime2isotimestr(last_timestamp.time, time_string, true));
        }
    }

    if (set_system_time)
    {
        int ret_val;
        std::string cmd = "set_system_time " + str(boost::format("%lld") % (long long int)(MS_HPTIME2EPOCH(cur_timestamp.time)));
        syslog(LOG_NOTICE, "[add_timestamp_sample] Setting the system time using the command: %s.", cmd.c_str());
        ret_val = system(cmd.c_str());
        if (ret_val != 0)
        {
            syslog(LOG_WARNING, "[WARNING][add_timestamp_sample] Couldn't set the system time. ret_val = %d; %s", ret_val, strerror(errno));
        }
        else
        {
            syslog(LOG_NOTICE, "[add_timestamp_sample] Successfully set the system time.");
        }
    }

    // Log the seconds since the last timestamp.
    if (cur_timestamp.sec_slts > 0)
    {
        sec_slts_cnt++;
        if (sec_slts_cnt >= 5)
        {
            syslog(LOG_NOTICE, "[add_timestamp_sample] Running without valid timestamp: timestamp = %s, sec_slts = %d", time_string, cur_timestamp.sec_slts);
            sec_slts_cnt = 0;
        }
    }
    else
    {
        sec_slts_cnt = 0;
    }

    last_timestamp = cur_timestamp;

    // Pack the samples to a miniseed file.
    std::thread t_pack_samplebuffer(&Recorder::pack_samplebuffer, this);
    t_pack_samplebuffer.detach();
}


void
Recorder::pack_samplebuffer(void)
{
    syslog(LOG_DEBUG, "[pack_samplebuffer] Packing the samplebuffer to miniseed.");

    auto start = std::chrono::high_resolution_clock::now();
   
    hptime_t last_starttime; 

    // TODO: Copy the sample_buffer, timestamp_pos and timestamps to current
    // working variables. Limit the access to the vectors to a single position
    // protected by the mutex locks.
    // Copy the data until the last valid timestamp. Keep the last timestamp in
    // the original vectors, remove the rest.

    std::vector <std::vector <std::int32_t> > pack_buffer;
    std::vector <std::vector <int> > pack_timestamp_pos;
    std::vector <std::vector <timestamp_t> > pack_timestamps;

    syslog(LOG_DEBUG, "[pack_samplebuffer] Copying from the recording vectors....");
    mut_timestamps.lock();
    for (unsigned int cur_channel = 0; cur_channel < n_channels; cur_channel++)
    {
        syslog(LOG_DEBUG, "[pack_samplebuffer] Processing channel %d.", cur_channel);
        // Copy the timestamps and timestamp_positions until the last
        // timestamp. Keep the last timestamp in the recording vector.
        syslog(LOG_DEBUG, "[pack_samplebuffer] timestamps size: %d", (int)timestamps[cur_channel].size());
        syslog(LOG_DEBUG, "[pack_samplebuffer] timestamp_pos size: %d", (int)timestamp_pos[cur_channel].size());
        syslog(LOG_DEBUG, "[pack_samplebuffer] timestamps_pos end: %d", (int)timestamp_pos[cur_channel].back());
        std::vector <timestamp_t> cur_ts(timestamps[cur_channel].begin(), timestamps[cur_channel].end());
        pack_timestamps.push_back(cur_ts);
        std::vector <std::int32_t> cur_ts_pos(timestamp_pos[cur_channel].begin(), timestamp_pos[cur_channel].end());
        pack_timestamp_pos.push_back(cur_ts_pos);
        timestamps[cur_channel].erase(timestamps[cur_channel].begin(), timestamps[cur_channel].end());
        timestamp_pos[cur_channel].erase(timestamp_pos[cur_channel].begin(), timestamp_pos[cur_channel].end());
        syslog(LOG_DEBUG, "[pack_samplebuffer] pack_timestamps size: %d", (int)pack_timestamps[cur_channel].size());
        syslog(LOG_DEBUG, "[pack_samplebuffer] pack_timestamp_pos size: %d", (int)pack_timestamp_pos[cur_channel].size());
        syslog(LOG_DEBUG, "[pack_samplebuffer] timestamps size after erase: %d", (int)timestamps[cur_channel].size());
        syslog(LOG_DEBUG, "[pack_samplebuffer] timestamp_pos size after erase: %d", (int)timestamp_pos[cur_channel].size());

        // Copy the samples up to the last timestamp_pos from the sample
        // buffer. Remove the copied samples from the sample buffer.
        int last_ts_pos = pack_timestamp_pos[cur_channel].back();
        mut_sample_buffer.lock();
        syslog(LOG_DEBUG, "[pack_samplebuffer] last_ts_pos: %d", last_ts_pos);
        syslog(LOG_INFO, "[pack_samplebuffer] sample_buffer[%d].size(): %d", cur_channel, (int)sample_buffer[cur_channel].size());
        std::vector <std::int32_t> cur_buffer(sample_buffer[cur_channel].begin(), sample_buffer[cur_channel].begin() + last_ts_pos);  // The last element is not included by the copy constructor.
        pack_buffer.push_back(cur_buffer);
        sample_buffer[cur_channel].erase(sample_buffer[cur_channel].begin(), sample_buffer[cur_channel].begin() + last_ts_pos);
        syslog(LOG_INFO, "[pack_samplebuffer] pack_buffer.size(): %d", (int)pack_buffer.size());
        syslog(LOG_INFO, "[pack_samplebuffer] pack_buffer[%d].size(): %d", cur_channel, (int)pack_buffer[cur_channel].size());
        syslog(LOG_INFO, "[pack_samplebuffer] sample_buffer size after erase: %d", (int)sample_buffer[cur_channel].size());
        mut_sample_buffer.unlock();
    }
    mut_timestamps.unlock();
    syslog(LOG_DEBUG, "[pack_samplebuffer] .....finished copying from the recording vectors.");


    // Process the copied data of each channel. 
    for (unsigned int cur_channel = 0; cur_channel < n_channels; cur_channel++)
    {
        syslog(LOG_INFO, "[pack_samplebuffer] Packing channel %d.", cur_channel + 1);
        hptime_t span_start;    // The start time of the timespan to write to file.         
        hptime_t span_end;      // The end time of the timespan to write to file.
        bool flush_trace = false;           // If true, write the whole data in trace to file.
        bool last_data_of_file = false;     // True if the end of the file time-span was reached.
        char time_string[27];               // Used to convert hptime to string.

        std::vector <std::int32_t> rem_buffer = std::vector <std::int32_t>();   // The remaining data if the file time-span was reached.

        MSTrace *cur_mst = traces[cur_channel];
        MSRecord *msr = (MSRecord*) cur_mst->prvtptr;

        // Compute the time of the first sample in the sample_buffer_vector.
        if (pack_timestamps[cur_channel].size() >= 1)
        {
            msr->starttime = pack_timestamps[cur_channel].front().time - (pack_timestamp_pos[cur_channel].front() * 1/(float)sps) * (double)HPTMODULUS; 
        }
        else
        {
            syslog(LOG_ERR, "[ERROR][pack_samplebuffer] No timestamp available. Writing data to miniseed without timestamp is not yet supported.");
            return;
        }

        // Add the timestamp quality information to the record.
        set_msr_timeflags(msr, pack_timestamps[cur_channel]);

        // Compute the time-span limits of the desired file length.
        hptime_t file_length_hp = MS_EPOCH2HPTIME((double)file_length);
        span_start = msr->starttime - (msr->starttime % file_length_hp);
        span_end = span_start + file_length_hp;

        // If the the data spans the split-limit of a file defined above. Write the
        // remaining samples to the miniseed file and keep the rest for the next
        // call of the method. Make sure to set the flag in mst_pack to write the
        // data even though that no complete record is filled.
        if (pack_timestamps[cur_channel].back().time >= span_end)
        {
            int pos_limit;
            pos_limit = pack_timestamp_pos[cur_channel].back();
            syslog(LOG_DEBUG, "[pack_samplebuffer] old pos_limit: %d", pos_limit);
            pos_limit = pos_limit - (pack_timestamps[cur_channel].back().time - span_end) / (double)HPTMODULUS * sps;
            syslog(LOG_DEBUG, "[pack_samplebuffer] new computed pos_limit: %d", pos_limit);

            // Check if the timegap between the span_end and the current
            // timestamp is within the valid data limit. If not, there was a
            // jump in the timestamp (e.g. new GPS fix) and the data before the
            // current timestamp can't be correctly assigned to a new file. 
            if (pos_limit < 0)
            {
                pos_limit = timestamp_pos[cur_channel].back();
                syslog(LOG_DEBUG, "[pack_samplebuffer] fixed zero pos_limit: %d", pos_limit);
            }

            // Copy the remaining samples. Add them to the trace group, after
            // the trace was packed.
            rem_buffer = pack_buffer[cur_channel];
            rem_buffer.erase(rem_buffer.begin(), rem_buffer.begin() + pos_limit);  // The last element is not included by the copy constructor.

            // resize the pack buffer
            pack_buffer[cur_channel].resize(pos_limit);

            flush_trace = true;
            last_data_of_file = true;
        }

        // Fill the MSRecord with the sample buffer data up to the current
        // timestamp.
        msr->datasamples = &pack_buffer[cur_channel][0];
        msr->numsamples = pack_buffer[cur_channel].size();
        msr->samplecnt = pack_buffer[cur_channel].size();

        // Add the MSRecord to the trace of the current channel.
        syslog(LOG_INFO, "[pack_samplebuffer] Adding MSR to the trace.");
        syslog(LOG_INFO, "[pack_samplebuffer] cur_mst->starttime isoformat: %s", ms_hptime2isotimestr(cur_mst->endtime, time_string, true));
        syslog(LOG_INFO, "[pack_samplebuffer] cur_mst->endtime isoformat: %s", ms_hptime2isotimestr(cur_mst->endtime, time_string, true));
        syslog(LOG_INFO, "[pack_samplebuffer] msr->starttime isoformat: %s", ms_hptime2isotimestr(msr->starttime, time_string, true));

        // Check, if the MSRecord can be added to the end of the trace.
        if(msr_fits_to_end(cur_mst, msr))
        {
            // Add the record to the end of the trace.
            syslog(LOG_INFO, "[pack_samplebuffer] Sample gap is ok. Adding the record to the end of the trace.");
            mst_addmsr(cur_mst, msr, 1);
        }
        else
        {
            // The record doesn't fit at the end of the trace.
            // Flush the trace to file and add the record to an empty trace.
            syslog(LOG_WARNING, "[pack_samplebuffer] Sample gap is too large or negative. Flush the trace to file and add the record to an empty trace.");
            write_trace_to_file(cur_mst, true, msr);

            // Reset the datasamples values in the MSRecord.
            cur_mst->datasamples = NULL;
            cur_mst->numsamples = 0;
            cur_mst->samplecnt = 0;

            // Set the new start time of the trace and reset the sequence
            // number.
            cur_mst->starttime = msr->starttime;
            msr->sequence_number = 1;

            mst_addmsr(cur_mst, msr, 1);
            syslog(LOG_NOTICE, "[pack_samplebuffer] mst->starttime: %s", ms_hptime2isotimestr(cur_mst->starttime, time_string, true));
            syslog(LOG_NOTICE, "[pack_samplebuffer] mst->endtime: %s", ms_hptime2isotimestr(cur_mst->endtime, time_string, true));
        }

        // Write the trace to file.
        write_trace_to_file(cur_mst, flush_trace, msr);

        if (last_data_of_file && rem_buffer.size() > 0)
        {
            syslog(LOG_NOTICE, "[pack_samplebuffer] Writing the last data to file."); 
            syslog(LOG_NOTICE, "[pack_samplebuffer] msr->starttime: %s", ms_hptime2isotimestr(msr->starttime, time_string, true));
            syslog(LOG_NOTICE, "[pack_samplebuffer] mst->starttime: %s", ms_hptime2isotimestr(cur_mst->starttime, time_string, true));
            syslog(LOG_NOTICE, "[pack_samplebuffer] mst->endtime: %s", ms_hptime2isotimestr(cur_mst->endtime, time_string, true));
            msr->starttime = cur_mst->endtime + (1/(float)sps * (double)HPTMODULUS); 
            syslog(LOG_NOTICE, "[pack_samplebuffer] new msr->starttime: %s", ms_hptime2isotimestr(msr->starttime, time_string, true));
            syslog(LOG_NOTICE, "[pack_samplebuffer] rem_buffer.size(): %d", (int)rem_buffer.size());
            msr->datasamples = &rem_buffer[0];
            msr->numsamples = rem_buffer.size();
            msr->samplecnt = rem_buffer.size();
            msr->sampletype = 'i';
            // TODO: Check for valid gap at end of the trace.
            mst_addmsr(cur_mst, msr, 1);
            syslog(LOG_NOTICE, "[pack_samplebuffer] new mst->starttime: %s", ms_hptime2isotimestr(cur_mst->starttime, time_string, true));
            syslog(LOG_NOTICE, "[pack_samplebuffer] new mst->endtime: %s", ms_hptime2isotimestr(cur_mst->endtime, time_string, true));
        }

        // Reset the datasamples values in the MSRecord.
        msr->datasamples = NULL;
        msr->numsamples = 0;
        msr->samplecnt = 0;

        // Store the last used mst starttime for later use in log message.
        last_starttime = cur_mst->starttime;
    }
    auto end = std::chrono::high_resolution_clock::now();               
    auto diff = end - start;
    long int pack_ms = (long int)std::chrono::duration_cast<std::chrono::milliseconds>(diff).count();
    syslog(LOG_INFO, "[pack_samplebuffer] Packed samples in %ld milliseconds.", pack_ms);
    if (pack_ms >= 800)
    {
        char last_starttime_string[27];
        ms_hptime2isotimestr(last_starttime, last_starttime_string, true);
        syslog(LOG_WARNING, "[pack_samplebuffer] Packed samples in %ld milliseconds. TOO LONG. Last mst start time: %s.", pack_ms, last_starttime_string);
    }

}


void
Recorder::init_msr (MSRecord* msr, unsigned int channel_number)
{
    // Create the fixed header structure.
    msr->fsdh = (struct fsdh_s *) calloc(1, sizeof(struct fsdh_s));  // Allocate momory for the fsdh. Taken from libmseed pack.c line 530.
    if ( msr->fsdh == NULL )
    {
        syslog(LOG_ERR, "[ERROR][init_msr] Can't allocate memory for the Miniseed record fixed header section.");
        return;
    }

    // Create the 1001 blockette used for tracking the gps_fix values.
    struct blkt_1001_s blkt_1001;
    memset(&blkt_1001, 0, sizeof(struct blkt_1001_s));
    msr_addblockette(msr, (char *) &blkt_1001, sizeof(struct blkt_1001_s), 1001, 0);

    strcpy(msr->network, "XX");
    strcpy(msr->station, this->serial_number.c_str());
    strcpy(msr->location, "00");
    msr->samprate = this->sps;
    msr->byteorder = 0;
    msr->dataquality = 'D';
    msr->sampletype = 'i';
    //msr->reclen = 512;
    //msr->encoding = 11;
    //msr->byteorder = 0;

    // Adjust the channel and in the MSRecord.
    strcpy(msr->channel, (str(boost::format("%03d") % channel_number).c_str()));

}


void
Recorder::set_msr_timeflags(MSRecord* msr, std::vector <timestamp_t>& timestamps)
{
    bool utc_available = true;
    bool gps_fix_ok = true;
    uint8_t gps_fix = 6;
    std::vector <timestamp_t>::iterator ts_it;
    
    for(ts_it = timestamps.begin(); ts_it != timestamps.end(); ts_it++)
    {
        if (!ts_it->gps_fix_ok)
        {
            gps_fix_ok = false;
        }

        // Get the lowest GPS_FIX value. This will be used in the
        // msr blockette 1001.
        if (ts_it->gps_fix < gps_fix)
        {
            gps_fix = ts_it->gps_fix;
        }

        if (!ts_it->utc_available)
        {
            utc_available = false;
        }
    }

    // Set the timelock flag.
    msr->fsdh->io_flags = 0;
    if(gps_fix_ok)
    {
        msr->fsdh->io_flags |= (1 << 5);
    }
    //std::cout << "io_flags: " << (int)msr->fsdh->io_flags << " - " << std::bitset<8>(msr->fsdh->io_flags).to_string() << std::endl;
    
    // Use the "time tag questionable" flag to record the utc_available flag.
    if (utc_available)
    {
        msr->fsdh->dq_flags |= (1 << 7);
    }

    // Set the timing quality.
    msr->Blkt1001->timing_qual = gps_fix;
    msr->Blkt1001->framecnt = 7;
}


bool
Recorder::msr_fits_to_end(MSTrace* mst, MSRecord* msr)
{
    hptime_t postgap;
    hptime_t hp_delta;
    hptime_t gap_limit;
    hp_delta = (hptime_t)((double)HPTMODULUS / (float)sps);
    gap_limit = 1.5 * hp_delta;
    postgap = msr->starttime - mst->endtime;
    syslog(LOG_INFO, "[msr_fits_to_end] postgap: %lld", (long long int)postgap);
    syslog(LOG_INFO, "[msr_fits_to_end] limit: %lld", (long long int)gap_limit);

    if (postgap > 0 && postgap <= gap_limit)
    {
        return true;
    }
    else
    {
        return false;
    }
}


void
Recorder::write_trace_to_file(MSTrace* mst, bool flush, MSRecord* msr_template)
{
    if (mst == NULL)
        return;

    if (mst->numsamples == 0)
    {
        syslog(LOG_WARNING, "[WARNING][write_trace_to_file] No data samples in the trace. Leaving write_trace_to_file.");
        return;
    }

    char time_string[27];
    hptime_t span_start;
    std::string cur_filename;
    BTime file_starttime;
    
    hptime_t file_length_hp = MS_EPOCH2HPTIME((double)file_length);
    span_start = mst->starttime - (mst->starttime % file_length_hp);
    ms_hptime2btime(span_start, &file_starttime);
    cur_filename = get_filename(file_starttime, mst->channel); 
    
    FILE *out_fid = NULL;
    int64_t packed_samples = 0;
    int64_t packed_records = 0;

    out_fid = fopen(cur_filename.c_str(), "ab");
            
    if (out_fid != NULL)
    {
        syslog(LOG_INFO, "[write_trace_to_file] mst->starttime: %s", ms_hptime2isotimestr(mst->starttime, time_string, true));
        syslog(LOG_INFO, "[write_trace_to_file] mst->endtime: %s", ms_hptime2isotimestr(mst->endtime, time_string, true));
        syslog(LOG_INFO, "[write_trace_to_file] Packing msr to file %s.", cur_filename.c_str());
        packed_records = mst_pack(mst, &record_handler, out_fid, 512, 11, 1, &packed_samples, (int)flush, 0, msr_template);
        if (packed_records >= 0)
        {
            syslog(LOG_INFO, "[write_trace_to_file] Packed %ld samples.", (long int)packed_samples);
        }
        else
        {
            syslog(LOG_ERR, "[ERROR][write_trace_to_file] Error when packing the records using mst_pack.");
        }
        syslog(LOG_INFO, "[write_trace_to_file] mst->starttime: %s", ms_hptime2isotimestr(mst->starttime, time_string, true));
        syslog(LOG_INFO, "[write_trace_to_file] mst->endtime: %s", ms_hptime2isotimestr(mst->endtime, time_string, true));
        fclose(out_fid);
    }
    else
    {
        syslog(LOG_ERR, "[ERROR][write_trace_to_file] Error opening the file %s.", cur_filename.c_str());
    }
    // TODO: If there is a permanent error opening the file, the mst
    // will grow and fill up the memory. Add some error handling
    // discarding data after some time to prevent a blocking due to
    // large memory use.
    
}


std::string
Recorder::get_filename(BTime start_time, char* channel)
{
    // Create the output filename. The length of the miniseed files should be
    // defined in seconds. The start of the file should be a multiple of the
    // file length after a full hour. 
    // e.g. 
    // length = 3600s: start of the file every hour.
    // length = 300s:  start of a file every 5 minutes of an hour (13:00,
    // 13:05, ...)
    std::string filename;
    std::string year_str = str(boost::format("%04d") % start_time.year);
    std::string doy_str = str(boost::format("%03d") % start_time.day);
    std::string hour_str = str(boost::format("%02d") % (int)start_time.hour);
    std::string min_str = str(boost::format("%02d") % (int)start_time.min);
    std::string sec_str = str(boost::format("%02d") % (int)start_time.sec);
    
    filename = year_str + "_" + doy_str + "_" + hour_str + min_str + sec_str + "_" + serial_number + "_" + channel + ".msd";

    // If needed create/update the ouput directory structure.
    // YEAR
    // ---DOY
    // ------SERIAL_NUMBER
    std::string dir = data_dir + "/" + year_str;
    if (!boost::filesystem::exists(dir))
    {
        boost::filesystem::create_directory(dir);
    }

    dir = dir + "/" + doy_str;
    if (!boost::filesystem::exists(dir))
    {
        boost::filesystem::create_directory(dir);
    }

    dir = dir + "/" + serial_number;
    if (!boost::filesystem::exists(dir))
    {
        boost::filesystem::create_directory(dir);
    }

    filename = dir + "/" + filename;

    return filename;
}


void
record_handler (char *record, int reclen, void *ptr)
{
  if ( fwrite(record, reclen, 1, (FILE*)ptr) != 1 )
    {
      syslog(LOG_DEBUG, "[record_handler] Cannot write to output file.");
    }
}