Warnings provider to access warnings text products

wxdata/source/scwx/awips/text_product_file.cpp

@@ -67,7 +67,7 @@ bool TextProductFile::LoadFile(const std::string& filename)
 
 bool TextProductFile::LoadData(std::istream& is)
 {
-   logger_->debug("Loading Data");
+   logger_->trace("Loading Data");
 
    while (!is.eof())
    {

wxdata/source/scwx/awips/wmo_header.cpp

@@ -53,7 +53,7 @@ public:
 WmoHeader::WmoHeader() : p(std::make_unique<WmoHeaderImpl>()) {}
 WmoHeader::~WmoHeader() = default;
 
-WmoHeader::WmoHeader(WmoHeader&&) noexcept = default;
+WmoHeader::WmoHeader(WmoHeader&&) noexcept            = default;
 WmoHeader& WmoHeader::operator=(WmoHeader&&) noexcept = default;
 
 bool WmoHeader::operator==(const WmoHeader& o) const
@@ -139,7 +139,7 @@ bool WmoHeader::Parse(std::istream& is)
 
    if (is.eof())
    {
-      logger_->debug("Reached end of file");
+      logger_->trace("Reached end of file");
       headerValid = false;
    }
    else

wxdata/source/scwx/provider/warnings_provider.cpp

@@ -0,0 +1,198 @@
+#include <scwx/provider/warnings_provider.hpp>
+#include <scwx/network/dir_list.hpp>
+#include <scwx/util/logger.hpp>
+
+#include <ranges>
+#include <regex>
+#include <shared_mutex>
+
+#pragma warning(push, 0)
+#define LIBXML_HTML_ENABLED
+#include <cpr/cpr.h>
+#include <libxml/HTMLparser.h>
+#pragma warning(pop)
+
+namespace scwx
+{
+namespace provider
+{
+
+static const std::string logPrefix_ = "scwx::provider::warnings_provider";
+static const auto        logger_    = util::Logger::Create(logPrefix_);
+
+static constexpr std::chrono::seconds kUpdatePeriod_ {15};
+
+class WarningsProvider::Impl
+{
+public:
+   struct FileInfoRecord
+   {
+      std::chrono::system_clock::time_point startTime_ {};
+      std::chrono::system_clock::time_point lastModified_ {};
+      size_t                                size_ {};
+      bool                                  updated_ {};
+   };
+
+   typedef std::map<std::string, FileInfoRecord> WarningFileMap;
+
+   explicit Impl(const std::string& baseUrl) :
+       baseUrl_ {baseUrl}, files_ {}, filesMutex_ {}
+   {
+   }
+
+   ~Impl() {}
+
+   std::string baseUrl_;
+
+   WarningFileMap    files_;
+   std::shared_mutex filesMutex_;
+};
+
+WarningsProvider::WarningsProvider(const std::string& baseUrl) :
+    p(std::make_unique<Impl>(baseUrl))
+{
+}
+WarningsProvider::~WarningsProvider() = default;
+
+WarningsProvider::WarningsProvider(WarningsProvider&&) noexcept = default;
+WarningsProvider&
+WarningsProvider::operator=(WarningsProvider&&) noexcept = default;
+
+std::pair<size_t, size_t>
+WarningsProvider::ListFiles(std::chrono::system_clock::time_point newerThan)
+{
+   using namespace std::chrono;
+
+   static const std::regex reWarningsFilename {
+      "warnings_[0-9]{8}_[0-9]{2}.txt"};
+   static const std::string dateTimeFormat {"warnings_%Y%m%d_%H.txt"};
+
+   logger_->trace("Listing files");
+
+   size_t updatedObjects = 0;
+   size_t totalObjects   = 0;
+
+   // Perform a directory listing
+   auto records = network::DirList(p->baseUrl_);
+
+   // Sort records by filename
+   std::sort(records.begin(),
+             records.end(),
+             [](auto& a, auto& b) { return a.filename_ < b.filename_; });
+
+   // Filter warning records
+   auto warningRecords =
+      records |
+      std::views::filter(
+         [](auto& record)
+         {
+            return record.type_ == std::filesystem::file_type::regular &&
+                   std::regex_match(record.filename_, reWarningsFilename);
+         });
+
+   std::unique_lock lock(p->filesMutex_);
+
+   Impl::WarningFileMap warningFileMap;
+
+   // Store records
+   for (auto& record : warningRecords)
+   {
+      // Determine start time
+      sys_time<hours>    startTime;
+      std::istringstream ssFilename {record.filename_};
+
+      ssFilename >> parse(dateTimeFormat, startTime);
+
+      // If start time is valid
+      if (!ssFilename.fail())
+      {
+         // Determine if the record should be marked updated
+         bool updated = true;
+         auto it      = p->files_.find(record.filename_);
+         if (it != p->files_.cend())
+         {
+            auto& existingRecord = it->second;
+
+            updated = existingRecord.updated_ ||
+                      record.size_ != existingRecord.size_ ||
+                      record.mtime_ != existingRecord.lastModified_;
+         }
+
+         // Update object counts, but only if newer than threshold
+         if (newerThan < startTime)
+         {
+            if (updated)
+            {
+               ++updatedObjects;
+            }
+            ++totalObjects;
+         }
+
+         // Store record
+         warningFileMap.emplace(
+            std::piecewise_construct,
+            std::forward_as_tuple(record.filename_),
+            std::forward_as_tuple(
+               startTime, record.mtime_, record.size_, updated));
+      }
+   }
+
+   p->files_ = std::move(warningFileMap);
+
+   return std::make_pair(updatedObjects, totalObjects);
+}
+
+std::vector<std::shared_ptr<awips::TextProductFile>>
+WarningsProvider::LoadUpdatedFiles(
+   std::chrono::system_clock::time_point newerThan)
+{
+   logger_->debug("Loading updated files");
+
+   std::vector<std::shared_ptr<awips::TextProductFile>> updatedFiles;
+
+   std::vector<std::pair<std::string, cpr::AsyncResponse>> asyncResponses;
+
+   std::unique_lock lock(p->filesMutex_);
+
+   // For each warning file
+   for (auto& record : p->files_)
+   {
+      // If file is updated, and time is later than the threshold
+      if (record.second.updated_ && newerThan < record.second.startTime_)
+      {
+         // Retrieve warning file
+         asyncResponses.emplace_back(
+            record.first,
+            cpr::GetAsync(cpr::Url {p->baseUrl_ + "/" + record.first}));
+
+         // Clear updated flag
+         record.second.updated_ = false;
+      }
+   }
+
+   lock.unlock();
+
+   // Wait for warning files to load
+   for (auto& asyncResponse : asyncResponses)
+   {
+      cpr::Response response = asyncResponse.second.get();
+      if (response.status_code == cpr::status::HTTP_OK)
+      {
+         logger_->debug("Loading file: {}", asyncResponse.first);
+
+         // Load file
+         std::shared_ptr<awips::TextProductFile> textProductFile {
+            std::make_shared<awips::TextProductFile>()};
+         std::istringstream responseBody {response.text};
+         if (textProductFile->LoadData(responseBody))
+         {
+            updatedFiles.push_back(textProductFile);
+         }
+      }
+   }
+
+   return updatedFiles;
+}
+
+} // namespace provider
+} // namespace scwx