HistoryReader.cpp

#include "HistoryReader.h"

#include <SD_MMC.h>
#include <algorithm>
#include <string.h>
#include <vector>

#include "Config.h"
#include "HistoryCache.h"
#include "HistoryLogger.h"

namespace {

// microSDの順番待ちの上限。これを超えたら、今回の描画では諦めます。
//
// グラフは利用者が要求して待っている画面なので、穴だらけの結果をすぐ返すより、
// 1秒弱待ってでも揃ったデータを返すほうが妥当です。それでも足りなかったことは
// 呼び出し側に伝わるので（outIncomplete）、少し後に再試行できます。
constexpr uint32_t kSdAccessTimeoutMs = 900;

// 1回の読み出しで取り込む件数。
//
// 1件（18バイト）ずつ読むと、記録1分あたり1回ファイルシステムを経由する
// ことになります。1日分でも1,440回です。512件ずつまとめて読めば3回で済みます。
constexpr size_t kSdChunkRecords = 512;

// 二分探索で求めた位置の、前後に余分に見る件数。
//
// 探索は「記録が古い順に並んでいる」ことを前提にしています。ふだんはその通り
// ですが、時刻合わせで時計がわずかに巻き戻ると、前後が入れ替わった記録が
// できることがあります。そのとき境目のデータを取りこぼさないよう、少し広めに
// 読んで、範囲の判定は1件ずつ行います。
//
// 記録は1分間隔なので、8件はおよそ8分ぶんです。
constexpr size_t kSearchMarginRecords = 8;

// ファイルのindex番目の記録を読みます。
bool readRecordAt(File &f, size_t index, HistorySample &out) {
  if (!f.seek(sizeof(HistoryFileHeader) + index * sizeof(HistorySample))) return false;
  return f.read(reinterpret_cast<uint8_t *>(&out), sizeof(out)) == int(sizeof(out));
}

// startEpoch以降の記録が始まる位置を二分探索で求めます。
//
// これが無いと、目的の期間が月末の数日でも、ファイルの先頭から全部を読むことに
// なります。1ヶ月分は約43,000件あり、数ヶ月ぶん溜まると描画のたびに待たされる
// ようになります。二分探索なら、43,000件でも16回ほど覗くだけで位置が決まります。
size_t findStartIndex(File &f, size_t recordCount, time_t startEpoch) {
  size_t lo = 0;
  size_t hi = recordCount;
  while (lo < hi) {
    size_t mid = lo + (hi - lo) / 2;
    HistorySample s;
    // 読めない場合は探索を諦め、先頭から読みます。遅くはなりますが、
    // 結果は正しいままです。
    if (!readRecordAt(f, mid, s)) return 0;
    if (time_t(s.timestamp) < startEpoch) lo = mid + 1;
    else hi = mid;
  }
  return lo;
}

// 記録1件から、指定された項目の値を取り出します。
// その項目にデータが無ければfalseを返します。
//
// 値は記録されている形のまま（温度なら10倍値）で扱います。集計は足して割るだけ
// なので、実際の単位に直す必要はありません。呼び出し側が同じ形で受け取ります。
bool readField(const HistorySample &s, HistoryReader::Field f, int32_t *out) {
  using F = HistoryReader::Field;
  switch (f) {
    case F::Temperature:
      if (s.tempX10 == INT16_MIN) return false;
      *out = s.tempX10;
      return true;
    case F::Humidity:
      if (s.humidityX10 == 0xFFFF) return false;
      *out = s.humidityX10;
      return true;
    case F::Battery:
      if (s.battery == 0xFF) return false;
      *out = s.battery;
      return true;
    case F::Co2: {
      // 上位2ビットには学習状態が入っているので、取り除いてから使います。
      uint16_t ppm = historyCo2Ppm(s);
      if (ppm == 0xFFFF) return false;
      *out = ppm;
      return true;
    }
    case F::Pressure:
      if (s.pressureX10 == 0xFFFF) return false;
      *out = s.pressureX10;
      return true;
    case F::Voc:
      if (s.vocX100 == 0xFFFF) return false;
      *out = s.vocX100;
      return true;
    // IAQは半分の値で記録されています。2倍に戻すのは表示する側です
    //（他の項目も、10倍値のまま集計して表示側で戻しています）。
    case F::Iaq:
      if (s.iaqHalf == 0xFF) return false;
      *out = s.iaqHalf;
      return true;
    case F::Illuminance:
      if (s.illuminance == 0xFFFF) return false;
      *out = s.illuminance;
      return true;
  }
  return false;
}

// 集計した値を、記録1件分の形に書き戻します。
void writeField(HistorySample &s, HistoryReader::Field f, int32_t v) {
  using F = HistoryReader::Field;
  switch (f) {
    case F::Temperature: s.tempX10 = int16_t(v); break;
    case F::Humidity: s.humidityX10 = uint16_t(v); break;
    case F::Battery: s.battery = uint8_t(v); break;
    case F::Co2: historySetCo2(s, uint16_t(v), 0); break;
    case F::Pressure: s.pressureX10 = uint16_t(v); break;
    case F::Voc: s.vocX100 = uint16_t(v); break;
    case F::Iaq: s.iaqHalf = uint8_t(v); break;
    case F::Illuminance: s.illuminance = uint16_t(v); break;
  }
}

// ファイルの見出しを読み、この形式かどうかを確かめます。
bool readAndCheckHeader(File &f) {
  HistoryFileHeader h;
  if (!f.seek(0)) return false;
  if (f.read(reinterpret_cast<uint8_t *>(&h), sizeof(h)) != int(sizeof(h))) return false;
  return memcmp(h.magic, HISTORY_FILE_MAGIC, sizeof(h.magic)) == 0 &&
         h.version == HISTORY_FILE_VERSION && h.recordSize == sizeof(HistorySample);
}

// MACアドレスからファイル名を作ります。
// HistoryLoggerの同名の関数と必ず同じ規則にしてください。1行で済む変換なので、
// 共通化せずそれぞれが持っています。
String sanitizedFilename(const String &mac) {
  String s = mac;
  s.replace(":", "");
  return String(SD_LOG_DIR) + "/" + s + ".bin";
}

std::vector<HistorySample> readSdRange(const String &mac, time_t startEpoch, time_t endEpoch,
                                      bool *outFailed) {
  std::vector<HistorySample> result;
  // カードが無いことは「データが無い」ことではありません。カードにしか
  // 記録されていない期間について、何も言えないというだけです。
  // 失敗として報告し、呼び出し側が「この期間はデータがありません」と
  // 断定してしまわないようにします。
  if (HistoryLogger::status() == HistoryLogger::SdStatus::NotMounted) {
    if (outFailed) *outFailed = true;
    return result;
  }
  if (!HistoryLogger::acquireSdAccess(kSdAccessTimeoutMs)) {
    if (outFailed) *outFailed = true;
    return result;
  }

  File f = SD_MMC.open(sanitizedFilename(mac), FILE_READ);
  if (!f) {
    // カードは使えるが、この機器のファイルが無い場合です。
    // 「一度も記録されていない」という確かな答えなので、失敗ではありません。
    HistoryLogger::releaseSdAccess();
    return result;
  }
  if (!readAndCheckHeader(f)) {
    // この形式ではないファイルです。Rev 0.1.0の古い履歴が残っている場合が
    // これにあたります。中身を推測して読むことはせず、無かったものとして扱います
    //（記録側が、次に書き込むときに別名へ退避します）。
    f.close();
    HistoryLogger::releaseSdAccess();
    return result;
  }

  size_t fileSize = f.size();
  size_t recordCount = fileSize > sizeof(HistoryFileHeader)
                         ? (fileSize - sizeof(HistoryFileHeader)) / sizeof(HistorySample)
                         : 0;
  size_t startIndex = findStartIndex(f, recordCount, startEpoch);
  // 求めた位置より少し手前から読みます（kSearchMarginRecordsの説明を参照）。
  startIndex = startIndex > kSearchMarginRecords ? startIndex - kSearchMarginRecords : 0;
  if (!f.seek(sizeof(HistoryFileHeader) + startIndex * sizeof(HistorySample))) {
    f.close();
    HistoryLogger::releaseSdAccess();
    if (outFailed) *outFailed = true;
    return result;
  }

  // 読み出し用のバッファ。PSRAM上に置き、staticにして描画のたびに
  // 確保し直さないようにしています。
  static std::vector<HistorySample, PsramAllocator<HistorySample>> s_chunk;
  if (s_chunk.size() != kSdChunkRecords) s_chunk.assign(kSdChunkRecords, HistorySample());

  // 期間の終わりを過ぎた記録を何件見たか。
  // 1件見ただけで打ち切らないのは、読み始めの位置と同じ理由です。
  size_t pastEnd = 0;
  while (pastEnd <= kSearchMarginRecords) {
    int got = f.read(reinterpret_cast<uint8_t *>(s_chunk.data()),
                     kSdChunkRecords * sizeof(HistorySample));
    if (got < int(sizeof(HistorySample))) break;
    size_t n = size_t(got) / sizeof(HistorySample);
    for (size_t i = 0; i < n; i++) {
      const HistorySample &s = s_chunk[i];
      time_t ts = time_t(s.timestamp);
      if (ts >= endEpoch) {
        pastEnd++;
        if (pastEnd > kSearchMarginRecords) break;
        continue;
      }
      if (ts >= startEpoch) result.push_back(s);
    }
    if (n < kSdChunkRecords) break;  // 要求より少なければファイルの終わり
  }
  f.close();
  HistoryLogger::releaseSdAccess();
  return result;
}

}  // namespace

