JsonStreamer.h

/* ----------------------
* Copyright 2023 Université Libre de Bruxelles(ULB), Universidad Politécnica de Madrid(UPM), CREAL, Deutsches Zentrum für Luft - und Raumfahrt(DLR)
 
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at < http://www.apache.org/licenses/LICENSE-2.0%3E
 
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissionsand
* limitations under the License.
---------------------- */
 
 
#pragma once
 
#include <filesystem>
#include <span>
#include <fstream>
#include <vector>
#include <optional>
#include <thread>
#include <mutex>
#include <atomic>
 
#include <glm/vec2.hpp>
 
#define VULKAN_HPP_DISPATCH_LOADER_DYNAMIC 1
#include <vulkan/vulkan.hpp>
 
#define HVT_FUNCTION_PROTOTYPES
 
#ifdef WIN32
#define HVT_ENTRY_ATTR __declspec(dllexport)
#endif
#include "../RVSVulkan/SourcesVulkan/HvtStreamingAPI.h"
 
#include "../RVSVulkan/RVSConfig/include/Config.h"
 
 
enum class HvtHandleType {
  CONTEXT,
  SEMAPHORE
};
 
template<typename OurType, typename HvtHandle, HvtHandleType htype>
struct Handle{
    Handle() : type(htype) {}
    static constexpr HvtHandleType staticType = htype;
 
    static OurType* check(HvtHandle hvtHandle) {
      Handle* h = reinterpret_cast<Handle*>(hvtHandle);
      if(h && staticType == h->type) {
        return static_cast<OurType*>(h);
      } else {
        throw HvtResult::HVT_ERROR_INVALID_HANDLE;
      }
    }
 
    static OurType* opt_check(HvtHandle hvtHandle) {
      if(!hvtHandle) {
        return nullptr;
      }
      return check(hvtHandle);
    }
 
    HvtHandle to_handle(){
      return reinterpret_cast<HvtHandle>(this);
    }
  private:
    HvtHandleType type;
};
 
struct Semaphore : public Handle<Semaphore, HvtSemaphore, HvtHandleType::SEMAPHORE>
{
    vk::UniqueSemaphore sem;
};
 
// Implementation class
class JsonStreamer : public Handle<JsonStreamer, HvtStreamingContext, HvtHandleType::CONTEXT>
{
    using Clock = std::chrono::steady_clock;
    static constexpr auto numSlots = 3;
  public:
    JsonStreamer(std::span<const uint8_t, VK_UUID_SIZE> uuid);
    void enumerateStreamsParameters(uint32_t* streamsCount, HvtRGBDStreamParameters* parameters) const;
    void importStreamImages(const HvtStreamImagesExportInfo& exportInfos);
    Semaphore* importSemaphore(const HvtSemaphoreExportInfo& exportInfos);
    void destroySemaphore(Semaphore* sem) const;
    void startStreaming();
    void acquireStreamsFrames(const HvtAcquireStreamFramesInfo& infos);
    void releaseStreamsFrames(Semaphore* waitSem);
    void stopStreaming();
    ~JsonStreamer();
  private:
    void initVk(std::span<const uint8_t, VK_UUID_SIZE> uuid);
    uint32_t findMemoryType(uint32_t typeFilter, vk::MemoryPropertyFlags properties);
    void initSettings();
    void initVkResources();
 
    template<typename Closure>
    void oneTimeSubmit(Closure&& func, vk::SubmitInfo submitInfo = vk::SubmitInfo({},{},{},{}));
 
    struct ImageSlot{
        vk::UniqueImage image;
        vk::UniqueDeviceMemory memory;
    };
 
    struct ReadStream{
        std::ifstream colorFile;
        std::ifstream depthFile;
        size_t colorFrameStride;
        size_t colorFrameSize;
        size_t depthFrameStride;
        size_t depthFrameSize;
 
        glm::ivec2 resolution;
        HvtProjectionType projectionType;
        HvtIntrinsics intrinsics;
        HvtExtrinsics extrinsics;
 
        float anear; float afar;
 
        vk::Format colorFormat;
        vk::Format depthFormat;
        vk::UniqueDeviceMemory stagingMemory;
        vk::UniqueBuffer stagingBuffer;
        char* stagingMapping = nullptr;
 
        std::array<ImageSlot, numSlots> colorSlots;
        std::array<ImageSlot, numSlots> depthSlots;
        bool importedColor = false;
        bool importedDepth = false;
 
        Clock::duration framePeriod;
        int frameCount = 1;
        int streamedFrame = -1;
 
        int frameIndex(Clock::duration time) const;
        bool nextFrameReady(Clock::duration time) const;
    };
 
    void streamingLoop();
    void uploadFrame(vk::CommandBuffer cmd, int streamId, ReadStream &stream, int frame);
    void swapStreamingToPending();
    void swapPendingToReading();
 
    std::optional<rvs::Config> config;
 
    std::vector<ReadStream> readStreams;
 
    vk::UniqueInstance instance;
    vk::PhysicalDevice phyDevice = VK_NULL_HANDLE;
    vk::UniqueDevice device;
    uint32_t queueFamilyIndex;
    vk::Queue queue;
    vk::Queue syncQueue;
    std::mutex queueMutex;
    //vk::UniqueCommandPool commandPool;
 
    std::array<Semaphore*, numSlots> usageWaitSemaphores{};
    uint32_t slotReadingIndex = 0; 
    uint32_t slotPendingIndex = 1; 
    uint32_t slotStreamingIndex = 2; 
    std::atomic_bool newDataInPending = false; 
    std::mutex indicesMutex; 
    std::atomic_bool running = false; 
    std::thread streamingThread; 
};