ASDP_Streamfile_Analysis_CXX.cpp

/*
 * Copyright (C) 2025: Arizona Board of Regents on Behalf of the University of Arizona
 */

// This is a client that analyzes a camera streamfile, reading and handling all messages in it.

/**
 * @file ASDP_Streamfile_Analysis_CXX.cpp
 * @brief Apache Strap-Down Pilotage example analysis module using C++.
 *
* @author ReliaSolve.
* @date November 3rd, 2025.
*/

#include <iostream>
#include <chrono>
#include <map>
#include <ASDP_Core_API.h>
#include <ASDP_StreamPacketSortedQueue.h>

using namespace asdp;

void usage(std::string programName)
{
  std::cerr << "Usage: " << programName << " [--analysisServer URL] <streamfile>" << std::endl;
  std::cerr << "  --analysisServer URL  URL of the analysis module server. Default is file:://analysis.json." << std::endl;
  std::cerr << "  <streamfile> Name of the camera stream file to read data from." << std::endl;
}

int main(int argc, char** argv)
{
  std::string analysisURL = "file://analysis.json";  ///< URL of the analysis module server.
  std::string streamFileName;       ///< IP Name of the stream file to read from.
  size_t realParams = 0;

  // Parse the command line arguments, with the first non-flag argument being the
  // name of the IP address to listen on.  There is a --serial flag to specify
  // the serial number of the server, which defaults to 1.
  for (int i = 1; i < argc; ++i) {
    if (std::string(argv[i]) == "--analysisServer") {
      if ((i + 1) >= argc) {
        usage(argv[0]);
        return 1;
      }
      analysisURL = argv[++i];
    }
    else if (argv[i][0] == '-' ) {
      std::cerr << "Unknown flag: " << argv[i] << std::endl;
      usage(argv[0]);
      return 1;
    } else switch (realParams++) {
      case 0:
        streamFileName = argv[i];
        break;
      default:
        usage(argv[0]);
        return 2;
    }
  }
  if (realParams != 1) {
    usage(argv[0]);
    return 2;
  }

  // Open a JSON string server to report analysis results to.
  std::shared_ptr<JSONStringSender> jsonSender;
  Status s = JSONStringSender::Create(analysisURL, jsonSender);
  if (s != OKAY) {
    std::cerr << "Failed to create JSONStringSender: " << ErrorMessage(s) << std::endl;
    return 100;
  }

  // Try streaming data from first camera at its highest rate.
  uint32_t camID = 1;
  {
    // Open a file receiver to read the file.
    ReceiverFile receiver(streamFileName);
    if (receiver.GetConstructorStatus() != OKAY) {
      std::cerr << "Error opening file " << streamFileName << ": " << ErrorMessage(receiver.GetConstructorStatus()) << std::endl;
      return 3;
    }

    // Use a sorting queue to ensure that we process the messages in order even if the UDP packets
    // arrive out of order.
    StreamPacketSortedQueue sortedQueue(50);

    // Do analysis on frames until we run out of data.
    // We initialize the data every FRAME_BEGIN, accumulate over all FRAME_DATA, and complete at FRAME_END
    size_t numFrames = 0;
    double sum = 0;
    size_t pixels = 0;

    while (true) {
      // Get the next packet.
      std::shared_ptr<asdp::StreamPacket> packet;
      size_t offset = 0;
      Status status = receiver.ReceiveStreamPacket(10.0, packet, offset);
      if (status == asdp::TIMEOUT) {
        break;  // No more packets available.
      }
      if (status != asdp::OKAY) {
        std::cerr << "Error receiving StreamPacket: " << ErrorMessage(status) << std::endl;
        return 33;
      }
      // Add to the sorted queue and then handle any messages that are ready to be processed.
      std::list< std::shared_ptr<StreamPacket> > readyPackets = sortedQueue.AddPacket(packet);
      if (readyPackets.size() > 1) {
        std::cerr << "Warning: More than one packet ready to process (re-ordered or missing packet)." << std::endl;
      }
      while (!readyPackets.empty()) {

        std::shared_ptr<asdp::StreamPacket> receiveStreamPacket = readyPackets.front();
        readyPackets.pop_front();

        // Get and handle all messages from the packet.
        std::shared_ptr<asdp::Message> message;
        status = receiveStreamPacket->GetNextMessage(message);
        if (status != asdp::OKAY) {
          std::cerr << "Error getting message from packet: " << ErrorMessage(status) << std::endl;
          return 35;
        }
        while (message != nullptr) {
          asdp::MessageID rID;
          status = message->GetType(rID);
          if (status != asdp::OKAY) {
            std::cerr << "Error getting type from message: " << ErrorMessage(status) << std::endl;
            return 36;
          }
          if (rID == CONSOLIDATED_FRAME_DATA) {
            MessageConsolidatedFrameData frameData(*message);
            if (frameData.GetConstructorStatus() != asdp::OKAY) {
              std::cerr << "Error constructing FrameData message: " << ErrorMessage(frameData.GetConstructorStatus()) << std::endl;
              return 37;
            }

            bool isBeginFrame;
            status = frameData.GetBeginFrameFlag(isBeginFrame);
            if (status != asdp::OKAY) {
              std::cerr << "Error getting begin frame flag from FrameData message: " << ErrorMessage(status) << std::endl;
              return 39;
            }
            if (isBeginFrame) {
              sum = 0;
              pixels = 0;
            }

            // Find out how many pixels are in the frame and sum their values.
            if (frameData.GetConstructorStatus() != asdp::OKAY) {
              std::cerr << "Error constructing FrameData message: " << ErrorMessage(frameData.GetConstructorStatus()) << std::endl;
              return 37;
            }
            uint16_t left, right, top, bottom;
            status = frameData.GetLeft(left);
            if (status != asdp::OKAY) {
              std::cerr << "Error getting left from FrameData message: " << ErrorMessage(status) << std::endl;
              return 38;
            }
            status = frameData.GetRight(right);
            if (status != asdp::OKAY) {
              std::cerr << "Error getting right from FrameData message: " << ErrorMessage(status) << std::endl;
              return 39;
            }
            status = frameData.GetTop(top);
            if (status != asdp::OKAY) {
              std::cerr << "Error getting top from FrameData message: " << ErrorMessage(status) << std::endl;
              return 40;
            }
            status = frameData.GetBottom(bottom);
            if (status != asdp::OKAY) {
              std::cerr << "Error getting bottom from FrameData message: " << ErrorMessage(status) << std::endl;
              return 41;
            }
            uint8_t* rawData;
            status = frameData.GetDataPointer(rawData);
            if (status != asdp::OKAY) {
              std::cerr << "Error getting data pointer from FrameData message: " << ErrorMessage(status) << std::endl;
              return 42;
            }
            pixels += (right - left + 1) * (bottom - top + 1);
            uint16_t* data = reinterpret_cast<uint16_t*>(rawData);
            uint16_t stride = (right - left + 1);
            for (uint16_t y = 0; y <= bottom - top; ++y) {
              for (uint16_t x = 0; x <= right - left; ++x) {
                sum += data[y * stride + x];
              }
            }

            bool isEndFrame;
            status = frameData.GetEndFrameFlag(isEndFrame);
            if (status != asdp::OKAY) {
              std::cerr << "Error getting end frame flag from FrameData message: " << ErrorMessage(status) << std::endl;
              return 42;
            }
            if (isEndFrame) {
              std::cout << "  Frame " << ++numFrames << " has average value " << sum / pixels << std::endl;
              // Send the analysis result as a JSON object.
              Time time;
              status = frameData.GetTime(time);
              if (status != asdp::OKAY) {
                std::cerr << "Error getting time from FrameData message: " << ErrorMessage(status) << std::endl;
                return 43;
              }
              std::string jsonString = std::string("{ \"CamID\": ") + std::to_string(camID)
                    + ", \"Time\": "
                    + "[" + std::to_string(time.seconds) + "," + std::to_string(time.microseconds) + "], "
                    + "\"Name\": \"Average\", "
                    + " \"Value\": " + std::to_string(sum/pixels) + "}";
              status = jsonSender->Send(jsonString);
              if (status != asdp::OKAY) {
                std::cerr << "Error sending JSON string: " << ErrorMessage(status) << std::endl;
                return 44;
              }
            }
          }

          status = receiveStreamPacket->GetNextMessage(message);
          if (status != asdp::OKAY) {
            std::cerr << "Error getting first message from packet: " << ErrorMessage(status) << std::endl;
            return 45;
          }
        }
      }
    }
    std::cout << "Analyzed " << numFrames << " images." << std::endl;
  }

  return 0;
}