diff --git a/display/frontend/src/App.tsx b/display/frontend/src/App.tsx index 81b93b2..2d8c30a 100644 --- a/display/frontend/src/App.tsx +++ b/display/frontend/src/App.tsx @@ -6,6 +6,7 @@ import Dashboard from './pages/Dashboard'; import MqttBrokers from './pages/MqttBrokers'; import MqttTopics from './pages/MqttTopics'; import PgSources from './pages/PgSources'; +import IoBroker from './pages/IoBroker'; import Datapoints from './pages/Datapoints'; import Displays from './pages/Displays'; import Layouts from './pages/Layouts'; @@ -31,6 +32,7 @@ export default function App() { } /> } /> } /> + } /> } /> } /> } /> diff --git a/display/frontend/src/assets/hero.png b/display/frontend/src/assets/hero.png new file mode 100644 index 0000000..cc51a3d Binary files /dev/null and b/display/frontend/src/assets/hero.png differ diff --git a/display/frontend/src/assets/react.svg b/display/frontend/src/assets/react.svg new file mode 100644 index 0000000..6c87de9 --- /dev/null +++ b/display/frontend/src/assets/react.svg @@ -0,0 +1 @@ + \ No newline at end of file diff --git a/display/frontend/src/assets/vite.svg b/display/frontend/src/assets/vite.svg new file mode 100644 index 0000000..5101b67 --- /dev/null +++ b/display/frontend/src/assets/vite.svg @@ -0,0 +1 @@ +Vite diff --git a/display/frontend/src/components/Sidebar.tsx b/display/frontend/src/components/Sidebar.tsx index b8456d0..bcba802 100644 --- a/display/frontend/src/components/Sidebar.tsx +++ b/display/frontend/src/components/Sidebar.tsx @@ -5,6 +5,7 @@ const links = [ { to: '/', label: 'Dashboard', exact: true }, { to: '/mqtt/brokers', label: 'MQTT Broker' }, { to: '/pg/sources', label: 'PostgreSQL Quellen' }, + { to: '/iobroker', label: 'ioBroker' }, { to: '/datapoints', label: 'Datenpunkte' }, { to: '/displays', label: 'Displays' }, { to: '/layouts', label: 'Layouts' }, diff --git a/display/frontend/src/pages/IoBroker.tsx b/display/frontend/src/pages/IoBroker.tsx new file mode 100644 index 0000000..b8c1a37 --- /dev/null +++ b/display/frontend/src/pages/IoBroker.tsx @@ -0,0 +1,367 @@ +import { useEffect, useState, useCallback } from 'react'; +import type { FormEvent } from 'react'; +import { api } from '../api/client'; +import StatusBadge from '../components/StatusBadge'; + +interface IoBrokerSource { + id: number; + name: string; + host: string; + port: number; + enabled: number; + status: string; + created_at: string; +} + +interface IoBrokerState { + id: number; + source_id: number; + state_id: string; + selected: number; + last_value: string | null; + last_ts: number | null; + discovered_at: string; +} + +interface StatesResponse { + states: IoBrokerState[]; + total: number; +} + +const emptyForm = { name: '', host: '', port: '8087' }; +const PAGE_SIZE = 50; + +export default function IoBroker() { + const [sources, setSources] = useState([]); + const [selectedSource, setSelectedSource] = useState(null); + const [form, setForm] = useState(emptyForm); + const [error, setError] = useState(''); + const [testResult, setTestResult] = useState(null); + const [discoveryMsg, setDiscoveryMsg] = useState(''); + + // States panel state + const [statesData, setStatesData] = useState({ states: [], total: 0 }); + const [search, setSearch] = useState(''); + const [searchInput, setSearchInput] = useState(''); + const [showSelected, setShowSelected] = useState(false); + const [page, setPage] = useState(0); + + async function loadSources() { + try { + setSources(await api.get('/api/iobroker/sources')); + } catch (e) { + setError((e as Error).message); + } + } + + const loadStates = useCallback(async () => { + if (!selectedSource) return; + try { + const params = new URLSearchParams(); + if (search) params.set('search', search); + if (showSelected) params.set('selected', '1'); + params.set('limit', String(PAGE_SIZE)); + params.set('offset', String(page * PAGE_SIZE)); + const result = await api.get( + `/api/iobroker/sources/${selectedSource.id}/states?${params}` + ); + setStatesData(result); + } catch (e) { + setError((e as Error).message); + } + }, [selectedSource, search, showSelected, page]); + + useEffect(() => { loadSources(); }, []); + useEffect(() => { loadStates(); }, [loadStates]); + + // Debounced search + useEffect(() => { + const timer = setTimeout(() => { setSearch(searchInput); setPage(0); }, 400); + return () => clearTimeout(timer); + }, [searchInput]); + + async function handleAdd(e: FormEvent) { + e.preventDefault(); + setError(''); + try { + await api.post('/api/iobroker/sources', { ...form, port: parseInt(form.port) }); + setForm(emptyForm); + loadSources(); + } catch (e) { + setError((e as Error).message); + } + } + + async function handleDelete(id: number) { + if (!confirm('Quelle wirklich loeschen?')) return; + try { + await api.del(`/api/iobroker/sources/${id}`); + if (selectedSource?.id === id) { + setSelectedSource(null); + setStatesData({ states: [], total: 0 }); + } + loadSources(); + } catch (e) { + setError((e as Error).message); + } + } + + async function handleTest(source: IoBrokerSource) { + setTestResult(null); + try { + const result = await api.post<{ success: boolean }>( + `/api/iobroker/sources/${source.id}/test` + ); + setTestResult(result.success ? 'Verbindung erfolgreich' : 'Nicht erreichbar'); + } catch (e) { + setTestResult('Fehler: ' + (e as Error).message); + } + } + + async function handleDiscover(source: IoBrokerSource) { + setDiscoveryMsg('Discovery laeuft...'); + try { + await api.post(`/api/iobroker/sources/${source.id}/discover`); + setDiscoveryMsg('Discovery gestartet. States werden geladen...'); + setTimeout(() => { + loadStates(); + setDiscoveryMsg(''); + }, 3000); + } catch (e) { + setDiscoveryMsg('Fehler: ' + (e as Error).message); + } + } + + async function handleSelectState(state: IoBrokerState, selected: boolean) { + try { + await api.put(`/api/iobroker/states/${state.id}`, { selected: selected ? 1 : 0 }); + loadStates(); + } catch (e) { + setError((e as Error).message); + } + } + + function handleSelectSource(source: IoBrokerSource) { + setSelectedSource(source); + setPage(0); + setSearch(''); + setSearchInput(''); + setShowSelected(false); + setDiscoveryMsg(''); + setTestResult(null); + } + + const totalPages = Math.ceil(statesData.total / PAGE_SIZE); + + return ( +
+

ioBroker Quellen

+ {error && ( +
+ {error} + +
+ )} + + {/* Sources table */} +
+ + + + + + + + + + + {sources.length === 0 && ( + + + + )} + {sources.map((s) => ( + handleSelectSource(s)} + > + + + + + + ))} + +
NameHost:PortStatusAktionen
+ Keine ioBroker Quellen konfiguriert +
{s.name}{s.host}:{s.port} + + + +
+
+ + {testResult && ( +
+ {testResult} +
+ )} + + {/* Add form */} +
+

Quelle hinzufuegen

+
+
+ + setForm({ ...form, name: e.target.value })} + /> +
+
+ + setForm({ ...form, host: e.target.value })} + /> +
+
+ + setForm({ ...form, port: e.target.value })} + /> +
+
+ +
+
+
+ + {/* States panel */} + {selectedSource && ( +
+

+ States — {selectedSource.name} +

+

+ {statesData.total} States gefunden +

+ + {discoveryMsg && ( +

{discoveryMsg}

+ )} + + {/* Controls */} +
+ setSearchInput(e.target.value)} + /> + +
+ +
+ + + + + + + + + + {statesData.states.length === 0 && ( + + + + )} + {statesData.states.map((s) => ( + + + + + + ))} + +
AktivState IDLetzter Wert
+ {statesData.total === 0 + ? 'Keine States gefunden. Starte Discovery.' + : 'Keine States fuer diese Suche.'} +
+ handleSelectState(s, e.target.checked)} + className="w-4 h-4 accent-green-500 cursor-pointer" + /> + + {s.state_id} + + {s.last_value ?? '—'} +
+
+ + {totalPages > 1 && ( +
+ + Seite {page + 1} von {totalPages} + +
+ )} +
+ )} +
+ ); +} diff --git a/display/middleware/src/api/index.ts b/display/middleware/src/api/index.ts index 635ca15..98f72f1 100644 --- a/display/middleware/src/api/index.ts +++ b/display/middleware/src/api/index.ts @@ -6,6 +6,7 @@ import displayApiRouter from './routes/display-api'; import mqttBrokersRouter from './routes/mqtt-brokers'; import mqttTopicsRouter from './routes/mqtt-topics'; import pgSourcesRouter from './routes/pg-sources'; +import iobrokerRouter from './routes/iobroker'; import datapointsRouter from './routes/datapoints'; import displaysRouter from './routes/displays'; import layoutsRouter from './routes/layouts'; @@ -30,6 +31,7 @@ export function createApp(): express.Application { app.use('/api/mqtt/brokers/:brokerId/topics', authMiddleware, mqttTopicsRouter); app.use('/api/mqtt/brokers', authMiddleware, mqttBrokersRouter); app.use('/api/pg/sources', authMiddleware, pgSourcesRouter); + app.use('/api/iobroker', authMiddleware, iobrokerRouter); app.use('/api/datapoints', authMiddleware, datapointsRouter); app.use('/api/displays', authMiddleware, displaysRouter); app.use('/api/layouts', authMiddleware, layoutsRouter); diff --git a/display/middleware/src/api/routes/iobroker.ts b/display/middleware/src/api/routes/iobroker.ts new file mode 100644 index 0000000..8c4540c --- /dev/null +++ b/display/middleware/src/api/routes/iobroker.ts @@ -0,0 +1,132 @@ +import { Router } from 'express'; +import { + getAllSources, + getSourceById, + createSource, + updateSource, + deleteSource, + testSource, + discoverStates, + getStatesForSource, + updateState, +} from '../../modules/iobroker/source-manager'; +import { + createDatapointFromIoBroker, + getDatapointByIoBrokerStateId, +} from '../../modules/datapoints/datapoint-manager'; +import { startPolling } from '../../modules/iobroker/state-poller'; +import { getDb } from '../../db/index'; + +const router = Router(); + +// GET /api/iobroker/sources +router.get('/sources', (_req, res): void => { + try { + res.json(getAllSources()); + } catch (err) { + res.status(500).json({ error: (err as Error).message }); + } +}); + +// POST /api/iobroker/sources +router.post('/sources', (req, res): void => { + try { + const { name, host } = req.body; + if (!name || !host) { + res.status(400).json({ error: 'name and host are required' }); + return; + } + res.status(201).json(createSource(req.body)); + } catch (err) { + res.status(500).json({ error: (err as Error).message }); + } +}); + +// PUT /api/iobroker/sources/:id +router.put('/sources/:id', (req, res): void => { + try { + const source = updateSource(parseInt(req.params.id), req.body); + if (!source) { res.status(404).json({ error: 'Source not found' }); return; } + res.json(source); + } catch (err) { + res.status(500).json({ error: (err as Error).message }); + } +}); + +// DELETE /api/iobroker/sources/:id +router.delete('/sources/:id', (req, res): void => { + try { + const deleted = deleteSource(parseInt(req.params.id)); + if (!deleted) { res.status(404).json({ error: 'Source not found' }); return; } + res.status(204).send(); + } catch (err) { + res.status(500).json({ error: (err as Error).message }); + } +}); + +// POST /api/iobroker/sources/:id/test +router.post('/sources/:id/test', (req, res): void => { + const id = parseInt(req.params.id); + const source = getSourceById(id); + if (!source) { res.status(404).json({ error: 'Source not found' }); return; } + + testSource(source.host, source.port) + .then((ok) => res.json({ success: ok })) + .catch((err) => res.status(500).json({ error: (err as Error).message })); +}); + +// POST /api/iobroker/sources/:id/discover +router.post('/sources/:id/discover', (req, res): void => { + const id = parseInt(req.params.id); + + res.json({ message: 'Discovery started', sourceId: id }); + + discoverStates(id).then((count) => { + console.log(`ioBroker discovery for source ${id}: found ${count} states`); + }).catch((err) => { + console.error(`ioBroker discovery error for source ${id}:`, (err as Error).message); + }); +}); + +// GET /api/iobroker/sources/:id/states?search=&selected=1&limit=50&offset=0 +router.get('/sources/:id/states', (req, res): void => { + try { + const sourceId = parseInt(req.params.id); + const search = req.query.search as string | undefined; + const selectedOnly = req.query.selected === '1'; + const limit = parseInt(req.query.limit as string) || 50; + const offset = parseInt(req.query.offset as string) || 0; + res.json(getStatesForSource(sourceId, { search, selectedOnly, limit, offset })); + } catch (err) { + res.status(500).json({ error: (err as Error).message }); + } +}); + +// PUT /api/iobroker/states/:id +router.put('/states/:id', (req, res): void => { + try { + const stateRowId = parseInt(req.params.id); + const stateRow = getDb() + .prepare('SELECT * FROM iobroker_states WHERE id = ?') + .get(stateRowId) as { id: number; source_id: number; state_id: string; selected: number } | undefined; + + if (!stateRow) { res.status(404).json({ error: 'State not found' }); return; } + + const updated = updateState(stateRowId, req.body); + + // If selecting a state, auto-create a datapoint and ensure poller is running + if (req.body.selected === 1 || req.body.selected === true) { + const existing = getDatapointByIoBrokerStateId(stateRow.state_id); + if (!existing) { + createDatapointFromIoBroker(stateRow.source_id, stateRow.state_id, stateRow.state_id); + } + startPolling(stateRow.source_id); + } + + res.json(updated); + } catch (err) { + res.status(500).json({ error: (err as Error).message }); + } +}); + +export default router; diff --git a/display/middleware/src/db/index.ts b/display/middleware/src/db/index.ts index 1c202a1..c673bcb 100644 --- a/display/middleware/src/db/index.ts +++ b/display/middleware/src/db/index.ts @@ -1,5 +1,6 @@ import Database from 'better-sqlite3'; import { migrate001 } from './migrations/001-initial'; +import { migrate002 } from './migrations/002-iobroker'; let dbInstance: Database.Database | null = null; @@ -11,6 +12,7 @@ export function initDb(path: string): Database.Database { // Run migrations migrate001(db); + migrate002(db); dbInstance = db; return db; diff --git a/display/middleware/src/db/migrations/002-iobroker.ts b/display/middleware/src/db/migrations/002-iobroker.ts new file mode 100644 index 0000000..bf1c325 --- /dev/null +++ b/display/middleware/src/db/migrations/002-iobroker.ts @@ -0,0 +1,57 @@ +import Database from 'better-sqlite3'; + +export function migrate002(db: Database.Database): void { + // Create ioBroker sources table + db.exec(` + CREATE TABLE IF NOT EXISTS iobroker_sources ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + host TEXT NOT NULL, + port INTEGER DEFAULT 8087, + enabled INTEGER DEFAULT 1, + status TEXT DEFAULT 'disconnected', + created_at TEXT DEFAULT (datetime('now')) + ); + + CREATE TABLE IF NOT EXISTS iobroker_states ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_id INTEGER NOT NULL REFERENCES iobroker_sources(id) ON DELETE CASCADE, + state_id TEXT NOT NULL, + selected INTEGER DEFAULT 0, + last_value TEXT, + last_ts INTEGER, + discovered_at TEXT DEFAULT (datetime('now')) + ); + `); + + // Update datapoints CHECK constraint to allow 'iobroker' + // SQLite does not support ALTER CHECK, so recreate the table if needed + const row = db + .prepare("SELECT sql FROM sqlite_master WHERE type='table' AND name='datapoints'") + .get() as { sql: string } | undefined; + + if (row && !row.sql.includes('iobroker')) { + db.exec(` + CREATE TABLE datapoints_new ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + label TEXT NOT NULL, + unit TEXT, + value TEXT, + status TEXT DEFAULT 'unknown', + source_type TEXT NOT NULL CHECK(source_type IN ('mqtt','postgresql','iobroker')), + source_id INTEGER NOT NULL, + topic_id INTEGER, + query_id INTEGER, + transform TEXT, + enabled INTEGER DEFAULT 1, + updated_at TEXT DEFAULT (datetime('now')), + FOREIGN KEY (topic_id) REFERENCES mqtt_topics(id) ON DELETE SET NULL, + FOREIGN KEY (query_id) REFERENCES pg_queries(id) ON DELETE SET NULL + ); + INSERT INTO datapoints_new SELECT * FROM datapoints; + DROP TABLE datapoints; + ALTER TABLE datapoints_new RENAME TO datapoints; + `); + } +} diff --git a/display/middleware/src/index.ts b/display/middleware/src/index.ts index b9273f4..edb6fc3 100644 --- a/display/middleware/src/index.ts +++ b/display/middleware/src/index.ts @@ -4,6 +4,7 @@ import { createApp } from './api/index'; import { connectAllEnabledBrokers } from './modules/mqtt/broker-manager'; import { connectAllEnabledSources } from './modules/postgres/source-manager'; import { startAllQueryTimers } from './modules/postgres/query-scheduler'; +import { startAllPollers } from './modules/iobroker/state-poller'; const PORT = process.env.PORT ? parseInt(process.env.PORT) : 4000; const DB_PATH = process.env.DB_PATH ?? './display.db'; @@ -32,6 +33,9 @@ async function main(): Promise { await connectAllEnabledSources(); startAllQueryTimers(); + // Start ioBroker state pollers for all enabled sources + startAllPollers(); + // Create and start Express app const app = createApp(); diff --git a/display/middleware/src/modules/datapoints/datapoint-manager.ts b/display/middleware/src/modules/datapoints/datapoint-manager.ts index 4157b8a..39aa756 100644 --- a/display/middleware/src/modules/datapoints/datapoint-manager.ts +++ b/display/middleware/src/modules/datapoints/datapoint-manager.ts @@ -137,3 +137,24 @@ export function getDatapointByQueryId(queryId: number): Datapoint | undefined { .prepare('SELECT * FROM datapoints WHERE query_id = ?') .get(queryId) as Datapoint | undefined; } + +export function createDatapointFromIoBroker( + sourceId: number, + stateId: string, + stateName: string +): Datapoint { + const db = getDb(); + const result = db.prepare( + `INSERT INTO datapoints (name, label, source_type, source_id, enabled) + VALUES (?, ?, 'iobroker', ?, 1)` + ).run(stateId, stateName, sourceId); + const id = result.lastInsertRowid as number; + return db.prepare('SELECT * FROM datapoints WHERE id = ?').get(id) as Datapoint; +} + +export function getDatapointByIoBrokerStateId(stateId: string): Datapoint | undefined { + const db = getDb(); + return db + .prepare("SELECT * FROM datapoints WHERE source_type = 'iobroker' AND name = ?") + .get(stateId) as Datapoint | undefined; +} diff --git a/display/middleware/src/modules/iobroker/source-manager.ts b/display/middleware/src/modules/iobroker/source-manager.ts new file mode 100644 index 0000000..78a09ca --- /dev/null +++ b/display/middleware/src/modules/iobroker/source-manager.ts @@ -0,0 +1,187 @@ +import { getDb } from '../../db/index'; + +export interface IoBrokerSource { + id: number; + name: string; + host: string; + port: number; + enabled: number; + status: string; + created_at: string; +} + +export interface CreateSourceData { + name: string; + host: string; + port?: number; + enabled?: number; +} + +export type UpdateSourceData = Partial; + +export interface IoBrokerState { + id: number; + source_id: number; + state_id: string; + selected: number; + last_value: string | null; + last_ts: number | null; + discovered_at: string; +} + +export function getAllSources(): IoBrokerSource[] { + const db = getDb(); + return db.prepare('SELECT * FROM iobroker_sources ORDER BY id ASC').all() as IoBrokerSource[]; +} + +export function getSourceById(id: number): IoBrokerSource | undefined { + const db = getDb(); + return db.prepare('SELECT * FROM iobroker_sources WHERE id = ?').get(id) as IoBrokerSource | undefined; +} + +export function createSource(data: CreateSourceData): IoBrokerSource { + const db = getDb(); + const result = db.prepare( + `INSERT INTO iobroker_sources (name, host, port, enabled) + VALUES (?, ?, ?, ?)` + ).run( + data.name, + data.host, + data.port ?? 8087, + data.enabled ?? 1, + ); + const id = result.lastInsertRowid as number; + return db.prepare('SELECT * FROM iobroker_sources WHERE id = ?').get(id) as IoBrokerSource; +} + +export function updateSource(id: number, data: UpdateSourceData): IoBrokerSource | undefined { + const db = getDb(); + const fields: string[] = []; + const values: unknown[] = []; + + if (data.name !== undefined) { fields.push('name = ?'); values.push(data.name); } + if (data.host !== undefined) { fields.push('host = ?'); values.push(data.host); } + if (data.port !== undefined) { fields.push('port = ?'); values.push(data.port); } + if (data.enabled !== undefined) { fields.push('enabled = ?'); values.push(data.enabled); } + + if (fields.length === 0) return getSourceById(id); + + values.push(id); + db.prepare(`UPDATE iobroker_sources SET ${fields.join(', ')} WHERE id = ?`).run(...values); + return getSourceById(id); +} + +export function deleteSource(id: number): boolean { + const db = getDb(); + const result = db.prepare('DELETE FROM iobroker_sources WHERE id = ?').run(id); + return (result.changes as number) > 0; +} + +function setStatus(id: number, status: string): void { + const db = getDb(); + db.prepare('UPDATE iobroker_sources SET status = ? WHERE id = ?').run(status, id); +} + +export async function testSource(host: string, port: number): Promise { + try { + const url = `http://${host}:${port}/get/system.adapter.admin.0.alive`; + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 5000); + const res = await fetch(url, { signal: controller.signal }); + clearTimeout(timeout); + return res.ok; + } catch { + return false; + } +} + +export function getStatesForSource( + sourceId: number, + opts?: { search?: string; selectedOnly?: boolean; limit?: number; offset?: number } +): { states: IoBrokerState[]; total: number } { + const db = getDb(); + const conditions = ['source_id = ?']; + const params: unknown[] = [sourceId]; + + if (opts?.search) { + conditions.push('state_id LIKE ?'); + params.push(`%${opts.search}%`); + } + if (opts?.selectedOnly) { + conditions.push('selected = 1'); + } + + const where = conditions.join(' AND '); + const total = (db.prepare(`SELECT count(*) as c FROM iobroker_states WHERE ${where}`).get(...params) as { c: number }).c; + + const limit = opts?.limit ?? 50; + const offset = opts?.offset ?? 0; + params.push(limit, offset); + + const states = db + .prepare(`SELECT * FROM iobroker_states WHERE ${where} ORDER BY selected DESC, state_id ASC LIMIT ? OFFSET ?`) + .all(...params) as IoBrokerState[]; + + return { states, total }; +} + +export function updateState(id: number, data: { selected?: number; last_value?: string; last_ts?: number }): IoBrokerState | undefined { + const db = getDb(); + const fields: string[] = []; + const values: unknown[] = []; + + if (data.selected !== undefined) { fields.push('selected = ?'); values.push(data.selected); } + if (data.last_value !== undefined) { fields.push('last_value = ?'); values.push(data.last_value); } + if (data.last_ts !== undefined) { fields.push('last_ts = ?'); values.push(data.last_ts); } + + if (fields.length === 0) { + return db.prepare('SELECT * FROM iobroker_states WHERE id = ?').get(id) as IoBrokerState | undefined; + } + + values.push(id); + db.prepare(`UPDATE iobroker_states SET ${fields.join(', ')} WHERE id = ?`).run(...values); + return db.prepare('SELECT * FROM iobroker_states WHERE id = ?').get(id) as IoBrokerState | undefined; +} + +export async function discoverStates(sourceId: number): Promise { + const source = getSourceById(sourceId); + if (!source) throw new Error(`ioBroker source ${sourceId} not found`); + + const db = getDb(); + + try { + setStatus(sourceId, 'discovering'); + + const url = `http://${source.host}:${source.port}/objects?type=state&pattern=*`; + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 15000); + const res = await fetch(url, { signal: controller.signal }); + clearTimeout(timeout); + + if (!res.ok) { + setStatus(sourceId, 'error'); + throw new Error(`ioBroker API returned ${res.status}`); + } + + const data = await res.json() as Record; + const stateIds = Object.keys(data).slice(0, 5000); + + const insertStmt = db.prepare( + `INSERT OR IGNORE INTO iobroker_states (source_id, state_id) VALUES (?, ?)` + ); + + const insertMany = db.transaction((ids: string[]) => { + for (const sid of ids) { + insertStmt.run(sourceId, sid); + } + }); + + insertMany(stateIds); + setStatus(sourceId, 'connected'); + + return stateIds.length; + } catch (err) { + setStatus(sourceId, 'error'); + throw err; + } +} diff --git a/display/middleware/src/modules/iobroker/state-poller.ts b/display/middleware/src/modules/iobroker/state-poller.ts new file mode 100644 index 0000000..6477365 --- /dev/null +++ b/display/middleware/src/modules/iobroker/state-poller.ts @@ -0,0 +1,86 @@ +import { getDb } from '../../db/index'; +import { getAllSources, getSourceById, type IoBrokerState } from './source-manager'; +import { + getDatapointByIoBrokerStateId, + updateDatapointValue, +} from '../datapoints/datapoint-manager'; + +const pollerTimers = new Map>(); + +const POLL_INTERVAL_MS = 10_000; + +async function pollSource(sourceId: number): Promise { + const source = getSourceById(sourceId); + if (!source || !source.enabled) return; + + const db = getDb(); + const selectedStates = db + .prepare('SELECT * FROM iobroker_states WHERE source_id = ? AND selected = 1') + .all(sourceId) as IoBrokerState[]; + + if (selectedStates.length === 0) return; + + const stateIds = selectedStates.map((s) => s.state_id).join(','); + const url = `http://${source.host}:${source.port}/getBulk/${encodeURIComponent(stateIds)}`; + + try { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 8000); + const res = await fetch(url, { signal: controller.signal }); + clearTimeout(timeout); + + if (!res.ok) return; + + const results = await res.json() as Array<{ id: string; val: unknown; ts?: number }>; + + for (const item of results) { + if (item.val === null || item.val === undefined) continue; + const rawValue = String(item.val); + + // Update last_value in iobroker_states + db.prepare( + 'UPDATE iobroker_states SET last_value = ?, last_ts = ? WHERE source_id = ? AND state_id = ?' + ).run(rawValue, item.ts ?? null, sourceId, item.id); + + // Update datapoint value if one exists + const dp = getDatapointByIoBrokerStateId(item.id); + if (dp) { + updateDatapointValue(dp.id, rawValue); + } + } + } catch (err) { + // Silently ignore polling errors — source may be temporarily unreachable + console.warn(`ioBroker poll error for source ${sourceId}:`, (err as Error).message); + } +} + +export function startPolling(sourceId: number): void { + if (pollerTimers.has(sourceId)) return; // already running + + const timer = setInterval(() => { + pollSource(sourceId).catch((err) => + console.error(`ioBroker poller ${sourceId}:`, (err as Error).message) + ); + }, POLL_INTERVAL_MS); + + pollerTimers.set(sourceId, timer); + console.log(`ioBroker poller started for source ${sourceId}`); +} + +export function stopPolling(sourceId: number): void { + const timer = pollerTimers.get(sourceId); + if (timer) { + clearInterval(timer); + pollerTimers.delete(sourceId); + console.log(`ioBroker poller stopped for source ${sourceId}`); + } +} + +export function startAllPollers(): void { + const sources = getAllSources(); + for (const source of sources) { + if (source.enabled) { + startPolling(source.id); + } + } +}