supercell-wx/wxdata/source/scwx/provider/aws_level3_data_provider.cpp
2023-04-20 00:27:32 -05:00

216 lines
5.9 KiB
C++

#include <scwx/provider/aws_level3_data_provider.hpp>
#include <scwx/common/sites.hpp>
#include <scwx/util/logger.hpp>
#include <scwx/util/time.hpp>
#include <shared_mutex>
#include <aws/s3/model/ListObjectsV2Request.h>
#include <fmt/chrono.h>
#include <fmt/format.h>
#if !defined(_MSC_VER)
# include <date/date.h>
#endif
namespace scwx
{
namespace provider
{
static const std::string logPrefix_ =
"scwx::provider::aws_level3_data_provider";
static const auto logger_ = util::Logger::Create(logPrefix_);
static const std::string kDefaultBucketName_ = "unidata-nexrad-level3";
static const std::string kDefaultRegion_ = "us-east-1";
static std::unordered_map<std::string, std::vector<std::string>> productMap_;
static std::shared_mutex productMutex_;
class AwsLevel3DataProvider::Impl
{
public:
explicit Impl(AwsLevel3DataProvider* self,
const std::string& radarSite,
const std::string& product,
const std::string& bucketName) :
self_ {self},
radarSite_ {radarSite},
siteId_ {common::GetSiteId(radarSite_)},
product_ {product},
bucketName_ {bucketName}
{
}
~Impl() = default;
void ListProducts();
AwsLevel3DataProvider* self_;
std::string radarSite_;
std::string siteId_;
std::string product_;
std::string bucketName_;
};
AwsLevel3DataProvider::AwsLevel3DataProvider(const std::string& radarSite,
const std::string& product) :
AwsLevel3DataProvider(
radarSite, product, kDefaultBucketName_, kDefaultRegion_)
{
}
AwsLevel3DataProvider::AwsLevel3DataProvider(const std::string& radarSite,
const std::string& product,
const std::string& bucketName,
const std::string& region) :
AwsNexradDataProvider(radarSite, bucketName, region),
p(std::make_unique<Impl>(this, radarSite, product, bucketName))
{
}
AwsLevel3DataProvider::~AwsLevel3DataProvider() = default;
AwsLevel3DataProvider::AwsLevel3DataProvider(AwsLevel3DataProvider&&) noexcept =
default;
AwsLevel3DataProvider&
AwsLevel3DataProvider::operator=(AwsLevel3DataProvider&&) noexcept = default;
std::string
AwsLevel3DataProvider::GetPrefix(std::chrono::system_clock::time_point date)
{
return fmt::format(
"{0}_{1}_{2:%Y_%m_%d}_", p->siteId_, p->product_, fmt::gmtime(date));
}
std::chrono::system_clock::time_point
AwsLevel3DataProvider::GetTimePointByKey(const std::string& key) const
{
return GetTimePointFromKey(key);
}
std::chrono::system_clock::time_point
AwsLevel3DataProvider::GetTimePointFromKey(const std::string& key)
{
std::chrono::system_clock::time_point time {};
constexpr size_t offset = 8;
// Filename format is GGG_PPP_YYYY_MM_DD_HH_MM_SS
static const size_t formatSize = std::string("YYYY_MM_DD_HH_MM_SS").size();
if (key.size() >= offset + formatSize)
{
using namespace std::chrono;
#if !defined(_MSC_VER)
using namespace date;
#endif
static const std::string timeFormat {"%Y_%m_%d_%H_%M_%S"};
std::string timeStr {key.substr(offset, formatSize)};
std::istringstream in {timeStr};
in >> parse(timeFormat, time);
if (in.fail())
{
logger_->warn("Invalid time: \"{}\"", timeStr);
}
}
else
{
logger_->warn("Time not parsable from key: \"{}\"", key);
}
return time;
}
void AwsLevel3DataProvider::RequestAvailableProducts()
{
p->ListProducts();
}
std::vector<std::string> AwsLevel3DataProvider::GetAvailableProducts()
{
std::shared_lock readLock(productMutex_);
auto siteProductMap = productMap_.find(p->radarSite_);
if (siteProductMap != productMap_.cend())
{
return siteProductMap->second;
}
return {};
}
void AwsLevel3DataProvider::Impl::ListProducts()
{
std::shared_lock readLock(productMutex_);
// Only list once per radar site
if (productMap_.contains(radarSite_))
{
return;
}
readLock.unlock();
logger_->debug("ListProducts()");
static const std::string delimiter {"_"};
// Prefix format: GGG_
const std::string prefix = fmt::format("{0}_", siteId_);
Aws::S3::Model::ListObjectsV2Request request;
request.SetBucket(bucketName_);
request.SetPrefix(prefix);
request.SetDelimiter(delimiter);
auto outcome = self_->client()->ListObjectsV2(request);
if (outcome.IsSuccess())
{
std::unique_lock writeLock(productMutex_);
// If the product was populated since first checked, don't re-populate
if (productMap_.contains(radarSite_))
{
return;
}
auto& prefixes = outcome.GetResult().GetCommonPrefixes();
// Create a vector with reserved capacity
std::vector<std::string> productList;
productList.reserve(prefixes.size());
std::for_each(prefixes.cbegin(),
prefixes.cend(),
[&](const Aws::S3::Model::CommonPrefix& commonPrefix)
{
// Prefix format: GGG_PPP_
std::string prefix = commonPrefix.GetPrefix();
size_t left = prefix.find('_');
size_t right = prefix.rfind('_');
// If left != npos, right != npos
if (left != std::string::npos && right > left)
{
// The product starts after the left delimeter, and
// ends before the right delimeter.
++left;
productList.push_back(
prefix.substr(left, right - left));
}
});
// Remove extra capacity if necessary
productList.shrink_to_fit();
productMap_.emplace(radarSite_, std::move(productList));
}
}
} // namespace provider
} // namespace scwx