protocol_oob.cpp

Go to the documentation of this file.

// Copyright (c) 2024 anabrid GmbH
// Contact: https://www.anabrid.com/licensing/
// SPDX-License-Identifier: MIT OR GPL-2.0-or-later
 
#include <bitset>
#include <cstring>
#include <sstream>
#include <daq/daq.h>
#include <proto/main.pb.h>
#include <protocol/protocol_oob.h>
#include <run/run.h>
#include <utils/logging.h>
#include <utils/streaming_json.h>
#include <Arduino.h>
 
 
 
 
void client::RunProtobufDataHandler::handle(const run::RunStateChange change,
                                                                const run::Run &run, const char *reason) {
  pb_Envelope envelope;
  auto& msg = transport::init_v1_message(envelope, pb_MessageV1_run_state_change_message_tag);
  auto& run_state_change = msg.kind.run_state_change_message = pb_RunStateChangeMessage pb_RunStateChangeMessage_init_default;
  run_state_change.has_run = true;
  std::strncpy(run_state_change.run.id, run.id.c_str(), sizeof(run_state_change.run.id));
  run_state_change.run.chunk = run.next_chunk();
 
  run_state_change.has_time = true;
  auto& time = run_state_change.time;
  time.prefix = pb_Prefix_NANO;
  time.value = change.t;
 
  run_state_change.old = static_cast<pb_RunState>(change.old);
  run_state_change.new_ = static_cast<pb_RunState>(change.new_);
 
  if (reason) {
    std::strncpy(run_state_change.reason, reason, sizeof(run_state_change.reason));
  }
 
  stream->write(envelope);
}
 
struct SizedPtr {
  size_t size;
  const void* data;
};
 
FASTRUN bool encode_byte_array(pb_ostream_t *stream, const pb_field_t *field, void * const *arg) {
  if (!pb_encode_tag_for_field(stream, field))
    return false;
 
  auto ptr = static_cast<SizedPtr*>(*arg);
  return pb_encode_string(stream, static_cast<const uint8_t*>(ptr->data), ptr->size);
}
 
FASTRUN void client::RunProtobufDataHandler::prepare(const run::Run &run, size_t channel_stride) {
  auto& msg = transport::init_v1_message(envelope, pb_MessageV1_run_data_message_tag);
  auto& run_data = msg.kind.run_data_message;
  run_data = pb_RunDataMessage pb_RunDataMessage_init_default;
  run_data.has_entity = true;
  run_data.has_run = true;
  std::strncpy(run_data.run.id, run.id.c_str(), sizeof(run_data.run.id));
 
  auto& entity = run_data.entity;
  auto& entity_id = carrier::Carrier::get().get_entity_id();
 
  std::stringstream ss;
  ss << entity_id;
  ss.read(entity.path, sizeof(entity.path));
 
  run_data.has_data = true;
  auto& data = run_data.data;
  data.has_type = true;
  data.type = {
    .which_kind = pb_DataType_integer_tag,
    .kind = {
      .integer = {
        .signess = pb_IntegerType_Signedness_Unsigned,
        .bitwidth = 16
      }
    }
  };
 
  data.channel_stride = channel_stride;
  data.data.funcs.encode = encode_byte_array;
 
  auto& carrier = carrier::Carrier::get();
  auto& adc_channels = carrier.get_adc_channels();
 
  auto& channel_idx = data.channels_count = 0;
  for (size_t idx = 0; idx < adc_channels.size(); ++idx) {
    auto& channel = adc_channels[idx];
    if (channel.idx == carrier::ADCChannel::DISABLED) continue;
 
    constexpr double adc_discrete_gain = -0.00015265824;
    constexpr double adc_discrete_offset = 1.25;
 
    auto &pb_channel = data.channels[channel_idx];
    pb_channel.idx = idx;
    pb_channel.gain = adc_discrete_gain * channel.gain;
    pb_channel.offset = adc_discrete_offset * channel.gain + channel.offset;
    pb_channel.probe = channel.probe;
    ++channel_idx;
  }
 
  chunk = 0;
}
 
FASTRUN run::RunHandleResult client::RunProtobufDataHandler::handle(const run::Run &run, uint16_t *data,
                                                                    size_t sample_count) {
  if (sample_count == 0)
    return run::RunHandleResult::ok({});
 
  auto& message = envelope.kind.message_v1;
  auto& run_data = message.kind.run_data_message;
 
  auto& pb_data = run_data.data;
  if (pb_data.channels_count == 0)
    return run::RunHandleResult::ok({});
 
  pb_data.sample_count = sample_count;
  run_data.run.chunk = run.next_chunk();
  auto& values = pb_data.data;
  values.funcs.encode = encode_byte_array;
  SizedPtr sized_ptr{
    .size = sample_count * pb_data.channel_stride * sizeof(uint16_t),
    .data = reinterpret_cast<uint8_t*>(data)
  };
  values.arg = &sized_ptr;
  return stream->write(envelope);
}
 
run::RunHandleResult client::RunProtobufDataHandler::handle_op_end(const run::Run &run,
                                                       std::array<std::array<float, 8>, 6> data) {
  auto& msg = transport::init_v1_message(envelope, pb_MessageV1_run_data_end_message_tag);
  auto& run_data = msg.kind.run_data_end_message;
  run_data = pb_RunDataEndMessage pb_RunDataEndMessage_init_default;
  run_data.has_run = true;
  std::strncpy(run_data.run.id, run.id.c_str(), sizeof(run_data.run.id));
  run_data.run.chunk = run.next_chunk();
 
  run_data.has_entity = true;
  auto& entity = run_data.entity;
  auto& entity_id = carrier::Carrier::get().get_entity_id();
  entity_id.copy(entity.path, sizeof(entity.path));
 
  run_data.has_data = true;
  auto& data_pb = run_data.data;
  data_pb.sample_count = 1;
  data_pb.channel_stride = 48;
  data_pb.has_type = true;
  data_pb.type = {
    .which_kind = pb_DataType_float__tag,
    .kind = {
      .float_ = {
        .bitwidth = 32
      }
    }
  };
 
  SizedPtr sized_ptr{
    .size = 48 * sizeof(float),
    .data = reinterpret_cast<void*>(data.data())
  };
 
  //cannot use adc_scaling, because config configurates only 8 channels
  auto& idx = data_pb.channels_count = 0;
  for (; idx < 48; ++idx) {
    auto& scaling = data_pb.channels[idx];
    scaling.idx = idx;
    scaling.gain = 1.0;
    scaling.offset = 0.0;
  }
 
  data_pb.data.funcs.encode = encode_byte_array;
  data_pb.data.arg = &sized_ptr;
  return stream->write(envelope);
}