EVCS/tools/importer-cellules.mjs

#!/usr/bin/env node
// Verse dans PostgreSQL les tensions cellule par cellule relevees sur la
// console des BMS Pylontech, pour qu'elles soient tracables dans Grafana a
// cote du reste du systeme.
//
// Deux sources, deux formats :
//   - les exports d'historique  (`--historique`, ou `datalist history dtls`)
//     un enregistrement toutes les 30 minutes, jusqu'a 1798 par module ;
//   - les CSV de suivi          (`--suivi`), au pas choisi pendant un essai.
//
// A executer LA OU le mot de passe existe, c'est-a-dire dans le conteneur
// d'historisation, qui monte le depot en lecture seule :
//
//   sudo docker compose exec historisation \
//        node tools/importer-cellules.mjs 'docs/mesures/batteries/*'
//
//   --a-blanc   analyse et compte, n'ecrit rien.

import fs from 'node:fs';
import path from 'node:path';
import pg from 'pg';

const args = process.argv.slice(2);
const BLANC = args.includes('--a-blanc');
const motifs = args.filter((a) => !a.startsWith('--'));

// Correspondance adresse -> code-barres pour les CSV de suivi, qui ne portent
// que l'adresse. Chaque chaine renumerote a partir de 1 : une adresse seule
// est ambigue dans ce parc. Completer au fil des relevés `info <n>`.
const CARTE_SUIVI = {
  1: 'K220803C30160322',   // pack ancien, adresse 1
  3: 'K220803C30160319',   // pack ancien, adresse 3
  // 2 : code-barres pas encore releve -> enregistre en 'pack_ancien:2'
};

const lireTexte = (f) => {
  const brut = fs.readFileSync(f);
  const utf16 = brut[0] === 0xff && brut[1] === 0xfe;
  return brut.toString(utf16 ? 'utf16le' : 'utf8').replace(/\r/g, '');
};

