nate/weather
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dedupe station readings and group rain stats by local day

Browse Source 
- Share a disk-backed per-day dateutc set between live recorder and backfill
- Validate timestamps before saving; dedupe existing files at startup and on read
- Use station-local day (not UTC) for rain totals, streaks, degree days and
  monthly summary so dailyrainin resets line up (Sept rain was 4.37in vs 2.24in)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Nate EmmertandClaude Sonnet 5.5 committed 2026-10-15 12:10:11 -04:00
1 parent 528dec753d
commit 4c2c619fb6
1 file changed
+91 -33
Download Patch File Download Diff File Expand all files Collapse all files
app/server.js
+91 -33
View File
@@ -101,35 +101,112 @@ function stationDateKey(ts) {
return new Date(ts).toISOString().slice(0, 10);
}
let lastRecordedDateutc = 0;
// Station "day" = local calendar day, since the station resets dailyrainin at local midnight.
// Uses the timezone stamped on each reading (falls back to the last one seen / America/New_York).
let stationTzFallback = 'America/New_York';
const dayFmtCache = new Map();
function stationLocalDay(r) {
const tz = r.tz || stationTzFallback;
let f = dayFmtCache.get(tz);
if (!f) {
try { f = new Intl.DateTimeFormat('en-CA', { timeZone: tz }); }
catch { f = new Intl.DateTimeFormat('en-CA', { timeZone: stationTzFallback }); }
dayFmtCache.set(tz, f);
}
return f.format(new Date(r.dateutc));
}
// Per-day set of dateutc values already on disk. Shared by the live recorder and
// the backfill so neither can write a reading the other (or a prior run) saved.
const stationSeen = new Map();
function stationDayFile(day) {
return path.join(STATION_DIR, `${day}.jsonl`);
}
function getStationSeen(day) {
let set = stationSeen.get(day);
if (set) return set;
set = new Set();
try {
for (const l of fs.readFileSync(stationDayFile(day), 'utf8').split('\n')) {
if (!l) continue;
try { set.add(JSON.parse(l).dateutc); } catch {}
}
} catch {}
stationSeen.set(day, set);
return set;
}
// A reading is only saved if it has a sane timestamp and isn't already stored.
// Returns true when a new row was written.
function saveStationReading(obs) {
const ts = obs && obs.dateutc;
if (!Number.isFinite(ts) || ts <= 0 || ts > Date.now() + 5 * 60 * 1000) return false;
const day = stationDateKey(ts);
const seen = getStationSeen(day);
if (seen.has(ts)) return false;
fs.appendFileSync(stationDayFile(day), JSON.stringify(obs) + '\n');
seen.add(ts);
return true;
}
function recordStationReading(obs) {
if (!obs.dateutc || obs.dateutc === lastRecordedDateutc) return;
try {
const day = stationDateKey(obs.dateutc);
const file = path.join(STATION_DIR, `${day}.jsonl`);
fs.appendFileSync(file, JSON.stringify(obs) + '\n');
lastRecordedDateutc = obs.dateutc;
if (obs && obs.tz) stationTzFallback = obs.tz;
saveStationReading(obs);
} catch (e) {
console.error('[station] record error:', e.message);
}
}
// One-time startup pass: drop duplicate dateutc rows from existing day files.
// Rewrites a file (tmp + rename) only when it actually contains duplicates.
function dedupeStationFiles() {
let files;
try { files = fs.readdirSync(STATION_DIR).filter(f => f.endsWith('.jsonl')); } catch { return; }
for (const f of files) {
try {
const file = path.join(STATION_DIR, f);
const seen = new Set();
const keep = [];
let dropped = 0;
for (const l of fs.readFileSync(file, 'utf8').split('\n')) {
if (!l) continue;
let ts;
try { ts = JSON.parse(l).dateutc; } catch { keep.push(l); continue; }
if (ts != null && seen.has(ts)) { dropped++; continue; }
seen.add(ts);
keep.push(l);
}
if (dropped) {
fs.writeFileSync(file + '.tmp', keep.join('\n') + '\n');
fs.renameSync(file + '.tmp', file);
console.log(`[station] ${f}: removed ${dropped} duplicate reading(s)`);
}
} catch (e) {
console.error('[station] dedupe error:', f, e.message);
}
}
}
dedupeStationFiles();
function loadStationReadings(rangeMs) {
const now = Date.now();
const cutoff = now - rangeMs;
const days = Math.ceil(rangeMs / 86400000) + 1;
const results = [];
const seen = new Set();
for (let i = 0; i < days; i++) {
const day = stationDateKey(now - i * 86400000);
const file = path.join(STATION_DIR, `${day}.jsonl`);
const file = stationDayFile(day);
try {
const lines = fs.readFileSync(file, 'utf8').split('\n').filter(Boolean);
for (const line of lines) {
try {
const obj = JSON.parse(line);
const ts = obj.dateutc || 0;
if (ts >= cutoff) results.push(obj);
if (ts >= cutoff && !seen.has(ts)) { seen.add(ts); results.push(obj); }
} catch {}
}
} catch {}
@@ -1114,19 +1191,6 @@ async function runBackfill(daysBack) {
const cutoff = Date.now() - daysBack * 86400000;
backfillState = { running: true, done: 0, total: daysBack, inserted: 0, error: null };
// Per-day cache of existing dateutc values so we don't write duplicates
const existingByDay = new Map();
function getExisting(day) {
if (existingByDay.has(day)) return existingByDay.get(day);
const set = new Set();
try {
const lines = fs.readFileSync(path.join(STATION_DIR, `${day}.jsonl`), 'utf8').split('\n').filter(Boolean);
for (const l of lines) { try { set.add(JSON.parse(l).dateutc); } catch {} }
} catch {}
existingByDay.set(day, set);
return set;
}
let endDate = Date.now();
let inserted = 0;
@@ -1144,13 +1208,7 @@ async function runBackfill(daysBack) {
if (!rec.dateutc) continue;
if (rec.dateutc < oldest) oldest = rec.dateutc;
if (rec.dateutc < cutoff) continue;
const day = stationDateKey(rec.dateutc);
const existing = getExisting(day);
if (!existing.has(rec.dateutc)) {
fs.appendFileSync(path.join(STATION_DIR, `${day}.jsonl`), JSON.stringify(rec) + '\n');
existing.add(rec.dateutc);
inserted++;
}
if (saveStationReading(rec)) inserted++;
}
backfillState.done = Math.min(daysBack, Math.ceil((Date.now() - oldest) / 86400000));
@@ -1252,7 +1310,7 @@ app.get('/api/station/rain-totals', requireDashboardAuth, (_req, res) => {
const byDay = {};
for (const r of readings) {
if (r.dailyrainin == null || !r.dateutc) continue;
const d = stationDateKey(r.dateutc);
const d = stationLocalDay(r);
if (byDay[d] == null || r.dailyrainin > byDay[d]) byDay[d] = r.dailyrainin;
}
return +Object.values(byDay).reduce((a, b) => a + b, 0).toFixed(2);
@@ -1272,7 +1330,7 @@ app.get('/api/station/degree-days', requireDashboardAuth, (req, res) => {
const byDay = {};
for (const r of readings) {
if (r.tempf == null || !r.dateutc) continue;
const d = stationDateKey(r.dateutc);
const d = stationLocalDay(r);
if (!byDay[d]) byDay[d] = [];
byDay[d].push(r.tempf);
}
@@ -1387,7 +1445,8 @@ app.get('/api/station/monthly-summary', requireDashboardAuth, (_req, res) => {
try {
const r = JSON.parse(line);
if (!r.dateutc) continue;
const key = new Date(r.dateutc).toISOString().slice(0, 7); // YYYY-MM
const day = stationLocalDay(r);
const key = day.slice(0, 7); // YYYY-MM (local)
if (!months[key]) months[key] = { temps: [], highTemp: null, highGust: null, dailyRainMax: {} };
const m = months[key];
if (r.tempf != null) {
@@ -1396,7 +1455,6 @@ app.get('/api/station/monthly-summary', requireDashboardAuth, (_req, res) => {
}
if (r.windgustmph != null && (m.highGust == null || r.windgustmph > m.highGust)) m.highGust = r.windgustmph;
if (r.dailyrainin != null) {
const day = stationDateKey(r.dateutc);
if (m.dailyRainMax[day] == null || r.dailyrainin > m.dailyRainMax[day]) m.dailyRainMax[day] = r.dailyrainin;
}
} catch {}
@@ -1473,7 +1531,7 @@ app.get('/api/station/streak', requireDashboardAuth, (_req, res) => {
const byDay = {};
for (const r of readings) {
if (r.dailyrainin == null || !r.dateutc) continue;
const d = stationDateKey(r.dateutc);
const d = stationLocalDay(r);
if (byDay[d] == null || r.dailyrainin > byDay[d]) byDay[d] = r.dailyrainin;
}
const days = Object.keys(byDay).sort().reverse();