You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
549 lines
17 KiB
JavaScript
549 lines
17 KiB
JavaScript
const fs = require('node:fs');
|
|
const fsPromises = require('node:fs/promises');
|
|
const path = require('node:path');
|
|
const { execFile } = require('node:child_process');
|
|
const { promisify } = require('node:util');
|
|
|
|
const execFileAsync = promisify(execFile);
|
|
|
|
const ROOT_DIRECTORY = path.join(__dirname, '..');
|
|
const DATA_DIRECTORY = process.env.DATA_DIRECTORY
|
|
? path.resolve(process.env.DATA_DIRECTORY)
|
|
: path.join(ROOT_DIRECTORY, 'data');
|
|
const LOGS_DIRECTORY = process.env.LOGS_DIRECTORY
|
|
? path.resolve(process.env.LOGS_DIRECTORY)
|
|
: path.join(ROOT_DIRECTORY, 'logs');
|
|
const RUNTIME_CONFIG_FILE = process.env.RUNTIME_CONFIG_FILE
|
|
? path.resolve(process.env.RUNTIME_CONFIG_FILE)
|
|
: path.join(ROOT_DIRECTORY, 'config', 'runtime.json');
|
|
const ALARM_CONFIG_FILE = process.env.ALARM_CONFIG_FILE
|
|
? path.resolve(process.env.ALARM_CONFIG_FILE)
|
|
: path.join(ROOT_DIRECTORY, 'config', 'alarms.json');
|
|
const ACQUISITION_HELPER = process.env.EZO_ACQUIRE_HELPER
|
|
? path.resolve(process.env.EZO_ACQUIRE_HELPER)
|
|
: path.join(ROOT_DIRECTORY, 'sensors', 'EZOCommand', 'EZO_ACQUIRE');
|
|
|
|
const SENSOR_FILES = {
|
|
temperature: {
|
|
json: 'EZORTD.json',
|
|
csv: 'temperature.csv',
|
|
key: 'temperature'
|
|
},
|
|
ph: { json: 'EZOPH.json', csv: 'ph.csv', key: 'ph' },
|
|
do: { json: 'EZODO.json', csv: 'do.csv', key: 'do' },
|
|
ec: { json: 'EZOEC.json', csv: 'ec.csv', key: 'ec' }
|
|
};
|
|
|
|
const ALLOWED_LOGGING_RATES = new Set([1, 5, 10, 60]);
|
|
const ALLOWED_RETENTION_DAYS = new Set([0, 7, 30, 90, 365]);
|
|
const WARNING_MARGIN_RATIO = 0.1;
|
|
const ALARM_LOG_FILE = 'alarms.csv';
|
|
const ALARM_LOG_HEADER = 'timestamp,sensor,state,value,message\n';
|
|
const SENSOR_ALIASES = {
|
|
rtd: 'temperature',
|
|
temperature: 'temperature',
|
|
ph: 'ph',
|
|
do: 'do',
|
|
ec: 'ec'
|
|
};
|
|
const DEFAULT_ENABLED_SENSORS = Object.keys(SENSOR_FILES);
|
|
const previousAlarmStates = new Map();
|
|
const lastRetentionChecks = new Map();
|
|
const RETENTION_CHECK_INTERVAL_MS = 10 * 60 * 1000;
|
|
|
|
function normalizeEnabledSensors(value, fallback = DEFAULT_ENABLED_SENSORS) {
|
|
const source = Array.isArray(value)
|
|
? value
|
|
: String(value || '').split(',');
|
|
const sensors = [
|
|
...new Set(
|
|
source
|
|
.map((sensor) => SENSOR_ALIASES[String(sensor).trim().toLowerCase()])
|
|
.filter(Boolean)
|
|
)
|
|
];
|
|
|
|
return sensors.length > 0 ? sensors : [...fallback];
|
|
}
|
|
|
|
async function atomicWriteFile(filePath, content) {
|
|
await fsPromises.mkdir(path.dirname(filePath), { recursive: true });
|
|
const temporaryPath = `${filePath}.${process.pid}.${Date.now()}.tmp`;
|
|
await fsPromises.writeFile(temporaryPath, content, 'utf8');
|
|
await fsPromises.rename(temporaryPath, filePath);
|
|
}
|
|
|
|
async function readRuntimeConfig() {
|
|
try {
|
|
const rawConfig = await fsPromises.readFile(RUNTIME_CONFIG_FILE, 'utf8');
|
|
const config = JSON.parse(rawConfig);
|
|
const rate = Number(config.loggingRateSeconds);
|
|
const retentionDays = Number(config.historyRetentionDays);
|
|
const environmentSensors = String(process.env.EZO_ENABLED_SENSORS || '').trim();
|
|
|
|
return {
|
|
loggingRateSeconds: ALLOWED_LOGGING_RATES.has(rate) ? rate : 1,
|
|
historyRetentionDays: ALLOWED_RETENTION_DAYS.has(retentionDays)
|
|
? retentionDays
|
|
: 30,
|
|
enabledSensors: environmentSensors
|
|
? normalizeEnabledSensors(environmentSensors)
|
|
: normalizeEnabledSensors(config.enabledSensors)
|
|
};
|
|
} catch {
|
|
return {
|
|
loggingRateSeconds: 1,
|
|
historyRetentionDays: 30,
|
|
enabledSensors: normalizeEnabledSensors(process.env.EZO_ENABLED_SENSORS)
|
|
};
|
|
}
|
|
}
|
|
|
|
async function writeRuntimeConfig(updates) {
|
|
const currentConfig = await readRuntimeConfig();
|
|
const numericRate = updates.rate === undefined
|
|
? currentConfig.loggingRateSeconds
|
|
: Number(updates.rate);
|
|
const retentionDays = updates.historyRetentionDays === undefined
|
|
? currentConfig.historyRetentionDays
|
|
: Number(updates.historyRetentionDays);
|
|
|
|
if (!ALLOWED_LOGGING_RATES.has(numericRate)) {
|
|
const error = new Error('Frecuencia no valida.');
|
|
error.status = 400;
|
|
throw error;
|
|
}
|
|
|
|
if (!ALLOWED_RETENTION_DAYS.has(retentionDays)) {
|
|
const error = new Error('Retencion historica no valida.');
|
|
error.status = 400;
|
|
throw error;
|
|
}
|
|
|
|
const enabledSensors = updates.enabledSensors === undefined
|
|
? currentConfig.enabledSensors
|
|
: normalizeEnabledSensors(updates.enabledSensors, []);
|
|
|
|
if (enabledSensors.length === 0) {
|
|
const error = new Error('Debe habilitar al menos un sensor.');
|
|
error.status = 400;
|
|
throw error;
|
|
}
|
|
|
|
const config = {
|
|
loggingRateSeconds: numericRate,
|
|
historyRetentionDays: retentionDays,
|
|
enabledSensors,
|
|
updatedAt: new Date().toISOString()
|
|
};
|
|
await atomicWriteFile(
|
|
RUNTIME_CONFIG_FILE,
|
|
`${JSON.stringify(config, null, 2)}\n`
|
|
);
|
|
return config;
|
|
}
|
|
|
|
function detectAcquisitionMode() {
|
|
const requestedMode = String(process.env.EZO_MODE || 'auto').toLowerCase();
|
|
const hardwareReady = process.platform === 'linux' &&
|
|
fs.existsSync('/dev/i2c-1') &&
|
|
fs.existsSync(ACQUISITION_HELPER);
|
|
|
|
if (requestedMode === 'hardware' && !hardwareReady) {
|
|
return {
|
|
mode: 'unavailable',
|
|
message: 'Se solicito hardware, pero /dev/i2c-1 o EZO_ACQUIRE no esta disponible.'
|
|
};
|
|
}
|
|
|
|
if (requestedMode === 'hardware' || (requestedMode === 'auto' && hardwareReady)) {
|
|
return { mode: 'hardware', message: 'Adquisicion I2C real activa.' };
|
|
}
|
|
|
|
return { mode: 'demo', message: 'Adquisicion de demostracion activa.' };
|
|
}
|
|
|
|
function buildDemoReadings() {
|
|
return {
|
|
temperature: Number((25 + Math.random() * 0.1 - 0.05).toFixed(3)),
|
|
ph: Number((7.2 + Math.random() * 0.04 - 0.02).toFixed(3)),
|
|
do: Number((8.5 + Math.random() * 0.1 - 0.05).toFixed(3)),
|
|
ec: Number((1050 + Math.random() * 10 - 5).toFixed(1))
|
|
};
|
|
}
|
|
|
|
async function readHardwareSensors(enabledSensors) {
|
|
const helperArguments = ['/dev/i2c-1', ...enabledSensors];
|
|
const { stdout } = await execFileAsync(
|
|
ACQUISITION_HELPER,
|
|
helperArguments,
|
|
{ timeout: 5000, windowsHide: true }
|
|
);
|
|
const readings = JSON.parse(stdout.trim());
|
|
|
|
if (!readings || typeof readings !== 'object') {
|
|
throw new Error('EZO_ACQUIRE devolvio una respuesta invalida.');
|
|
}
|
|
|
|
return readings;
|
|
}
|
|
|
|
async function appendReading(filePath, timestamp, value) {
|
|
await fsPromises.mkdir(path.dirname(filePath), { recursive: true });
|
|
let needsHeader = false;
|
|
|
|
try {
|
|
const stats = await fsPromises.stat(filePath);
|
|
needsHeader = stats.size === 0;
|
|
} catch (error) {
|
|
if (error.code !== 'ENOENT') throw error;
|
|
needsHeader = true;
|
|
}
|
|
|
|
const row = `${timestamp},${value}\n`;
|
|
await fsPromises.appendFile(
|
|
filePath,
|
|
`${needsHeader ? 'timestamp,value\n' : ''}${row}`,
|
|
'utf8'
|
|
);
|
|
}
|
|
|
|
function escapeCsvValue(value) {
|
|
const text = value === null || value === undefined ? '' : String(value);
|
|
return `"${text.replace(/"/g, '""')}"`;
|
|
}
|
|
|
|
async function appendCsvRow(filePath, header, values) {
|
|
await fsPromises.mkdir(path.dirname(filePath), { recursive: true });
|
|
let needsHeader = false;
|
|
|
|
try {
|
|
const stats = await fsPromises.stat(filePath);
|
|
needsHeader = stats.size === 0;
|
|
} catch (error) {
|
|
if (error.code !== 'ENOENT') throw error;
|
|
needsHeader = true;
|
|
}
|
|
|
|
const row = `${values.map(escapeCsvValue).join(',')}\n`;
|
|
await fsPromises.appendFile(
|
|
filePath,
|
|
`${needsHeader ? header : ''}${row}`,
|
|
'utf8'
|
|
);
|
|
}
|
|
|
|
async function pruneCsvByRetention(filePath, header, retentionDays, now = new Date()) {
|
|
if (!retentionDays) return false;
|
|
|
|
let csvText = '';
|
|
|
|
try {
|
|
csvText = await fsPromises.readFile(filePath, 'utf8');
|
|
} catch (error) {
|
|
if (error.code === 'ENOENT') return false;
|
|
throw error;
|
|
}
|
|
|
|
const lines = csvText.split(/\r?\n/).filter(Boolean);
|
|
|
|
if (lines.length <= 1) return false;
|
|
|
|
const cutoffMs = now.getTime() - retentionDays * 24 * 60 * 60 * 1000;
|
|
const rows = lines.slice(1);
|
|
const keptRows = rows.filter((line) => {
|
|
const timestamp = line.split(',', 1)[0].replace(/^"|"$/g, '');
|
|
const parsedMs = new Date(timestamp).getTime();
|
|
|
|
return Number.isNaN(parsedMs) || parsedMs >= cutoffMs;
|
|
});
|
|
|
|
if (keptRows.length === rows.length) return false;
|
|
|
|
await atomicWriteFile(
|
|
filePath,
|
|
`${header}${keptRows.length > 0 ? `${keptRows.join('\n')}\n` : ''}`
|
|
);
|
|
return true;
|
|
}
|
|
|
|
async function enforceHistoryRetention(retentionDays, now = new Date()) {
|
|
if (!retentionDays) return [];
|
|
|
|
const prunedFiles = [];
|
|
const logFiles = [
|
|
...Object.values(SENSOR_FILES).map((sensor) => ({
|
|
file: sensor.csv,
|
|
header: 'timestamp,value\n'
|
|
})),
|
|
{ file: ALARM_LOG_FILE, header: ALARM_LOG_HEADER }
|
|
];
|
|
|
|
for (const logFile of logFiles) {
|
|
const filePath = path.join(LOGS_DIRECTORY, logFile.file);
|
|
const lastCheck = lastRetentionChecks.get(filePath) || 0;
|
|
|
|
if (now.getTime() - lastCheck < RETENTION_CHECK_INTERVAL_MS) {
|
|
continue;
|
|
}
|
|
|
|
lastRetentionChecks.set(filePath, now.getTime());
|
|
|
|
if (await pruneCsvByRetention(filePath, logFile.header, retentionDays, now)) {
|
|
prunedFiles.push(logFile.file);
|
|
}
|
|
}
|
|
|
|
return prunedFiles;
|
|
}
|
|
|
|
async function readAlarmThresholds() {
|
|
try {
|
|
return JSON.parse(await fsPromises.readFile(ALARM_CONFIG_FILE, 'utf8'));
|
|
} catch {
|
|
return {};
|
|
}
|
|
}
|
|
|
|
function isValidThreshold(limits) {
|
|
return Boolean(limits) &&
|
|
Number.isFinite(Number(limits.min)) &&
|
|
Number.isFinite(Number(limits.max)) &&
|
|
Number(limits.min) < Number(limits.max);
|
|
}
|
|
|
|
function isNearThreshold(value, limits) {
|
|
const min = Number(limits.min);
|
|
const max = Number(limits.max);
|
|
const warningMargin = (max - min) * WARNING_MARGIN_RATIO;
|
|
|
|
return value <= min + warningMargin || value >= max - warningMargin;
|
|
}
|
|
|
|
function evaluateAlarmState(sensorId, result, thresholds) {
|
|
if (result.disabled) {
|
|
return {
|
|
state: 'DISABLED',
|
|
message: 'Sensor deshabilitado por configuracion'
|
|
};
|
|
}
|
|
|
|
if (!result.online) {
|
|
return {
|
|
state: 'OFFLINE',
|
|
message: 'Lectura no disponible'
|
|
};
|
|
}
|
|
|
|
const limits = thresholds[sensorId];
|
|
|
|
if (!isValidThreshold(limits)) {
|
|
return {
|
|
state: 'WARNING',
|
|
message: 'Limites de alarma no disponibles'
|
|
};
|
|
}
|
|
|
|
if (result.value < Number(limits.min) || result.value > Number(limits.max)) {
|
|
return {
|
|
state: 'CRITICAL',
|
|
message: `Valor fuera de rango (${limits.min} - ${limits.max})`
|
|
};
|
|
}
|
|
|
|
if (isNearThreshold(result.value, limits)) {
|
|
return {
|
|
state: 'WARNING',
|
|
message: `Valor cerca del limite (${limits.min} - ${limits.max})`
|
|
};
|
|
}
|
|
|
|
return {
|
|
state: 'NORMAL',
|
|
message: 'Dentro del rango configurado'
|
|
};
|
|
}
|
|
|
|
async function persistAlarmEvents(results, timestamp) {
|
|
const thresholds = await readAlarmThresholds();
|
|
const events = [];
|
|
|
|
for (const result of results) {
|
|
const evaluation = evaluateAlarmState(result.sensorId, result, thresholds);
|
|
const previousState = previousAlarmStates.get(result.sensorId) || 'NORMAL';
|
|
|
|
if (evaluation.state === 'DISABLED') {
|
|
previousAlarmStates.delete(result.sensorId);
|
|
continue;
|
|
}
|
|
|
|
const enteredAlarm = ['WARNING', 'CRITICAL', 'OFFLINE'].includes(evaluation.state) &&
|
|
previousState !== evaluation.state;
|
|
const recovered = evaluation.state === 'NORMAL' &&
|
|
['WARNING', 'CRITICAL', 'OFFLINE'].includes(previousState);
|
|
|
|
if (enteredAlarm || recovered) {
|
|
events.push({
|
|
timestamp,
|
|
sensorId: result.sensorId,
|
|
state: recovered ? 'RECOVERY' : evaluation.state,
|
|
value: Number.isFinite(result.value) ? result.value : '',
|
|
message: recovered ? 'Sensor recuperado a NORMAL' : evaluation.message
|
|
});
|
|
}
|
|
|
|
previousAlarmStates.set(result.sensorId, evaluation.state);
|
|
}
|
|
|
|
for (const event of events) {
|
|
await appendCsvRow(
|
|
path.join(LOGS_DIRECTORY, ALARM_LOG_FILE),
|
|
ALARM_LOG_HEADER,
|
|
[
|
|
event.timestamp,
|
|
event.sensorId,
|
|
event.state,
|
|
event.value,
|
|
event.message
|
|
]
|
|
);
|
|
}
|
|
|
|
return events;
|
|
}
|
|
|
|
async function publishReading(sensorId, rawValue, timestamp, mode) {
|
|
const sensor = SENSOR_FILES[sensorId];
|
|
const jsonPath = path.join(DATA_DIRECTORY, sensor.json);
|
|
const hasReading = rawValue !== null &&
|
|
rawValue !== undefined &&
|
|
rawValue !== '';
|
|
const value = hasReading ? Number(rawValue) : NaN;
|
|
|
|
if (!Number.isFinite(value)) {
|
|
await atomicWriteFile(jsonPath, `${JSON.stringify({
|
|
online: false,
|
|
timestamp,
|
|
error: 'Lectura no disponible'
|
|
})}\n`);
|
|
return false;
|
|
}
|
|
|
|
await Promise.all([
|
|
atomicWriteFile(jsonPath, `${JSON.stringify({
|
|
[sensor.key]: value,
|
|
online: true,
|
|
timestamp,
|
|
mode
|
|
})}\n`),
|
|
appendReading(path.join(LOGS_DIRECTORY, sensor.csv), timestamp, value)
|
|
]);
|
|
return true;
|
|
}
|
|
|
|
async function publishDisabled(sensorId, timestamp) {
|
|
const sensor = SENSOR_FILES[sensorId];
|
|
|
|
await atomicWriteFile(
|
|
path.join(DATA_DIRECTORY, sensor.json),
|
|
`${JSON.stringify({
|
|
online: false,
|
|
disabled: true,
|
|
timestamp,
|
|
error: 'Sensor deshabilitado por configuracion'
|
|
})}\n`
|
|
);
|
|
}
|
|
|
|
async function collectReadings() {
|
|
const modeInfo = detectAcquisitionMode();
|
|
const runtimeConfig = await readRuntimeConfig();
|
|
const enabledSensorSet = new Set(runtimeConfig.enabledSensors);
|
|
|
|
if (modeInfo.mode === 'unavailable') {
|
|
throw new Error(modeInfo.message);
|
|
}
|
|
|
|
const readings = modeInfo.mode === 'hardware'
|
|
? await readHardwareSensors(runtimeConfig.enabledSensors)
|
|
: buildDemoReadings();
|
|
const timestamp = new Date().toISOString();
|
|
const results = await Promise.all(
|
|
Object.keys(SENSOR_FILES).map(async (sensorId) => {
|
|
if (!enabledSensorSet.has(sensorId)) {
|
|
await publishDisabled(sensorId, timestamp);
|
|
return { sensorId, online: false, disabled: true, value: null };
|
|
}
|
|
|
|
const rawValue = readings[sensorId];
|
|
const value = rawValue === null || rawValue === undefined || rawValue === ''
|
|
? NaN
|
|
: Number(rawValue);
|
|
const online = await publishReading(
|
|
sensorId,
|
|
rawValue,
|
|
timestamp,
|
|
modeInfo.mode
|
|
);
|
|
|
|
return {
|
|
sensorId,
|
|
online,
|
|
value: Number.isFinite(value) ? value : null,
|
|
disabled: false
|
|
};
|
|
})
|
|
);
|
|
const alarmEvents = await persistAlarmEvents(results, timestamp);
|
|
const prunedFiles = await enforceHistoryRetention(
|
|
runtimeConfig.historyRetentionDays,
|
|
new Date(timestamp)
|
|
);
|
|
|
|
return {
|
|
mode: modeInfo.mode,
|
|
timestamp,
|
|
results,
|
|
alarmEvents,
|
|
prunedFiles,
|
|
enabledSensors: runtimeConfig.enabledSensors
|
|
};
|
|
}
|
|
|
|
async function publishAllOffline(error) {
|
|
const timestamp = new Date().toISOString();
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
|
|
await Promise.all(
|
|
Object.values(SENSOR_FILES).map((sensor) =>
|
|
atomicWriteFile(
|
|
path.join(DATA_DIRECTORY, sensor.json),
|
|
`${JSON.stringify({
|
|
online: false,
|
|
timestamp,
|
|
error: message
|
|
})}\n`
|
|
)
|
|
)
|
|
);
|
|
}
|
|
|
|
module.exports = {
|
|
ACQUISITION_HELPER,
|
|
ALLOWED_LOGGING_RATES,
|
|
ALLOWED_RETENTION_DAYS,
|
|
DEFAULT_ENABLED_SENSORS,
|
|
RUNTIME_CONFIG_FILE,
|
|
SENSOR_FILES,
|
|
atomicWriteFile,
|
|
collectReadings,
|
|
detectAcquisitionMode,
|
|
enforceHistoryRetention,
|
|
normalizeEnabledSensors,
|
|
publishAllOffline,
|
|
publishReading,
|
|
pruneCsvByRetention,
|
|
readRuntimeConfig,
|
|
writeRuntimeConfig
|
|
};
|