`src/ingest/fema_nri.ts`

115 lines
/**
 * FEMA National Risk Index — county table, source 'fema_nri'.
 * URL DRIFT (2026-07-21): the old hazards.fema.gov NRI site 301s to a generic FEMA page.
 * Current home is OpenFEMA: fema.gov/about/reports-and-data/openfema/nri/v120/NRI_Table_Counties.zip
 * (v1.20.0, December 2025 release — period stored as 2025-12-01). See SOURCES.md.
 * Metrics: nri_risk_score, nri_eal_score, nri_sovi_score, nri_resl_score.
 * Display strings (rating + top-3 hazards by <PFX>_RISKS score) go to nri_ratings.
 */
import { execFileSync } from 'node:child_process';
import { readFileSync, mkdirSync } from 'node:fs';
import { join } from 'node:path';
import { openRun, closeRun, download, DOWNLOADS } from './run.ts';
import { loadRegionMaps, upsertMetricRows, type MetricRow } from './crosswalk.ts';
import { query, pool } from '../../db/pool.ts';

const NRI_URL = 'https://www.fema.gov/about/reports-and-data/openfema/nri/v120/NRI_Table_Counties.zip';
const RELEASE_PERIOD = '2025-12-01'; // NRI v1.20.0, December 2025 release

const HAZARDS: Record<string, string> = {
  AVLN: 'Avalanche', CFLD: 'Coastal Flooding', CWAV: 'Cold Wave', DRGT: 'Drought',
  ERQK: 'Earthquake', HAIL: 'Hail', HWAV: 'Heat Wave', HRCN: 'Hurricane',
  ISTM: 'Ice Storm', IFLD: 'Inland Flooding', LNDS: 'Landslide', LTNG: 'Lightning',
  SWND: 'Strong Wind', TRND: 'Tornado', TSUN: 'Tsunami', VLCN: 'Volcanic Activity',
  WFIR: 'Wildfire', WNTW: 'Winter Weather',
};

function parseCsvLine(line: string): string[] {
  const out: string[] = [];
  let cur = '', inQ = false;
  for (let i = 0; i < line.length; i++) {
    const ch = line[i];
    if (inQ) {
      if (ch === '"') { if (line[i + 1] === '"') { cur += '"'; i++; } else inQ = false; }
      else cur += ch;
    } else if (ch === '"') inQ = true;
    else if (ch === ',') { out.push(cur); cur = ''; }
    else cur += ch;
  }
  out.push(cur);
  return out;
}

async function main() {
  const runId = await openRun('fema_nri', NRI_URL);
  try {
    const { path: zipPath, sha256 } = await download(NRI_URL, 'NRI_Table_Counties.zip', { reuseCached: true });
    const extractDir = join(DOWNLOADS, 'nri_counties');
    mkdirSync(extractDir, { recursive: true });
    execFileSync('unzip', ['-o', '-q', zipPath, '-d', extractDir]);

    const text = readFileSync(join(extractDir, 'NRI_Table_Counties.csv'), 'utf8');
    const lines = text.split('\n');
    const header = parseCsvLine(lines[0].replace(/^/, '').replace(/\r$/, ''));
    const col = new Map(header.map((h, i) => [h, i]));
    for (const c of ['STCOFIPS', 'RISK_SCORE', 'RISK_RATNG', 'EAL_SCORE', 'SOVI_SCORE', 'RESL_SCORE']) {
      if (!col.has(c)) throw new Error(`NRI header drift: missing column ${c}`);
    }
    const hazardCols = Object.keys(HAZARDS)
      .map(pfx => ({ pfx, idx: col.get(pfx + '_RISKS') }))
      .filter((h): h is { pfx: string; idx: number } => h.idx != null);

    const maps = await loadRegionMaps();
    const metricRows: MetricRow[] = [];
    const ratingRows: { fips: string; rating: string | null; top: { hazard: string; score: number }[] }[] = [];
    let skipped = 0;

    for (let i = 1; i < lines.length; i++) {
      const line = lines[i].replace(/\r$/, '');
      if (!line) continue;
      const f = parseCsvLine(line);
      const fips = (f[col.get('STCOFIPS')!] || '').padStart(5, '0');
      const region = maps.byFips.get(fips);
      if (!region) { skipped++; continue; }
      const metrics: [string, string][] = [
        ['nri_risk_score', 'RISK_SCORE'], ['nri_eal_score', 'EAL_SCORE'],
        ['nri_sovi_score', 'SOVI_SCORE'], ['nri_resl_score', 'RESL_SCORE'],
      ];
      for (const [metric, colName] of metrics) {
        const v = Number(f[col.get(colName)!]);
        if (!Number.isFinite(v)) continue;
        metricRows.push({ regionId: region.id, source: 'fema_nri', metric, period: RELEASE_PERIOD, value: v });
      }
      const top = hazardCols
        .map(h => ({ hazard: HAZARDS[h.pfx], score: Number(f[h.idx]) }))
        .filter(h => Number.isFinite(h.score))
        .sort((a, b) => b.score - a.score)
        .slice(0, 3)
        .map(h => ({ hazard: h.hazard, score: Math.round(h.score * 100) / 100 }));
      ratingRows.push({ fips, rating: f[col.get('RISK_RATNG')!] || null, top });
    }

    const upserted = await upsertMetricRows(metricRows, runId);
    for (const r of ratingRows) {
      await query(
        `INSERT INTO nri_ratings (county_fips, risk_rating, top_hazards) VALUES ($1, $2, $3)
         ON CONFLICT (county_fips) DO UPDATE SET risk_rating = EXCLUDED.risk_rating, top_hazards = EXCLUDED.top_hazards`,
        [r.fips, r.rating, JSON.stringify(r.top)],
      );
    }
    if (upserted === 0) throw new Error('0 metric rows parsed — NRI format drift');
    await closeRun(runId, 'ok', {
      upserted, skipped, fileHash: sha256,
      notes: `v1.20.0 Dec-2025 release, period ${RELEASE_PERIOD}; ${ratingRows.length} counties, ${upserted} metric rows`,
    });
    console.log(`fema_nri: ${ratingRows.length} counties, ${upserted} metric rows, ${skipped} unmatched`);
  } catch (e: any) {
    await closeRun(runId, 'failed', { notes: String(e.message || e) });
    throw e;
  } finally {
    await pool.end();
  }
}

main().catch((e) => { console.error(e); process.exit(1); });