// ---------------------------------------------------------------- historique
function lireHistorique(f) {
  const txt = lireTexte(f);
  const bc = txt.match(/Barcode\s*:\s*(\S+)/);
  if (!bc) return null;
  const recs = new Map();
  for (const ligne of txt.split('\n')) {
    const c = ligne.split(/\s+/).filter(Boolean);
    if (c.length < 18 + 15 * 6) continue;
    if (!/^\d+$/.test(c[0]) || !/^\d\d-\d\d-\d\d$/.test(c[1])) continue;
    const [a, m, j] = c[1].split('-').map(Number);
    const [H, M, S] = c[2].split(':').map(Number);
    recs.set(+c[0], {
      // heure MODULE, corrigee plus bas
      brut: Date.UTC(2000 + a, m - 1, j, H, M, S),
      courant: +c[4], soc: parseInt(c[16], 10), etat: c[10],
      mv: Array.from({ length: 15 }, (_, k) => +c[18 + k * 6 + 1]),
    });
  }
  if (!recs.size) return null;

  // L'horloge des modules derive de plusieurs heures. Plutot que de la
  // demander, on la DEDUIT : l'enregistrement le plus RECENT d'un export est,
  // a une demi-periode pres, contemporain de l'export lui-meme.
  //
  // ⚠️ Le plus recent n'est PAS celui de plus grand index : le tampon du BMS
  // est circulaire, il reecrit par-dessus le plus ancien. Trier par index
  // donnait des decalages de -865 h.
  const rangs = [...recs.keys()].sort((x, y) => recs.get(x).brut - recs.get(y).brut);
  // Depuis le 05/09, `pylontech-console.mjs --historique` inscrit le decalage
  // en tete, mesure a la seconde de l'export. C'est la seule valeur exacte :
  // on la prefere toujours a la deduction.
  const entete = txt.match(/^# decalage_minutes\s*:\s*(-?\d+)/m);
  const force = args.find((a) => a.startsWith('--decalage='));
  let decalageMin, origine;
  if (force) { decalageMin = +force.split('=')[1]; origine = 'impose'; }
  else if (entete) { decalageMin = +entete[1]; origine = 'mesure a l\'export'; }
  else {
    const dernier = recs.get(rangs[rangs.length - 1]).brut;
    decalageMin = Math.round((dernier - fs.statSync(f).mtimeMs) / 60000);
    origine = 'DEDUIT, approximatif';
  }
  const decalageMs = decalageMin * 60000;
  // Garde : au-dela d'une journee, la deduction ne tient plus (fichier
  // recopie, horodatage de fichier perdu). On refuse plutot que de deplacer
  // des mesures n'importe ou.
  if (Math.abs(decalageMin) > 24 * 60) {
    throw new Error(`decalage deduit aberrant (${Math.round(decalageMin / 60)} h)`);
  }

  const lignes = [];
  for (const r of rangs) {
    const x = recs.get(r);
    // Le module ecrit en heure locale de Paris : on repasse par l'offset reel.
    const t = new Date(x.brut - decalageMs);
    const local = new Date(t.getTime() - new Date(t).getTimezoneOffset() * 60000);
    for (let k = 0; k < 15; k++) {
      lignes.push([local.toISOString(), bc[1], k + 1, x.mv[k], x.soc, x.courant, x.etat, 'historique']);
    }
  }
  return { module: bc[1], decalageMin, origine, lignes, nb: rangs.length };
}

// --------------------------------------------------------------------- suivi
function lireSuivi(f) {
  const lignes = [];
  const modules = new Set();
  for (const l of lireTexte(f).trim().split('\n').slice(1)) {
    const c = l.split(',');
    if (c.length < 18) continue;
    const ad = parseInt(String(c[1]).replace(/\D/g, ''), 10);
    if (!Number.isFinite(ad)) continue;
    const mod = CARTE_SUIVI[ad] || `pack_ancien:${ad}`;
    modules.add(mod);
    const cour = c[2] === '' ? null : +c[2];
    for (let k = 0; k < 15; k++) {
      const mv = +c[3 + k];
      if (!(mv > 1000)) continue;
      lignes.push([c[0], mod, k + 1, mv, null, cour, c[1] || null, 'suivi']);
    }
  }
  return lignes.length ? { modules: [...modules], lignes } : null;
}

// ---------------------------------------------------------------------- main
const fichiers = [];
for (const m of motifs) {
  const d = path.dirname(m), base = path.basename(m);
  const re = new RegExp('^' + base.replace(/[.+^${}()|[\]\\]/g, '\\$&').replace(/\*/g, '.*') + '$');
  for (const n of fs.readdirSync(d)) if (re.test(n)) fichiers.push(path.join(d, n));
}
if (!fichiers.length) { console.error('Aucun fichier.'); process.exit(1); }

const paquets = [];
for (const f of fichiers.sort()) {
  const n = path.basename(f);
  try {
    if (n.endsWith('-suivi.csv')) {
      const r = lireSuivi(f);
      if (r) { paquets.push(r.lignes); console.log(`  ${n}  suivi, ${r.lignes.length / 15} releves, modules ${r.modules.join(' ')}`); }
    } else if (/histor/i.test(n) && n.endsWith('.txt')) {
      // Les exports d'EVENEMENTS portent les memes colonnes mais decrivent des
      // declenchements, pas un echantillonnage regulier : les melanger ferait
      // des courbes fausses.
      const r = lireHistorique(f);
      if (r) {
        paquets.push(r.lignes);
        const d = r.decalageMin;
        const t0 = r.lignes[0][0].slice(0, 16).replace('T', ' ');
        const t1 = r.lignes[r.lignes.length - 1][0].slice(0, 16).replace('T', ' ');
        console.log(`  ${n}\n      ${r.module}, ${r.nb} enregistrements, horloge `
          + `${d >= 0 ? '+' : '-'}${Math.trunc(Math.abs(d) / 60)} h ${Math.abs(d % 60)} min`
          + ` (${r.origine})  ->  ${t0} a ${t1}`);
      }
    }
  } catch (e) { console.error(`  ${n}  ignore : ${e.message}`); }
}
const toutes = paquets.flat();
console.log(`\n${toutes.length} lignes cellule au total.`);
if (BLANC) { console.log('A blanc : rien ecrit.'); process.exit(0); }

const client = new pg.Client({
  host: process.env.PGHOST || '10.10.10.22',
  port: Number(process.env.PGPORT || 5432),
  user: process.env.PGUSER || 'evcs',
  database: process.env.PGDATABASE || 'evcs',
  password: process.env.PGPASSWORD,
});
await client.connect();
let ecrites = 0;
for (let i = 0; i < toutes.length; i += 500) {
  const lot = toutes.slice(i, i + 500);
  const val = lot.map((_, k) => `($${k * 8 + 1},$${k * 8 + 2},$${k * 8 + 3},$${k * 8 + 4},$${k * 8 + 5},$${k * 8 + 6},$${k * 8 + 7},$${k * 8 + 8})`).join(',');
  const r = await client.query(
    `INSERT INTO cellule (ts, module, cellule, mv, soc, courant_ma, etat, source)
     VALUES ${val} ON CONFLICT (ts, module, cellule) DO NOTHING`, lot.flat());
  ecrites += r.rowCount;
  process.stdout.write(`\r  ${i + lot.length} / ${toutes.length}   `);
}
console.log(`\n${ecrites} lignes nouvelles, ${toutes.length - ecrites} deja presentes.`);
await client.end();