src/lib/server/controllerReader.js - master - Alcatraz/Wagomon

src/lib/server/controllerReader.js
300 строк10 KB
saitgalineu
vkr
26 май 2026, 10:08
26 май 2026, 10:08f9b2919
import { db } from './db.js';
import { readRegisters, readCoils } from './modbusClient.js';
import { dispatchAlertEvents, queueAnalogAlert, queueBinaryAlert, queueControllerAlert } from './alerts.js';
import net from 'net';
import { convertAnalog, convertBinary, buildRange, resolveRawValue } from './controllerTransforms.js';
import { insertAnalogReadings, insertBinaryReadings } from './sensorDataRepo.js';
import { getSystemSettings } from './systemSettings.js';
import { fromCelsiusToUnit, isTemperatureMeasureName } from './temperature.js';

const lastBinaryValues = new Map();

function shouldInsertBinary(sensorId, value) {
	const boolValue = value === null || value === undefined ? null : !!value;
	if (boolValue === null) return false;

	const prev = lastBinaryValues.get(sensorId);
	if (!prev) {
		lastBinaryValues.set(sensorId, { value: boolValue });
		return true;
	}

	if (prev.value === boolValue) {
		return false;
	}

	lastBinaryValues.set(sensorId, { value: boolValue });
	return true;
}

async function touchPollerHeartbeat() {
	try {
		await db.query(
			`INSERT INTO public.poller_heartbeat (id, updated_at)
			 VALUES (TRUE, NOW())
			 ON CONFLICT (id) DO UPDATE SET updated_at = EXCLUDED.updated_at`
		);
	} catch (err) {
		console.error('Failed to touch poller heartbeat:', err?.message ?? err);
	}
}

async function upsertControllerStatus(wagoId, status, message) {
	try {
		await db.query(
			`INSERT INTO public.controller_status (wago_id, status, message, updated_at)
			 VALUES ($1, $2, $3, NOW())
			 ON CONFLICT (wago_id)
			 DO UPDATE SET status = EXCLUDED.status, message = EXCLUDED.message, updated_at = NOW()`,
			[wagoId, status, message]
		);
	} catch (err) {
		console.error(
			`Failed to persist controller_status for wago_id=${wagoId}:`,
			err?.message ?? err
		);
	}
}

async function tcpPing(ip, port = 502, timeoutMs = 1000) {
    return new Promise((resolve) => {
        try {
            const socket = new net.Socket();
            let done = false;

            const finalize = (ok) => {
                if (done) return;
                done = true;
                try { socket.destroy(); } catch {}
                resolve(ok);
            };

            socket.setTimeout(timeoutMs);
            socket.once('connect', () => finalize(true));
            socket.once('timeout', () => finalize(false));
            socket.once('error', () => finalize(false));

            socket.connect(port, ip);
        } catch {
            resolve(false);
        }
    });
}