namespace HistoryReader {

size_t getAggregated(const String &mac, Field field, time_t startEpoch, time_t endEpoch,
                     int maxPoints, HistorySample *outBuf, HistorySample *outMinBuf,
                     HistorySample *outMaxBuf, bool *outIncomplete, uint8_t *outAccuracyBuf) {
  if (outIncomplete) *outIncomplete = false;
  // キャッシュ本体と同じ理由でPSRAMに置きます。1機器分のバッファをまるごと
  // 複製するので（24時間で約14KB）、内蔵RAMから取るには大きすぎます。
  static std::vector<HistorySample, PsramAllocator<HistorySample>> s_ramScratch;
  s_ramScratch.assign(size_t(HISTORY_RAM_CAPACITY_HOURS) * 60, HistorySample());
  size_t ramCount = HistoryCache::getRecent(mac, s_ramScratch.data(), s_ramScratch.size());

  // RAMキャッシュが持っている範囲はRAMから、それより古い部分だけをカードから
  // 読みます。
  //
  // 直近の数分はまだカードに書き出されておらず、RAMにしか存在しません
  // （書き出しは最大で10分間隔）。そのため、どちらか一方だけを使うのではなく、
  // 両方を組み合わせる必要があります。
  std::vector<HistorySample> source;
  const time_t ramOldest = ramCount ? time_t(s_ramScratch[0].timestamp) : endEpoch;
  for (size_t i = 0; i < ramCount; i++) {
    time_t ts = time_t(s_ramScratch[i].timestamp);
    if (ts >= startEpoch && ts < endEpoch) source.push_back(s_ramScratch[i]);
  }
  if (ramOldest > startEpoch) {
    time_t sdEnd = ramOldest < endEpoch ? ramOldest : endEpoch;
    bool failed = false;
    std::vector<HistorySample> sd = readSdRange(mac, startEpoch, sdEnd, &failed);
    source.insert(source.end(), sd.begin(), sd.end());
    if (failed && outIncomplete) *outIncomplete = true;
  }

  double bucketWidth = double(endEpoch - startEpoch) / double(maxPoints);

  // 区切りごとの合計と件数。集めるのは指定された1項目だけです。
  //
  // 日表示は1440区切りあるので、この作業領域もPSRAMから取ります。内蔵RAMから
  // 取ると、グラフを開くたびに数十KBを確保して返すことになります。
  static std::vector<double, PsramAllocator<double>> s_sum;
  static std::vector<int, PsramAllocator<int>> s_count;
  s_sum.assign(maxPoints, 0);
  s_count.assign(maxPoints, 0);
  auto &sum = s_sum;
  auto &count = s_count;

  // 最小値・最大値は、呼び出し側が要求したときだけ集計します。
  // 通常の表示では使わないためです。
  const bool wantBounds = (outMinBuf != nullptr || outMaxBuf != nullptr);
  static std::vector<int32_t, PsramAllocator<int32_t>> s_lo;
  static std::vector<int32_t, PsramAllocator<int32_t>> s_hi;
  auto &lo = s_lo;
  auto &hi = s_hi;
  if (wantBounds) {
    lo.assign(maxPoints, INT32_MAX);
    hi.assign(maxPoints, INT32_MIN);
  }

  // 空気質の学習状態も、呼び出し側が要求したときだけ集計します。
  //
  // 区間の中でいちばん低い値を採ります。学習が途中で進んだ区間には、まだ
  // 低精度だったころの記録が混ざっているためです。高いほうを採ると、実際より
  // 確かなデータだったように見えてしまいます。
  //
  // 0xFFは「この区間の記録をまだ1件も見ていない」ことを表す作業用の印です。
  // 0が「学習が済んでいない」という意味を持つので、区別する必要があります。
  static std::vector<uint8_t, PsramAllocator<uint8_t>> s_acc;
  auto &acc = s_acc;
  if (outAccuracyBuf) acc.assign(maxPoints, 0xFF);

  for (const auto &s : source) {
    if (time_t(s.timestamp) < startEpoch || time_t(s.timestamp) >= endEpoch) continue;
    int32_t value;
    if (!readField(s, field, &value)) continue;

    int bucket = int(double(time_t(s.timestamp) - startEpoch) / bucketWidth);
    bucket = std::max(0, std::min(maxPoints - 1, bucket));

    sum[bucket] += value;
    count[bucket]++;
    if (wantBounds) {
      lo[bucket] = std::min(lo[bucket], value);
      hi[bucket] = std::max(hi[bucket], value);
    }
    // ここへ来た記録は、要求された項目の値を持っています。学習状態を要求するのは
    // 空気質を見るときだけなので、この時点で空気質があることも確かです。
    if (outAccuracyBuf) {
      uint8_t a = historyIaqAccuracy(s);
      if (acc[bucket] == 0xFF || a < acc[bucket]) acc[bucket] = a;
    }
  }

  for (int i = 0; i < maxPoints; i++) {
    uint32_t stamp = uint32_t(startEpoch + (i + 0.5) * bucketWidth);
    outBuf[i] = HistorySample();
    outBuf[i].timestamp = stamp;
    if (count[i]) writeField(outBuf[i], field, int32_t(round(sum[i] / count[i])));

    if (outMinBuf) {
      outMinBuf[i] = HistorySample();
      outMinBuf[i].timestamp = stamp;
      if (count[i]) writeField(outMinBuf[i], field, lo[i]);
    }
    if (outMaxBuf) {
      outMaxBuf[i] = HistorySample();
      outMaxBuf[i].timestamp = stamp;
      if (count[i]) writeField(outMaxBuf[i], field, hi[i]);
    }
    // 記録が無かった区間は0にします。呼び出し側は0を「帯を引かない」として扱います。
    if (outAccuracyBuf) outAccuracyBuf[i] = (acc[i] == 0xFF) ? 0 : acc[i];
  }
  return size_t(maxPoints);
}

}  // namespace HistoryReader