Hannes/SE-LTD
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(bacnet-collector): add initial implementation of BACnet collector with scanning and reading capabilities

Browse Source 
- Created package.json for the bacnet-collector project with necessary dependencies.
- Implemented main server logic in src/index.js to handle BACnet device scanning and reading.
- Added raw discovery functionality in src/raw-discover.js to facilitate device detection over the network.
- Introduced utility functions for handling IP addresses, broadcasting, and BACnet packet construction.
- Implemented JSON response handling for health checks and capabilities endpoints.
This commit is contained in:
jhartworks
2026-07-09 11:13:26 +02:00
parent 7a1fe401f2
commit f247d2827d
10 changed files with 1833 additions and 77233 deletions
Download Patch File Download Diff File Expand all files Collapse all files
collector/src/index.js
+373 -29
View File
@@ -11,6 +11,7 @@ const dataDir = join(__dirname, "..", "data");
const storeFile = join(dataDir, "sources.json");
const port = Number(process.env.PORT || 18110);
const pollTickMs = Number(process.env.POLL_TICK_MS || 1000);
const bacnetCollectorUrl = String(process.env.BACNET_COLLECTOR_URL || "").replace(/\/$/, "");
const LOCAL_POINT_LIMIT = 2000;
let transactionId = 1;
let polling = false;
@@ -126,6 +127,16 @@ function normalizePollInterval(value) {
return Number.isFinite(numeric) && numeric > 0 ? Math.round(numeric) : 60;
}
function normalizeBacnetMode(value) {
const mode = String(value || "broadcast").trim().toLowerCase();
return ["broadcast", "bbmd", "foreign-device"].includes(mode) ? mode : "broadcast";
}
function normalizeBacnetTtl(value) {
const numeric = Number(value);
return Number.isFinite(numeric) && numeric > 0 ? Math.round(numeric) : 120;
}
function normalizePointKind(value) {
return String(value || "").toLowerCase() === "digital" ? "digital" : "analog";
}
@@ -188,10 +199,15 @@ function ensureSource(store, body) {
const now = new Date().toISOString();
const existingIndex = store.sources.findIndex((item) => item.id === body.id);
const existing = existingIndex >= 0 ? store.sources[existingIndex] : null;
const protocol = normalizeProtocol(body.protocol);
const bacnetMode = normalizeBacnetMode(body.bacnetMode || body.options?.bacnetMode || existing?.options?.bacnetMode);
const bacnetBroadcastAddress = normalizeText(body.bacnetBroadcastAddress || body.options?.bacnetBroadcastAddress || existing?.options?.bacnetBroadcastAddress || "255.255.255.255") || "255.255.255.255";
const bacnetForeignDeviceTtl = normalizeBacnetTtl(body.bacnetForeignDeviceTtl || body.options?.bacnetForeignDeviceTtl || existing?.options?.bacnetForeignDeviceTtl);
const bacnetScanSubnets = normalizeText(body.bacnetScanSubnets || body.options?.bacnetScanSubnets || existing?.options?.bacnetScanSubnets || "");
const tableName = sanitizeTrendTable(existing?.tableName || buildSourceTableName(body));
const source = {
id: body.id || `src-${Date.now()}`,
protocol: normalizeProtocol(body.protocol),
protocol,
name: normalizeText(body.name) || "Neue Quelle",
host: normalizeText(body.host),
port: normalizeText(body.port),
@@ -201,9 +217,14 @@ function ensureSource(store, body) {
writeMode: normalizeWriteMode(body.writeMode),
pollIntervalSeconds: normalizePollInterval(body.pollIntervalSeconds),
options: {
...(existing?.options || {}),
...(body.options || {}),
writeMode: normalizeWriteMode(body.writeMode),
pollIntervalSeconds: normalizePollInterval(body.pollIntervalSeconds),
bacnetMode,
bacnetBroadcastAddress,
bacnetForeignDeviceTtl,
bacnetScanSubnets,
},
lastValues: body.lastValues || {},
lastPollAt: body.lastPollAt || null,
@@ -416,12 +437,15 @@ function buildScannedDatapoints(store, source, rawPoints, toPayload) {
nextPointIndex = Math.max(nextPointIndex, pointIndex + 1);
}
const scanned = createScanDatapoint(source, { ...payload, pointIndex }, index);
const existingAlias = normalizeText(existing?.alias || "");
const existingName = normalizeText(existing?.name || "");
const existingLooksLikeScanDefault = existingAlias && [existingName, normalizeText(existing?.nodeId), normalizeText(existing?.address)].includes(existingAlias);
items.push(!existing ? scanned : {
...existing,
...scanned,
pointIndex: existing.pointIndex,
alias: existing.alias || scanned.alias,
name: existing.name || scanned.name,
alias: !existingAlias || existingLooksLikeScanDefault ? scanned.alias : existing.alias,
name: scanned.name || existing.name,
unit: normalizePointUnit(existing.unit || scanned.unit, existing.kind || scanned.kind),
kind: existing.kind || scanned.kind,
enabled: existing.enabled === true,
@@ -800,6 +824,248 @@ function bacnetObjectLabel(objectId) {
return prefix + objectId.instance;
}
function getBacnetRouting(sourceOrBody) {
return {
mode: normalizeBacnetMode(sourceOrBody?.bacnetMode || sourceOrBody?.options?.bacnetMode),
host: normalizeText(sourceOrBody?.host || sourceOrBody?.options?.bbmdHost || ""),
port: Number(sourceOrBody?.port || sourceOrBody?.options?.bbmdPort || 47808) || 47808,
broadcastAddress: normalizeText(sourceOrBody?.bacnetBroadcastAddress || sourceOrBody?.options?.bacnetBroadcastAddress || "255.255.255.255") || "255.255.255.255",
foreignDeviceTtl: normalizeBacnetTtl(sourceOrBody?.bacnetForeignDeviceTtl || sourceOrBody?.options?.bacnetForeignDeviceTtl),
};
}
function createBacnetClientInstance(bacnet, sourceOrBody, timeout) {
const routing = getBacnetRouting(sourceOrBody);
const client = new bacnet({
apduTimeout: timeout,
port: Number(sourceOrBody?.port || 47808) || 47808,
broadcastAddress: routing.broadcastAddress,
});
return { client, routing };
}
function getBacnetTransport(client) {
const transport = client?._transport;
if (!transport || typeof transport.send !== "function" || !transport._server) {
throw new Error("BACnet-Transport konnte nicht initialisiert werden.");
}
return transport;
}
async function waitForBacnetClientListening(client, timeout = 2000) {
const transport = getBacnetTransport(client);
try {
const address = transport._server.address?.();
if (address && typeof address.port === "number") {
return;
}
} catch {}
await new Promise((resolve, reject) => {
let settled = false;
const cleanup = () => {
client.off("listening", onListening);
client.off("error", onError);
clearTimeout(timer);
};
const finish = (error) => {
if (settled) return;
settled = true;
cleanup();
if (error) {
reject(error instanceof Error ? error : new Error(String(error)));
} else {
resolve();
}
};
const onListening = () => finish();
const onError = (error) => finish(error);
const timer = setTimeout(() => finish(new Error("BACnet-Socket wurde nicht rechtzeitig bereit.")), timeout);
client.once("listening", onListening);
client.once("error", onError);
});
}
function captureBacnetWhoIsPacket(client, deviceId) {
const transport = getBacnetTransport(client);
const originalSend = transport.send.bind(transport);
let packet = null;
transport.send = (buffer, offset) => {
packet = Buffer.from(buffer.slice(0, offset));
};
try {
const numericDeviceId = Number(deviceId);
if (Number.isFinite(numericDeviceId)) {
client.whoIs(undefined, { lowLimit: numericDeviceId, highLimit: numericDeviceId });
} else {
client.whoIs();
}
} finally {
transport.send = originalSend;
}
if (!packet || packet.length < 4 || packet[0] !== 0x81) {
throw new Error("BACnet Who-Is konnte nicht aufgebaut werden.");
}
return packet;
}
function bacnetResultText(resultCode) {
const known = {
0x0000: "Erfolgreich",
0x0010: "Write-Broadcast-Distribution-Table NAK",
0x0020: "Read-Broadcast-Distribution-Table NAK",
0x0030: "Register-Foreign-Device NAK",
0x0040: "Read-Foreign-Device-Table NAK",
0x0050: "Delete-Foreign-Device-Table-Entry NAK",
0x0060: "Distribute-Broadcast-To-Network NAK",
};
return known[resultCode] || `BVLC-Fehler ${resultCode}`;
}
async function sendBacnetControlPacket(client, receiver, packet, matcher, timeout = 2500) {
const transport = getBacnetTransport(client);
await waitForBacnetClientListening(client, timeout);
return await new Promise((resolve, reject) => {
let settled = false;
const socket = transport._server;
const cleanup = () => {
socket.off("message", onMessage);
clearTimeout(timer);
};
const finish = (error, result) => {
if (settled) return;
settled = true;
cleanup();
if (error) {
reject(error instanceof Error ? error : new Error(String(error)));
} else {
resolve(result);
}
};
const onMessage = (message, rinfo) => {
try {
if (!matcher(message, rinfo)) {
return;
}
finish(null, { message, rinfo });
} catch (error) {
finish(error);
}
};
const timer = setTimeout(() => finish(new Error("BACnet-Steuertelegramm hat nicht rechtzeitig geantwortet.")), timeout);
socket.on("message", onMessage);
try {
transport.send(packet, packet.length, receiver);
} catch (error) {
finish(error);
}
});
}
async function registerBacnetForeignDevice(client, routing) {
if (routing.mode !== "foreign-device") {
return;
}
if (!routing.host) {
throw new Error("F?r BACnet Foreign Device bitte BBMD Host/IP angeben.");
}
if (typeof client.registerForeignDevice === "function") {
await new Promise((resolve, reject) => {
let settled = false;
const done = (error) => {
if (settled) return;
settled = true;
if (error) {
reject(error instanceof Error ? error : new Error(String(error)));
} else {
resolve();
}
};
try {
const maybe = client.registerForeignDevice(routing.host, routing.port, routing.foreignDeviceTtl, done);
if (maybe && typeof maybe.then === "function") {
maybe.then(() => done(), done);
} else if (client.registerForeignDevice.length < 4) {
done();
} else {
setTimeout(() => done(), 1200);
}
} catch (error) {
done(error);
}
});
return;
}
const packet = Buffer.alloc(6);
packet[0] = 0x81;
packet[1] = 0x05;
packet.writeUInt16BE(6, 2);
packet.writeUInt16BE(routing.foreignDeviceTtl, 4);
const { message } = await sendBacnetControlPacket(
client,
routing.host,
packet,
(candidate, rinfo) => candidate?.length >= 6 && candidate[0] === 0x81 && candidate[1] === 0x00 && Number(rinfo?.port) === routing.port,
3000,
);
const resultCode = message.readUInt16BE(4);
if (resultCode !== 0x0000) {
throw new Error(`Foreign-Device-Register wurde abgelehnt: ${bacnetResultText(resultCode)}.`);
}
}
function sendBacnetWhoIs(client, routing, deviceId) {
if (routing.mode === "bbmd" || routing.mode === "foreign-device") {
if (!routing.host) {
throw new Error("F?r BACnet ?ber BBMD bitte Host/IP angeben.");
}
const packet = captureBacnetWhoIsPacket(client, deviceId);
packet[1] = 0x09;
const transport = getBacnetTransport(client);
transport.send(packet, packet.length, routing.host);
return;
}
const numericDeviceId = Number(deviceId);
const low = Number.isFinite(numericDeviceId) ? numericDeviceId : undefined;
const high = low;
const targets = [null];
const attempts = [];
const addAttempt = (fn) => attempts.push(fn);
for (const target of targets.length ? targets : [null]) {
addAttempt(() => (low === undefined ? (target ? client.whoIs(target) : client.whoIs()) : (target ? client.whoIs(low, high, target) : client.whoIs(low, high))));
addAttempt(() => client.whoIs({ lowLimit: low, highLimit: high, address: target || undefined }));
addAttempt(() => client.whoIs({ lowLimit: low, highLimit: high }));
addAttempt(() => (target ? client.whoIs(target, low, high) : client.whoIs(low, high)));
}
let lastError = null;
for (const attempt of attempts) {
try {
attempt();
return;
} catch (error) {
lastError = error;
}
}
if (lastError) {
throw lastError;
}
throw new Error("Who-Is konnte nicht ausgel?st werden.");
}
function readBacnetProperty(client, host, objectId, propertyId) {
return new Promise((resolve) => {
client.readProperty(host, objectId, propertyId, (error, value) => {
@@ -823,9 +1089,13 @@ async function scanBacnetObjects(client, device, portNumber) {
if (!objectId || objectId.type === 8 || !Number.isFinite(Number(objectId.instance))) {
continue;
}
const objectName = await readBacnetProperty(client, host, objectId, 77);
const description = await readBacnetProperty(client, host, objectId, 28);
const presentValue = await readBacnetProperty(client, host, objectId, 85);
const raw = presentValue?.values?.[0]?.value ?? null;
const label = bacnetObjectLabel(objectId);
const name = normalizeText(objectName?.values?.[0]?.value) || label;
const alias = normalizeText(description?.values?.[0]?.value) || name;
points.push({
id: host + "-" + objectId.type + "-" + objectId.instance,
host,
@@ -834,9 +1104,10 @@ async function scanBacnetObjects(client, device, portNumber) {
objectInstance: objectId.instance,
propertyId: 85,
address: String(objectId.instance),
dataType: label.replace(/[0-9]+$/, ""),
name: label,
alias: label,
dataType: bacnetObjectLabel(objectId).replace(/[0-9]+$/, ""),
name,
description: alias,
alias,
kind: typeof raw === "boolean" || objectId.type >= 3 && objectId.type <= 5 ? "digital" : "analog",
currentValue: raw,
hasValue: raw !== null,
@@ -845,67 +1116,134 @@ async function scanBacnetObjects(client, device, portNumber) {
return points;
}
async function requestExternalBacnetScan(body) {
if (!bacnetCollectorUrl) {
return null;
}
const response = await fetch(`${bacnetCollectorUrl}/scan/bacnet`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body || {}),
});
const data = await response.json().catch(() => ({}));
if (!response.ok) {
throw new Error(data.error || "BACnet-Scan im separaten Collector fehlgeschlagen.");
}
return data;
}
async function scanBacnetNetwork(body) {
const externalScan = await requestExternalBacnetScan(body);
if (externalScan) {
return externalScan;
}
const bacnet = await loadOptionalPackage("node-bacnet");
const portNumber = Number(body.port || 47808);
const timeout = Number(body.timeout || 6000);
const client = new bacnet({ apduTimeout: timeout, port: portNumber });
const { client, routing } = createBacnetClientInstance(bacnet, body, timeout);
const devices = [];
return await new Promise((resolve, reject) => {
const timer = setTimeout(async () => {
try {
for (const device of devices) {
const deviceObject = { type: 8, instance: Number(device.deviceId) };
const deviceName = await readBacnetProperty(client, device.address, deviceObject, 77);
const resolvedName = repairText(String(deviceName?.values?.[0]?.value || "").trim());
if (resolvedName) {
device.name = resolvedName;
}
device.points = await scanBacnetObjects(client, device, portNumber);
}
client.close();
resolve({ devices, points: devices.flatMap((device) => device.points || []), scannedAt: new Date().toISOString() });
try { client.close(); } catch {}
resolve({
routing,
devices,
points: devices.flatMap((device) => device.points || []),
scannedAt: new Date().toISOString(),
});
} catch (error) {
client.close();
try { client.close(); } catch {}
reject(error);
}
}, timeout);
client.on("iAm", (device) => {
if (!devices.some((item) => item.host === device.address && item.deviceId === device.deviceId)) {
const address = normalizeText(device?.address || device?.header?.sender?.address || "");
const deviceId = Number(device?.deviceId);
if (!address || !Number.isFinite(deviceId)) {
return;
}
if (!devices.some((item) => item.host === address && item.deviceId === deviceId)) {
devices.push({
host: device.address,
address,
host: address,
port: portNumber,
reachable: true,
deviceId: device.deviceId,
name: "BACnet Gerät " + device.deviceId,
deviceId,
name: "BACnet Ger?t " + deviceId,
vendorId: device?.vendorId ?? null,
maxApdu: device?.maxApdu ?? null,
segmentation: device?.segmentation ?? null,
points: [],
});
}
});
try {
if (body.deviceId) {
client.whoIs(Number(body.deviceId), Number(body.deviceId));
} else {
client.whoIs();
(async () => {
try {
if (routing.mode === "bbmd" && !routing.host) {
throw new Error("F?r BACnet ?ber BBMD bitte Host/IP angeben.");
}
await registerBacnetForeignDevice(client, routing);
sendBacnetWhoIs(client, routing, body.deviceId);
} catch (error) {
clearTimeout(timer);
try { client.close(); } catch {}
reject(error);
}
} catch (error) {
clearTimeout(timer);
client.close();
reject(error);
}
})();
});
}
async function requestExternalBacnetRead(source, datapoint) {
if (!bacnetCollectorUrl) {
return { delegated: false };
}
const response = await fetch(`${bacnetCollectorUrl}/read/bacnet`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ source, datapoint }),
});
const data = await response.json().catch(() => ({}));
if (!response.ok) {
throw new Error(data.error || "BACnet-Livewert im separaten Collector fehlgeschlagen.");
}
return { delegated: true, value: data.value };
}
async function readBacnetValue(source, datapoint) {
const externalRead = await requestExternalBacnetRead(source, datapoint);
if (externalRead.delegated) {
return externalRead.value;
}
const bacnet = await loadOptionalPackage("node-bacnet");
const client = new bacnet({ apduTimeout: 5000, port: Number(source.port || 47808) });
const { client, routing } = createBacnetClientInstance(bacnet, source, 5000);
const propertyId = Number(datapoint.propertyId || 85);
const objectId = parseBacnetObject(datapoint);
const host = String(source.host || datapoint.host || "").trim();
const host = String(datapoint.host || source.host || "").trim();
if (!host) {
throw new Error("BACnet Host fehlt.");
}
await registerBacnetForeignDevice(client, routing);
return await new Promise((resolve, reject) => {
client.readProperty(host, objectId, propertyId, (error, value) => {
client.close();
try { client.close(); } catch {}
if (error) {
reject(error);
return;
@@ -1834,6 +2172,9 @@ const server = createServer(async (req, res) => {
if (req.method === "POST" && url.pathname === "/scan/bacnet") {
try {
const body = await readBody(req);
if (!body.clientHost && req.socket?.remoteAddress) {
body.clientHost = String(req.socket.remoteAddress).replace(/^::ffff:/, "");
}
const payload = await scanBacnetNetwork(body);
let scannedCandidates = [];
touchStore((store) => {
@@ -1846,8 +2187,8 @@ const server = createServer(async (req, res) => {
id: point.id || `${source.id}::${point.host || source.host}::${point.dataType || point.objectType || "object"}::${point.objectInstance || index + 1}`,
protocol: "bacnet-ip",
pointIndex: index + 1,
alias: point.alias || point.name || `${point.dataType || "obj"}${point.objectInstance || index + 1}`,
name: point.name || point.alias || `${point.dataType || "obj"}${point.objectInstance || index + 1}`,
alias: point.description || point.alias || point.name || `${point.dataType || "obj"}${point.objectInstance || index + 1}`,
name: point.name || point.alias || point.description || `${point.dataType || "obj"}${point.objectInstance || index + 1}`,
}));
scannedCandidates = candidates;
upsertDatapoints(store, candidates);
@@ -1878,3 +2219,6 @@ server.listen(port, () => {