export async function pollControllers() {
	const result = {};
	const alertEvents = [];

	const { temperature_unit } = await getSystemSettings();

	const controllers = (
		await db.query(
			'SELECT * FROM wago WHERE (analog = TRUE OR "binary" = TRUE) ORDER BY sort_order ASC NULLS LAST, id ASC'
		)
	).rows;

	for (const [ctrlIdx, ctrl] of controllers.entries()) {
        const ip = ctrl.ip;
        const unitId = 1; 
        const ctrlKey = `controller_${ctrl.id}`; 
        const ctrlLabel = ctrl.name || `Controller ${ctrl.id}`;
		let controllerStatus = 'ok';
		let controllerMessage = null;

		const analogRows = ctrl.analog
			? (
					await db.query(
						`SELECT sa.*,
                            af.id   AS formula_id_join,
                            af.name AS formula_name,
                            af.description AS formula_description,
                            af.input_type AS formula_input_type,
                            af.expression AS formula_expression,
                            mt.name AS measure_type_name
                     FROM sensors_analog sa
                     LEFT JOIN analog_formulas af ON sa.formula_id = af.id
                     LEFT JOIN measure_types mt ON sa.measure_type_id = mt.id
                     WHERE sa.wago_id = $1
                     ORDER BY sa.sort_order ASC NULLS LAST, sa.connection_id ASC NULLS LAST, sa.id ASC`,
						[ctrl.id]
					)
			  ).rows.map((row) => ({
					...row,
					formula: row.formula_expression
						? {
								id: row.formula_id_join,
								name: row.formula_name,
								description: row.formula_description,
								input_type: row.formula_input_type,
								expression: row.formula_expression
						  }
						: null,
					measure_type: row.measure_type_name ? { name: row.measure_type_name } : null
			  }))
			: [];
        const binaryRows = ctrl.binary
            ? (
                await db.query(
                    'SELECT * FROM sensors_binary WHERE wago_id = $1 ORDER BY sort_order ASC NULLS LAST, connection_id ASC NULLS LAST, id ASC',
                    [ctrl.id]
                )
              ).rows
            : [];

		const analogPlaceholder = analogRows.length
			? analogRows.map((row) => {
					const isTemp = isTemperatureMeasureName(row.measure_type?.name ?? row.measure_type_name);
					const baseCritMin = Number(row.crit_min ?? row.min_value ?? null);
					const baseCritMax = Number(row.crit_max ?? row.max_value ?? null);

					const crit_min = isTemp ? fromCelsiusToUnit(baseCritMin, temperature_unit) : baseCritMin;
					const crit_max = isTemp ? fromCelsiusToUnit(baseCritMax, temperature_unit) : baseCritMax;

					return {
						id: row.id,
						name: row.name,
						sort_order: row.sort_order ?? null,
						connection_id: row.connection_id,
						value: null,
						crit_min,
						crit_max,
						description: row.description || null
					};
			  })
			: undefined;
        const binaryPlaceholder = binaryRows.length
            ? binaryRows.map(row => ({
                id: row.id,
                name: row.name,
                sort_order: row.sort_order ?? null,
                connection_id: row.connection_id,
                value: null,
                description: row.description || null
            }))
            : undefined;

        result[ctrlKey] = {
            analog: analogPlaceholder,
            binary: binaryPlaceholder
        };

        try {
            if (!ip) {
                throw new Error('IP не задан');
            }
            const reachable = await tcpPing(ip, 502, 1200);
            if (!reachable) {
                throw new Error('Контроллер недоступен');
            }

			if (analogRows.length > 0) {
				const range = buildRange(analogRows);
				if (range) {
					const analogData = await readRegisters(ip, unitId, range.start, range.length);
					if (!analogData) throw new Error('Контроллер недоступен');
					const analogResults = [];
					const analogInsertBuffer = [];
					for (const row of analogRows) {
						const raw = resolveRawValue(row, range, analogData);
						const baseValue = convertAnalog(raw, row);

						const isTemp = isTemperatureMeasureName(row.measure_type?.name ?? row.measure_type_name);
						const baseCritMin = Number(row.crit_min ?? row.min_value ?? null);
						const baseCritMax = Number(row.crit_max ?? row.max_value ?? null);

						const value = isTemp ? fromCelsiusToUnit(baseValue, temperature_unit) : baseValue;
						const crit_min = isTemp ? fromCelsiusToUnit(baseCritMin, temperature_unit) : baseCritMin;
						const crit_max = isTemp ? fromCelsiusToUnit(baseCritMax, temperature_unit) : baseCritMax;

						analogResults.push({
							id: row.id,
							name: row.name,
							sort_order: row.sort_order ?? null,
							connection_id: row.connection_id,
							value,
							crit_min,
							crit_max,
							description: row.description || null
						});

						if (baseValue != null) {
							analogInsertBuffer.push({ sensorId: row.id, value: baseValue });
						}
						queueAnalogAlert(row, baseValue, alertEvents, ctrlLabel);
					}
					if (analogInsertBuffer.length > 0) {
						await insertAnalogReadings(analogInsertBuffer);
					}
					result[ctrlKey].analog = analogResults;
				}
			}

            if (binaryRows.length > 0) {
                const range = buildRange(binaryRows);
                if (range) {
                    const binaryData = await readCoils(ip, unitId, range.start, range.length);
                    if (!binaryData) throw new Error('Контроллер недоступен');
                    const binaryResults = [];
                    const binaryInsertBuffer = [];
                    for (const row of binaryRows) {
                        const raw = resolveRawValue(row, range, binaryData);
                        const value = convertBinary(raw, row);
                        binaryResults.push({
                            id: row.id,
                            name: row.name,
                            sort_order: row.sort_order ?? null,
                            connection_id: row.connection_id,
                            value,
                            description: row.description || null
                        });

                        if (value != null && shouldInsertBinary(row.id, value)) {
                            binaryInsertBuffer.push({ sensorId: row.id, value });
                        }
                        queueBinaryAlert(row, value, alertEvents, ctrlLabel);
                    }
                    if (binaryInsertBuffer.length > 0) {
                        await insertBinaryReadings(binaryInsertBuffer);
                    }
                    result[ctrlKey].binary = binaryResults;
                }
            }
            queueControllerAlert(
                ctrlKey,
                ctrlLabel,
                ctrl.id,
                'ok',
                null,
                alertEvents,
                ctrlIdx,
                ctrl.alert_status || 'INFO',
                ctrl.location
            );
        } catch (err) {
			controllerStatus = 'down';
			controllerMessage = err?.message ?? String(err);
            result[ctrlKey].error = `Ошибка: ${err.message}`;
            queueControllerAlert(
                ctrlKey,
                ctrlLabel,
                ctrl.id,
                'down',
                err.message,
                alertEvents,
                ctrlIdx,
                ctrl.alert_status || 'INFO',
                ctrl.location
            );
        } finally {
			await upsertControllerStatus(ctrl.id, controllerStatus, controllerMessage);
		}
    }

    try {
		await dispatchAlertEvents(alertEvents);
	} finally {
		await touchPollerHeartbeat();
	}

    return result;
}