diff --git a/CloudServer/server.js b/CloudServer/server.js index ef35636..b11cea4 100644 --- a/CloudServer/server.js +++ b/CloudServer/server.js @@ -211,7 +211,11 @@ app.use(express.static(path.join(__dirname, 'public'))); app.set('view engine', 'ejs'); // Data Store (In-memory) +// stations[name] = { name, code, platform, audioCableGroup, version, lastSeen, upcomingTrains, upcomingCount, currentTrain, nodes: [] } +// nodes[]: { ws, hasStationPeripheral, stationPeripheralSide, hasBlockReader, hasCreateTarget, identityMode, nodeId, wsId, connectedAt } let stations = {}; +let audioCableGroups = {}; // audioCableGroups[id] = { id, name, color, description, stationNames: [] } +const audioCableTokens = new Map(); // cableGroupId -> { holderStationName, holderAudioId, holderPriority, acquiredAt, cancelToken } let activeTrains = []; let stationLogs = []; let audioVersion = String(Date.now()); @@ -230,6 +234,363 @@ function delay(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); } +// ===== A1 Core Helpers ===== +function generateNodeId() { + return 'n_' + Date.now().toString(36) + '_' + Math.random().toString(36).slice(2, 8); +} + +function generateWsId(ws) { + if (!ws) return ''; + if (ws.__wsId) return ws.__wsId; + ws.__wsId = 'ws_' + Date.now().toString(36) + '_' + Math.random().toString(36).slice(2, 8); + return ws.__wsId; +} + +function getGroupKeyForStation(stationOrName) { + if (!stationOrName) return null; + if (typeof stationOrName === 'string') { + const clean = String(stationOrName || '').trim(); + return clean || null; + } + const name = String(stationOrName?.name || '').trim(); + return name || null; +} + +function getStationGroupMembers(stationName) { + const key = getGroupKeyForStation(stationName); + if (!key) return []; + const st = stations[key]; + if (!st || !Array.isArray(st.nodes)) return []; + return st.nodes.slice(); +} + +function getStationMaster(stationName) { + const key = getGroupKeyForStation(stationName); + if (!key) return { master: null, ws: null }; + const st = stations[key]; + if (!st || !Array.isArray(st.nodes) || st.nodes.length === 0) return { master: null, ws: null }; + + // First pass: OPEN ws + hasStationPeripheral priority, weighted by lastSeen + let bestScore = -Infinity; + let bestNode = null; + const now = Date.now(); + for (const node of st.nodes) { + if (!node.ws || node.ws.readyState !== WebSocket.OPEN) continue; + let score = (Number(node.lastSeen || st.lastSeen || now) % 1000000) / 1000000.0; + if (node.hasStationPeripheral) score += 1000; + if (score > bestScore) { + bestScore = score; + bestNode = node; + } + } + if (bestNode) return { master: bestNode, ws: bestNode.ws }; + + // Second pass: any OPEN ws (even without station peripheral) + for (const node of st.nodes) { + if (node.ws && node.ws.readyState === WebSocket.OPEN) { + return { master: node, ws: node.ws }; + } + } + return { master: null, ws: null }; +} + +function getAllStationConnections(stationName) { + const out = []; + for (const node of getStationGroupMembers(stationName)) { + if (node.ws && node.ws.readyState === WebSocket.OPEN) out.push(node.ws); + } + return out; +} + +function ensureStationRecord(name) { + const clean = String(name || '').trim(); + if (!clean) return null; + if (!stations[clean]) { + stations[clean] = { + name: clean, + code: '', + platform: getPlatformNumberFromName(clean), + audioCableGroup: '', + version: '', + lastSeen: Date.now(), + nodes: [] + }; + normalizeStationRecord(stations[clean]); + } + if (!Array.isArray(stations[clean].nodes)) { + stations[clean].nodes = []; + } + return stations[clean]; +} + +function upsertStationNode(stationName, ws, data) { + const clean = String(stationName || '').trim(); + if (!clean) return null; + const st = ensureStationRecord(clean); + if (!st) return null; + + const wsId = generateWsId(ws); + const now = Date.now(); + + // Update station-level fields + st.lastSeen = now; + if (data?.stationCode) st.code = String(data.stationCode); + if (data?.stationPlatform) st.platform = String(data.stationPlatform); + if (data?.stationVersion) st.version = String(data.stationVersion); + + // Sync cable group from station perspective (if not already set) + if (data?.audioCableGroup && !st.audioCableGroup) { + st.audioCableGroup = String(data.audioCableGroup); + const grp = audioCableGroups[st.audioCableGroup]; + if (grp && Array.isArray(grp.stationNames) && !grp.stationNames.includes(clean)) { + grp.stationNames.push(clean); + } + } + + // Create node-level fields + const nodeFields = { + hasStationPeripheral: false, + stationPeripheralSide: '', + hasBlockReader: false, + hasCreateTarget: false, + identityMode: '', + upcomingTrains: [], + upcomingCount: 0, + currentTrain: null, + tickRate: 20, + syncedAt: 0 + }; + if (data?.hasStationPeripheral !== undefined) nodeFields.hasStationPeripheral = Boolean(data.hasStationPeripheral); + if (data?.stationPeripheralSide !== undefined) nodeFields.stationPeripheralSide = String(data.stationPeripheralSide || ''); + if (data?.hasBlockReader !== undefined) nodeFields.hasBlockReader = Boolean(data.hasBlockReader); + if (data?.hasCreateTarget !== undefined) nodeFields.hasCreateTarget = Boolean(data.hasCreateTarget); + if (data?.identityMode !== undefined) nodeFields.identityMode = String(data.identityMode || ''); + if (data?.upcomingTrains !== undefined) nodeFields.upcomingTrains = Array.isArray(data.upcomingTrains) ? data.upcomingTrains : []; + if (data?.upcomingCount !== undefined) nodeFields.upcomingCount = Number(data.upcomingCount) || 0; + if (data?.currentTrain !== undefined) nodeFields.currentTrain = data.currentTrain || null; + if (data?.tickRate !== undefined) nodeFields.tickRate = Number(data.tickRate) || 20; + if (data?.syncedAt !== undefined) nodeFields.syncedAt = Number(data.syncedAt) || now; + nodeFields.lastSeen = now; + + // Upsert by wsId + let node = st.nodes.find(n => n.wsId === wsId); + if (!node) { + // Also try to find by ws object (in case wsId was reset) + node = st.nodes.find(n => n.ws === ws); + } + if (node) { + Object.assign(node, nodeFields); + node.ws = ws; + node.connectedAt = node.connectedAt || now; + node.lastSeen = now; + } else { + node = { + ...nodeFields, + ws, + wsId, + nodeId: generateNodeId(), + connectedAt: now, + lastSeen: now + }; + st.nodes.push(node); + } + + // Aggregate station-level schedule from the latest synced node + let bestSynced = 0; + let bestSyncedNode = null; + for (const n of st.nodes) { + if (n.syncedAt > bestSynced) { + bestSynced = n.syncedAt; + bestSyncedNode = n; + } + } + if (bestSyncedNode) { + st.upcomingTrains = bestSyncedNode.upcomingTrains || []; + st.upcomingCount = bestSyncedNode.upcomingCount || 0; + st.currentTrain = bestSyncedNode.currentTrain || null; + } + + normalizeStationRecord(st); + return { st, node }; +} + +function removeStationWs(ws) { + const wsId = generateWsId(ws); + const affected = []; + for (const name of Object.keys(stations)) { + const st = stations[name]; + if (!st || !Array.isArray(st.nodes)) continue; + const before = st.nodes.length; + st.nodes = st.nodes.filter(n => n.ws !== ws && n.wsId !== wsId); + if (before !== st.nodes.length) { + affected.push(name); + // Release any cable tokens held by this station + for (const [cgId, tok] of Array.from(audioCableTokens.entries())) { + if (tok && tok.holderStationName === name) { + if (tok.cancelToken) tok.cancelToken.cancelled = true; + audioCableTokens.delete(cgId); + } + } + } + } + // Notify affected groups (send station_group_state to remaining members) + for (const name of affected) { + sendStationGroupState(name); + } + return affected; +} + +function getCableGroupByStationName(stationName) { + const clean = String(stationName || '').trim(); + if (!clean) return null; + const st = stations[clean]; + if (st && st.audioCableGroup) { + const grp = audioCableGroups[st.audioCableGroup]; + if (grp) return grp; + } + for (const id of Object.keys(audioCableGroups)) { + const grp = audioCableGroups[id]; + if (Array.isArray(grp?.stationNames) && grp.stationNames.includes(clean)) { + if (st) st.audioCableGroup = id; + return grp; + } + } + return null; +} + +function sendWsToStationMaster(stationName, msgObj) { + const { ws } = getStationMaster(stationName); + if (!ws || ws.readyState !== WebSocket.OPEN) return false; + try { ws.send(JSON.stringify(msgObj)); return true; } catch { return false; } +} + +function sendStationGroupState(stationName) { + const clean = String(stationName || '').trim(); + if (!clean) return; + const groupKey = getGroupKeyForStation(clean); + const members = getStationGroupMembers(clean); + const { master } = getStationMaster(clean); + const cableGrp = getCableGroupByStationName(clean); + const msg = { + type: 'station_group_state', + groupKey, + memberCount: members.length, + masterNodeId: master?.nodeId || '', + cableGroupId: cableGrp?.id || '' + }; + for (const node of members) { + if (node.ws && node.ws.readyState === WebSocket.OPEN) { + try { node.ws.send(JSON.stringify(msg)); } catch {} + } + } +} + +async function acquireCableToken(cableGroupId, stationName, audioId, priority, cancelToken) { + const cgId = String(cableGroupId || '').trim(); + const stName = String(stationName || '').trim(); + const audId = String(audioId || ''); + const prio = Number(priority || 0); + + if (!cgId) return { ok: true, reason: 'no_cable_group' }; + if (!stName) return { ok: false, reason: 'no_station' }; + + const existing = audioCableTokens.get(cgId); + if (!existing) { + audioCableTokens.set(cgId, { + holderStationName: stName, + holderAudioId: audId, + holderPriority: prio, + acquiredAt: Date.now(), + cancelToken: cancelToken || null + }); + sendWsToStationMaster(stName, { + type: 'cable_token_granted', + cableGroupId: cgId, + stationName: stName, + audioId: audId + }); + return { ok: true }; + } + + // Same holder and audio - already granted + if (existing.holderStationName === stName && existing.holderAudioId === audId) { + return { ok: true, reason: 'already_holder' }; + } + + // Same holder but different audioId: release old (ignore if in-flight cancel already) + if (existing.holderStationName === stName) { + audioCableTokens.set(cgId, { + holderStationName: stName, + holderAudioId: audId, + holderPriority: prio, + acquiredAt: Date.now(), + cancelToken: cancelToken || null + }); + return { ok: true, reason: 'same_station_updated' }; + } + + // Higher priority can preempt lower + if (prio > existing.holderPriority) { + sendWsToStationMaster(existing.holderStationName, { + type: 'audio_preempt', + reason: 'higher_priority', + byStationName: stName, + priority: prio, + audioId: audId + }); + // Send denied to previous holder + sendWsToStationMaster(existing.holderStationName, { + type: 'cable_token_denied', + cableGroupId: cgId, + stationName: existing.holderStationName, + reason: 'preempted', + holderStationName: stName, + holderPriority: prio + }); + if (existing.cancelToken) existing.cancelToken.cancelled = true; + await delay(200); + audioCableTokens.set(cgId, { + holderStationName: stName, + holderAudioId: audId, + holderPriority: prio, + acquiredAt: Date.now(), + cancelToken: cancelToken || null + }); + sendWsToStationMaster(stName, { + type: 'cable_token_granted', + cableGroupId: cgId, + stationName: stName, + audioId: audId + }); + return { ok: true, reason: 'preempted_lower' }; + } + + // Lower or equal priority: deny + sendWsToStationMaster(stName, { + type: 'cable_token_denied', + cableGroupId: cgId, + stationName: stName, + reason: 'lower_priority', + holderStationName: existing.holderStationName, + holderPriority: existing.holderPriority + }); + return { ok: false, reason: 'lower_priority', holderStationName: existing.holderStationName, holderPriority: existing.holderPriority }; +} + +function releaseCableToken(cableGroupId, stationName, audioId) { + const cgId = String(cableGroupId || '').trim(); + const stName = String(stationName || '').trim(); + const audId = String(audioId || ''); + if (!cgId) return false; + const existing = audioCableTokens.get(cgId); + if (!existing) return false; + if (existing.holderStationName !== stName) return false; + if (audId && existing.holderAudioId && existing.holderAudioId !== audId) return false; + if (existing.cancelToken) existing.cancelToken.cancelled = true; + audioCableTokens.delete(cgId); + return true; +} + function getLocalStationScriptVersion() { try { if (!fs.existsSync(stationScriptPath)) return ''; @@ -331,11 +692,12 @@ async function publishStationUpdate(req) { let onlineStations = 0; let notifiedStations = 0; - for (const [name, station] of Object.entries(stations)) { - const ws = station?.ws; + for (const [name] of Object.entries(stations)) { + const { ws } = getStationMaster(name); + const st = stations[name]; if (!ws || ws.readyState !== WebSocket.OPEN) continue; onlineStations++; - if (maybeSendStationUpdate(ws, name, station.version)) { + if (maybeSendStationUpdate(ws, name, st?.version)) { notifiedStations++; } } @@ -400,22 +762,20 @@ function getStationData(stationName) { } function getStationWs(stationName, fallbackWs) { - const st = getStationData(stationName); - if (st && st.ws && st.ws.readyState === WebSocket.OPEN) return st.ws; + const { ws } = getStationMaster(stationName); + if (ws && ws.readyState === WebSocket.OPEN) return ws; if (fallbackWs && fallbackWs.readyState === WebSocket.OPEN) return fallbackWs; return null; } function getStationGroupKey(station) { if (!station) return null; - const code = normalizeStationCode(station.code || getStationCodeFromName(station.name)); - const platform = normalizeStationPlatform(station.platform || getPlatformNumberFromName(station.name)); - if (code) return `${code}@@${platform}`; - return `name@@${String(station.name || '').trim()}`; + const name = String(station.name || '').trim(); + return name || null; } function getStationGroupTarget(station) { - const groupKey = getStationGroupKey(station); + const groupKey = getGroupKeyForStation(station); return groupKey ? `group:${groupKey}` : null; } @@ -425,111 +785,101 @@ function getStationGroups() { for (const k in stations) { if (!Object.prototype.hasOwnProperty.call(stations, k)) continue; const st = stations[k]; + const key = getGroupKeyForStation(st); + if (!key) continue; + if (groups.has(key)) continue; // already processed by exact name + const code = normalizeStationCode(st.code || getStationCodeFromName(st.name)) || ''; const platform = normalizeStationPlatform(st.platform || getPlatformNumberFromName(st.name)); - const key = getStationGroupKey(st); - if (!key) continue; + const cableGrp = getCableGroupByStationName(st.name); + const { master } = getStationMaster(st.name); + const nodes = Array.isArray(st.nodes) ? st.nodes : []; - const memberTags = []; - if (st.hasStationPeripheral) memberTags.push('STATION'); - if (st.hasBlockReader) memberTags.push('BLOCKREADER'); - if (st.hasCreateTarget) memberTags.push('TARGET'); - const isHost = Boolean(st.hasStationPeripheral); + let hostCount = 0; + let slaveCount = 0; + let hasAnyStationPeripheral = false; + let hasBlockReader = false; + let hasCreateTarget = false; + let lastSeen = st.lastSeen || 0; - const memberInfo = { + const membersArr = nodes.map((node) => { + const isOnline = !!(node.ws && node.ws.readyState === WebSocket.OPEN); + const isHost = Boolean(node.hasStationPeripheral); + if (isHost) { + hostCount++; + hasAnyStationPeripheral = true; + } else { + slaveCount++; + } + if (node.hasBlockReader) hasBlockReader = true; + if (node.hasCreateTarget) hasCreateTarget = true; + const nodeLastSeen = Number(node.lastSeen || 0); + if (nodeLastSeen > lastSeen) lastSeen = nodeLastSeen; + const tags = []; + if (node.hasStationPeripheral) tags.push('STATION'); + if (node.hasBlockReader) tags.push('BLOCKREADER'); + if (node.hasCreateTarget) tags.push('TARGET'); + return { + nodeId: node.nodeId || '', + wsId: node.wsId || '', + online: isOnline, + isMaster: master && master.nodeId === node.nodeId, + connectedAt: Number(node.connectedAt) || 0, + lastSeen: nodeLastSeen, + hasStationPeripheral: isHost, + stationPeripheralSide: node.stationPeripheralSide || '', + hasBlockReader: Boolean(node.hasBlockReader), + hasCreateTarget: Boolean(node.hasCreateTarget), + identityMode: node.identityMode || '', + tickRate: Number(node.tickRate) || 20, + memberTags: tags + }; + }); + + const memberNames = nodes.map(n => n.nodeId || ''); + const mergedTags = ['CC']; + if (hostCount > 0) mergedTags.push(`HOST:${hostCount}`); + if (slaveCount > 0) mergedTags.push(`SLAVE:${slaveCount}`); + if (nodes.length > 1) mergedTags.push(`GROUP:${nodes.length}`); + if (hasAnyStationPeripheral) mergedTags.push('HAS_STATION'); + if (hasBlockReader) mergedTags.push('HAS_BLOCKREADER'); + if (hasCreateTarget) mergedTags.push('HAS_TARGET'); + + groups.set(key, { + key, + target: `group:${key}`, name: st.name, code, - platform: st.platform || platform, + platform, + lastSeen, version: st.version || '', - lastSeen: st.lastSeen || 0, - hasStationPeripheral: isHost, - stationPeripheralSide: st.stationPeripheralSide || '', - hasBlockReader: Boolean(st.hasBlockReader), - hasCreateTarget: Boolean(st.hasCreateTarget), - identityMode: st.identityMode || '', + source: 'cc', + tags: mergedTags, + memberNames, + memberCount: nodes.length, + primaryName: st.name, + hasStationPeripheral: Boolean(master?.hasStationPeripheral), + hasBlockReader, + hasCreateTarget, + hasAnyStationPeripheral, + hostCount, + slaveCount, upcomingCount: Number(st.upcomingCount) || 0, - memberTags - }; - - if (!groups.has(key)) { - const mergedTags = ['CC']; - if (isHost) mergedTags.push('HOST'); - else mergedTags.push('SLAVE'); - groups.set(key, { - key, - target: getStationGroupTarget(st), - name: st.name, - code, - platform, - lastSeen: st.lastSeen || 0, - version: st.version || '', - source: 'cc', - tags: mergedTags, - memberNames: [st.name], - memberCount: 1, - primaryName: st.name, - hasStationPeripheral: isHost, - hasBlockReader: Boolean(st.hasBlockReader), - hasCreateTarget: Boolean(st.hasCreateTarget), - hasAnyStationPeripheral: isHost, - hostCount: isHost ? 1 : 0, - slaveCount: isHost ? 0 : 1, - upcomingCount: Number(st.upcomingCount) || 0, - upcomingTrains: Array.isArray(st.upcomingTrains) ? st.upcomingTrains.slice() : [], - currentTrain: st.currentTrain || null, - members: [memberInfo] - }); - continue; - } - - const group = groups.get(key); - group.memberNames.push(st.name); - group.members.push(memberInfo); - group.memberCount = group.memberNames.length; - if (isHost) group.hostCount = (group.hostCount || 0) + 1; - else group.slaveCount = (group.slaveCount || 0) + 1; - if (!group.hasAnyStationPeripheral && isHost) group.hasAnyStationPeripheral = true; - if (!group.hasBlockReader && st.hasBlockReader) group.hasBlockReader = true; - if (!group.hasCreateTarget && st.hasCreateTarget) group.hasCreateTarget = true; - if (!group.hasStationPeripheral && isHost) { - group.hasStationPeripheral = true; - group.primaryName = st.name; - group.name = st.name; - } - if ((st.lastSeen || 0) > (group.lastSeen || 0)) { - group.lastSeen = st.lastSeen || 0; - group.version = st.version || group.version || ''; - } else if (!group.version && st.version) { - group.version = st.version; - } - const existingKeys = new Set((group.upcomingTrains || []).map(t => - String(t.trainName || '') + '@@' + String(t.dest || '') + '@@' + Number(t.platform || 0) - )); - for (const t of (st.upcomingTrains || [])) { - const tk = String(t.trainName || '') + '@@' + String(t.dest || '') + '@@' + Number(t.platform || 0); - if (!existingKeys.has(tk)) { - if (!group.upcomingTrains) group.upcomingTrains = []; - group.upcomingTrains.push(t); - existingKeys.add(tk); - } - } - group.upcomingCount = (group.upcomingTrains || []).length; - if (!group.currentTrain && st.currentTrain) group.currentTrain = st.currentTrain; + upcomingTrains: Array.isArray(st.upcomingTrains) ? st.upcomingTrains.slice() : [], + currentTrain: st.currentTrain || null, + members: membersArr, + nodes: membersArr, + masterNodeId: master?.nodeId || '', + audioCableGroup: cableGrp ? { + id: cableGrp.id, + name: cableGrp.name, + color: cableGrp.color || '' + } : null, + cableGroupId: cableGrp?.id || '' + }); } - return Array.from(groups.values()).map(group => { - const mergedTags = ['CC']; - if ((group.hostCount || 0) > 0) mergedTags.push(`HOST:${group.hostCount}`); - if ((group.slaveCount || 0) > 0) mergedTags.push(`SLAVE:${group.slaveCount}`); - if (group.memberCount > 1) mergedTags.push(`GROUP:${group.memberCount}`); - if (group.hasAnyStationPeripheral) mergedTags.push('HAS_STATION'); - if (group.hasBlockReader) mergedTags.push('HAS_BLOCKREADER'); - if (group.hasCreateTarget) mergedTags.push('HAS_TARGET'); - return { - ...group, - tags: mergedTags - }; - }); + return Array.from(groups.values()); } function resolveStationGroup(selector) { @@ -538,19 +888,24 @@ function resolveStationGroup(selector) { if (Object.prototype.hasOwnProperty.call(stations, clean)) { const st = stations[clean]; + const nodes = Array.isArray(st.nodes) ? st.nodes : []; + const code = normalizeStationCode(st.code || getStationCodeFromName(st.name)) || ''; + const platform = normalizeStationPlatform(st.platform || getPlatformNumberFromName(st.name)); + const { master } = getStationMaster(clean); return { - target: getStationGroupTarget(st) || clean, + target: `group:${clean}`, name: st.name, - code: normalizeStationCode(st.code || getStationCodeFromName(st.name)) || '', - platform: normalizeStationPlatform(st.platform || getPlatformNumberFromName(st.name)), - memberNames: [st.name], - memberCount: 1, - primaryName: st.name + code, + platform, + memberNames: nodes.map(n => n.nodeId || ''), + memberCount: nodes.length, + primaryName: st.name, + masterNodeId: master?.nodeId || '' }; } for (const group of getStationGroups()) { - if (group.target === clean || group.name === clean) { + if (group.target === clean || group.name === clean || group.key === clean) { return group; } } @@ -579,21 +934,37 @@ function getMergedStationList() { } function sendAudioControl(stationName, msg, fallbackWs) { - const ws = getStationWs(stationName, fallbackWs); - if (!ws) return false; - try { - ws.send(JSON.stringify(msg)); - return true; - } catch { + const allWs = getAllStationConnections(stationName); + if (allWs.length === 0) { + if (fallbackWs && fallbackWs.readyState === WebSocket.OPEN) { + try { fallbackWs.send(JSON.stringify(msg)); return true; } catch { return false; } + } return false; } + let ok = false; + const payload = JSON.stringify(msg); + for (const ws of allWs) { + try { ws.send(payload); ok = true; } catch {} + } + return ok; } async function sendAudioBufferToStation(stationName, fallbackWs, task) { - const targetWs = getStationWs(stationName, fallbackWs); - if (!targetWs) throw new Error(`Station "${stationName}" not connected`); + const { ws: targetWs } = getStationMaster(stationName); + const masterWs = targetWs || (fallbackWs && fallbackWs.readyState === WebSocket.OPEN ? fallbackWs : null); + if (!masterWs) throw new Error(`Station "${stationName}" not connected`); + const cableGrp = getCableGroupByStationName(stationName); const audioId = String(task.audioId || `${Date.now()}-${Math.random().toString(16).slice(2)}`); + const prio = Number(task.priority || 0); + + if (cableGrp) { + const res = await acquireCableToken(cableGrp.id, stationName, audioId, prio, task.cancelToken || null); + if (!res.ok) { + throw new Error(`Cable token denied: ${res.reason || 'unknown'} (holder=${res.holderStationName || 'none'} p=${res.holderPriority ?? 0})`); + } + } + const buf = task.buffer || null; const durationSec = Number(task.durationSec || 0); const audioFormat = String(task.audioFormat || 'dfpwm').toLowerCase(); @@ -609,45 +980,74 @@ async function sendAudioBufferToStation(stationName, fallbackWs, task) { audioFormat, streamUrl, loop: !!task.loop, - priority: Number(task.priority || 0), - interruptMode: task.interruptMode || getInterruptModeForPriority(Number(task.priority || 0)), - source: task.source || 'unknown' + priority: prio, + interruptMode: task.interruptMode || getInterruptModeForPriority(prio), + source: task.source || 'unknown', + cableGroupId: cableGrp?.id || '' }; - targetWs.send(JSON.stringify(startMsg)); + masterWs.send(JSON.stringify(startMsg)); if (streamUrl) { - return { audioId, cancelled: false, durationSec }; + if (cableGrp && durationSec > 0) { + setTimeout(() => { + try { releaseCableToken(cableGrp.id, stationName, audioId); } catch {} + }, Math.max(100, Math.floor(durationSec * 1000) + 500)); + } + return { audioId, cancelled: false, durationSec, cableGroupId: cableGrp?.id || '' }; } const token = task.cancelToken; const SINGLE_SEND_LIMIT = 32 * 1024; const CHUNK_SIZE = 16 * 1024; - if (buf.length <= SINGLE_SEND_LIMIT) { - if (token && token.cancelled) return { audioId, cancelled: true, durationSec }; - targetWs.send(buf); - } else { - for (let offset = 0; offset < buf.length; offset += CHUNK_SIZE) { - if (token && token.cancelled) return { audioId, cancelled: true, durationSec }; - const end = Math.min(offset + CHUNK_SIZE, buf.length); - targetWs.send(buf.slice(offset, end)); - // For streaming, we can send slightly faster initially then steady - const isInitial = offset < CHUNK_SIZE * 4; - await delay(isInitial ? 20 : 60); + try { + if (buf.length <= SINGLE_SEND_LIMIT) { + if (token && token.cancelled) { + if (cableGrp) releaseCableToken(cableGrp.id, stationName, audioId); + return { audioId, cancelled: true, durationSec, cableGroupId: cableGrp?.id || '' }; + } + masterWs.send(buf); + } else { + for (let offset = 0; offset < buf.length; offset += CHUNK_SIZE) { + if (token && token.cancelled) { + if (cableGrp) releaseCableToken(cableGrp.id, stationName, audioId); + return { audioId, cancelled: true, durationSec, cableGroupId: cableGrp?.id || '' }; + } + const end = Math.min(offset + CHUNK_SIZE, buf.length); + masterWs.send(buf.slice(offset, end)); + const isInitial = offset < CHUNK_SIZE * 4; + await delay(isInitial ? 20 : 60); + } } + + if (token && token.cancelled) { + if (cableGrp) releaseCableToken(cableGrp.id, stationName, audioId); + return { audioId, cancelled: true, durationSec, cableGroupId: cableGrp?.id || '' }; + } + + masterWs.send(JSON.stringify({ + type: 'audio_end', + stationName, + audioId, + durationSec + })); + + if (cableGrp && durationSec > 0) { + setTimeout(() => { + try { releaseCableToken(cableGrp.id, stationName, audioId); } catch {} + }, Math.max(100, Math.floor(durationSec * 1000) + 500)); + } else if (cableGrp) { + releaseCableToken(cableGrp.id, stationName, audioId); + } + + return { audioId, cancelled: false, durationSec, cableGroupId: cableGrp?.id || '' }; + } catch (e) { + if (cableGrp) { + try { releaseCableToken(cableGrp.id, stationName, audioId); } catch {} + } + throw e; } - - if (token && token.cancelled) return { audioId, cancelled: true, durationSec }; - - targetWs.send(JSON.stringify({ - type: 'audio_end', - stationName, - audioId, - durationSec - })); - - return { audioId, cancelled: false, durationSec }; } function buildWavChunkFromPcm(buffer, sampleRate = 48000, channels = 1, bitsPerSample = 16) { @@ -674,15 +1074,32 @@ function buildWavChunkFromPcm(buffer, sampleRate = 48000, channels = 1, bitsPerS } async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { - const targetWs = getStationWs(stationName, fallbackWs); - if (!targetWs) throw new Error(`Station "${stationName}" not connected`); + const { ws: targetWs } = getStationMaster(stationName); + const masterWs = targetWs || (fallbackWs && fallbackWs.readyState === WebSocket.OPEN ? fallbackWs : null); + if (!masterWs) throw new Error(`Station "${stationName}" not connected`); + + const cableGrp = getCableGroupByStationName(stationName); + const audioId = String(task.audioId || `${Date.now()}-${Math.random().toString(16).slice(2)}`); + const prio = Number(task.priority || 0); + const durationSec = Number(task.durationSec || 0); + + if (cableGrp) { + const res = await acquireCableToken(cableGrp.id, stationName, audioId, prio, task.cancelToken || null); + if (!res.ok) { + throw new Error(`Cable token denied: ${res.reason || 'unknown'} (holder=${res.holderStationName || 'none'} p=${res.holderPriority ?? 0})`); + } + } const sourcePath = String(task.sourcePath || '').trim(); - if (!sourcePath) throw new Error('Live WAV sourcePath required'); - if (!fs.existsSync(sourcePath)) throw new Error(`Live WAV source missing: ${sourcePath}`); + if (!sourcePath) { + if (cableGrp) releaseCableToken(cableGrp.id, stationName, audioId); + throw new Error('Live WAV sourcePath required'); + } + if (!fs.existsSync(sourcePath)) { + if (cableGrp) releaseCableToken(cableGrp.id, stationName, audioId); + throw new Error(`Live WAV source missing: ${sourcePath}`); + } - const audioId = String(task.audioId || `${Date.now()}-${Math.random().toString(16).slice(2)}`); - const durationSec = Number(task.durationSec || 0); const loop = !!task.loop; const token = task.cancelToken; const sampleRate = 48000; @@ -691,7 +1108,7 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { const chunkDurationSec = 0.25; const chunkBytes = Math.max(4096, Math.floor(sampleRate * channels * (bitsPerSample / 8) * chunkDurationSec)); - targetWs.send(JSON.stringify({ + masterWs.send(JSON.stringify({ type: 'audio_start', stationName, audioId, @@ -700,9 +1117,10 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { channels, audioFormat: 'wav-stream', loop, - priority: Number(task.priority || 0), - interruptMode: task.interruptMode || getInterruptModeForPriority(Number(task.priority || 0)), - source: task.source || 'unknown' + priority: prio, + interruptMode: task.interruptMode || getInterruptModeForPriority(prio), + source: task.source || 'unknown', + cableGroupId: cableGrp?.id || '' })); const ffmpegArgs = ['-hide_banner', '-loglevel', 'error']; @@ -716,18 +1134,25 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { let proc = null; let pcmPending = Buffer.alloc(0); + const releaseToken = () => { + if (cableGrp) { + try { releaseCableToken(cableGrp.id, stationName, audioId); } catch {} + } + }; + const finish = (result, error) => { if (settled) return; settled = true; if (task.onCancel === cancelStream) task.onCancel = null; if (error) { + releaseToken(); reject(error); return; } try { - const ws = getStationWs(stationName, fallbackWs); - if (ws && ws.readyState === WebSocket.OPEN) { - ws.send(JSON.stringify({ + const { ws: endWs } = getStationMaster(stationName); + if (endWs && endWs.readyState === WebSocket.OPEN) { + endWs.send(JSON.stringify({ type: 'audio_end', stationName, audioId, @@ -735,7 +1160,12 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { })); } } catch {} - resolve(result); + if (cableGrp && !loop && durationSec > 0) { + setTimeout(() => releaseToken(), Math.max(100, Math.floor(durationSec * 1000) + 500)); + } else { + releaseToken(); + } + resolve({ ...result, cableGroupId: cableGrp?.id || '' }); }; const cancelStream = () => { @@ -756,14 +1186,15 @@ async function sendLiveWavChunksToStation(stationName, fallbackWs, task) { const flushChunk = (pcmChunk) => { if (!pcmChunk || !pcmChunk.length) return; - const ws = getStationWs(stationName, fallbackWs); - if (!ws || ws.readyState !== WebSocket.OPEN) { + const { ws: fws } = getStationMaster(stationName); + const sendWs = fws || masterWs; + if (!sendWs || sendWs.readyState !== WebSocket.OPEN) { cancelStream(); return; } try { - ws.send(buildWavChunkFromPcm(pcmChunk, sampleRate, channels, bitsPerSample)); + sendWs.send(buildWavChunkFromPcm(pcmChunk, sampleRate, channels, bitsPerSample)); } catch (err) { finish(null, err); } @@ -1025,12 +1456,15 @@ async function processStationAudioQueue(stationName, fallbackWs) { const task = q.queue.shift(); task.cancelToken = { cancelled: false }; q.current = task; + const cableGrp = getCableGroupByStationName(cleanStation); + let tokenAcquiredForThisTask = false; try { if (task.consoleStreamStart) broadcastToConsole(task.consoleStreamStart); const result = task.audioFormat === 'wav-stream-live' ? await sendLiveWavChunksToStation(cleanStation, fallbackWs, task) : await sendAudioBufferToStation(cleanStation, fallbackWs, task); + tokenAcquiredForThisTask = true; if (task.consoleStreamEnd) { if (result.cancelled) { broadcastToConsole(task.consoleStreamEnd); @@ -1042,6 +1476,9 @@ async function processStationAudioQueue(stationName, fallbackWs) { } if (task.onDone) task.onDone({ ...result, stationName: cleanStation }); } catch (e) { + if (cableGrp && !tokenAcquiredForThisTask && task.audioId) { + try { releaseCableToken(cableGrp.id, cleanStation, task.audioId); } catch {} + } if (task.consoleStreamEnd) broadcastToConsole(task.consoleStreamEnd); if (task.onDone) task.onDone({ cancelled: true, error: e?.message || String(e), stationName: cleanStation }); } finally { @@ -1057,7 +1494,48 @@ function loadData() { if (fs.existsSync(DATA_FILE)) { try { const data = JSON.parse(fs.readFileSync(DATA_FILE, 'utf8')); - if (data.stations) stations = data.stations; + if (data.stations && typeof data.stations === 'object') { + const raw = data.stations; + const newStations = {}; + for (const key of Object.keys(raw)) { + const old = raw[key]; + if (!old) continue; + const name = String(old.name || key).trim(); + newStations[name] = { + name, + code: String(old.code || ''), + platform: String(old.platform || getPlatformNumberFromName(name)), + audioCableGroup: String(old.audioCableGroup || ''), + version: String(old.version || ''), + lastSeen: Number(old.lastSeen || Date.now()), + upcomingCount: Number(old.upcomingCount || 0), + upcomingTrains: Array.isArray(old.upcomingTrains) ? old.upcomingTrains : [], + currentTrain: old.currentTrain || null, + nodes: Array.isArray(old.nodes) ? old.nodes.map(n => ({ + ...n, + ws: null, + nodeId: n.nodeId || generateNodeId(), + wsId: n.wsId || '', + connectedAt: Number(n.connectedAt || 0), + lastSeen: Number(n.lastSeen || 0) + })) : [] + }; + normalizeStationRecord(newStations[name]); + } + stations = newStations; + } + if (data.audioCableGroups && typeof data.audioCableGroups === 'object') { + audioCableGroups = data.audioCableGroups; + for (const id of Object.keys(audioCableGroups)) { + const g = audioCableGroups[id]; + if (!Array.isArray(g?.stationNames)) g.stationNames = []; + for (const stName of g.stationNames) { + if (stations[stName] && !stations[stName].audioCableGroup) { + stations[stName].audioCableGroup = id; + } + } + } + } if (data.activeTrains) activeTrains = data.activeTrains; if (data.audioVersion) audioVersion = String(data.audioVersion); if (Array.isArray(data.versions)) versions = data.versions; @@ -1078,15 +1556,33 @@ function loadData() { function saveData() { try { - // Create a copy of stations without ws objects const stationsToSave = {}; for (const key in stations) { - const { ws, ...rest } = stations[key]; - stationsToSave[key] = rest; + const st = stations[key]; + if (!st) continue; + const strippedNodes = Array.isArray(st.nodes) + ? st.nodes.map(n => { + const { ws: _w, ...rest } = n; + return rest; + }) + : []; + stationsToSave[key] = { + name: st.name, + code: st.code || '', + platform: st.platform || '', + audioCableGroup: st.audioCableGroup || '', + version: st.version || '', + lastSeen: st.lastSeen || 0, + upcomingCount: Number(st.upcomingCount) || 0, + upcomingTrains: Array.isArray(st.upcomingTrains) ? st.upcomingTrains : [], + currentTrain: st.currentTrain || null, + nodes: strippedNodes + }; } - + const data = { stations: stationsToSave, + audioCableGroups, activeTrains, audioVersion, versions, @@ -1921,26 +2417,24 @@ wss.on('connection', (ws, req) => { }); ws.on('close', () => { - // Handle disconnection if necessary + removeStationWs(ws); }); }); function handleMessage(ws, data) { - // Check if it's a console connection if (data.type === 'console_connect') { ws.isConsole = true; - // Send initial state ws.send(JSON.stringify({ type: 'init_state', stations: getMergedStationList(), activeTrains, logs: stationLogs, - stationUpdate: getStationUpdateSummary() + stationUpdate: getStationUpdateSummary(), + audioCableGroups: Object.values(audioCableGroups) })); return; } - // Handle Station Messages switch (data.type) { case 'station_register': handleStationRegister(ws, data); @@ -1963,7 +2457,6 @@ function handleMessage(ws, data) { handleStationTimeRequest(ws, data); break; case 'request_list_txt': - // TODO: Send list.txt content break; case 'heartbeat': handleHeartbeat(ws, data); @@ -2000,70 +2493,41 @@ function assignDeviceFields(target, data) { } function handleHeartbeat(ws, data) { - if (data.stationName) { - const cleanName = data.stationName.trim(); - if (!stations[cleanName]) { - stations[cleanName] = { - name: cleanName, - code: data.stationCode || '', - platform: data.stationPlatform || getPlatformNumberFromName(cleanName), - version: data.stationVersion || '', - lastSeen: Date.now(), - ws: ws - }; - assignDeviceFields(stations[cleanName], data); - normalizeStationRecord(stations[cleanName]); - const tags = []; - if (stations[cleanName].hasStationPeripheral) tags.push('HAS_STATION'); - if (stations[cleanName].hasBlockReader) tags.push('HAS_BLOCKREADER'); - if (stations[cleanName].hasCreateTarget) tags.push('HAS_TARGET'); - log('INFO', `Station auto-registered via heartbeat: "${cleanName}"${data.stationVersion ? ` ${data.stationVersion}` : ''}${tags.length ? ' [' + tags.join(',') + ']' : ''}`); - maybeSendStationUpdate(ws, cleanName, data.stationVersion || stations[cleanName].version); - broadcastUpdate(); - } else { - stations[cleanName].lastSeen = Date.now(); - if (data.stationCode) stations[cleanName].code = data.stationCode; - if (data.stationPlatform) stations[cleanName].platform = data.stationPlatform; - if (data.stationVersion) stations[cleanName].version = data.stationVersion; - const oldWS = stations[cleanName].ws; - assignDeviceFields(stations[cleanName], data); - if (oldWS !== ws) { - stations[cleanName].ws = ws; - log('INFO', `Station WebSocket updated via heartbeat: "${cleanName}"${data.stationVersion ? ` ${data.stationVersion}` : ''}`); - broadcastUpdate(); - } else if (data.upcomingTrains || data.currentTrain || data.hasStationPeripheral !== undefined) { - broadcastUpdate(); - } - normalizeStationRecord(stations[cleanName]); - maybeSendStationUpdate(ws, cleanName, data.stationVersion || stations[cleanName].version); - } + if (!data.stationName) return; + const cleanName = String(data.stationName || '').trim(); + const before = stations[cleanName]; + const result = upsertStationNode(cleanName, ws, data); + if (!result) return; + const { st, node } = result; + const wasNew = !before || before.nodes.length === 0 || (before.nodes.length === 1 && result && !before.nodes.find(n => n.wsId === node.wsId)); + + const tags = []; + if (node?.hasStationPeripheral) tags.push('HAS_STATION'); + if (node?.hasBlockReader) tags.push('HAS_BLOCKREADER'); + if (node?.hasCreateTarget) tags.push('HAS_TARGET'); + + if (wasNew) { + log('INFO', `Station auto-registered via heartbeat: "${cleanName}"${data.stationVersion ? ` ${data.stationVersion}` : ''}${tags.length ? ' [' + tags.join(',') + ']' : ''} nodeId=${node?.nodeId || ''}`); + } + maybeSendStationUpdate(ws, cleanName, data.stationVersion || st?.version); + if (wasNew || data.upcomingTrains || data.currentTrain || data.hasStationPeripheral !== undefined) { + sendStationGroupState(cleanName); + broadcastUpdate(); } } function handleScheduleSync(ws, data) { if (!data.stationName) return; const cleanName = String(data.stationName || '').trim(); - if (!stations[cleanName]) { - stations[cleanName] = { - name: cleanName, - code: data.stationCode || '', - platform: data.stationPlatform || getPlatformNumberFromName(cleanName), - version: '', - lastSeen: Date.now(), - ws: ws - }; - assignDeviceFields(stations[cleanName], data); - normalizeStationRecord(stations[cleanName]); - log('INFO', `Station auto-registered via schedule sync: "${cleanName}" (upcoming: ${Number(data.upcomingCount) || 0})`); - broadcastUpdate(); - return; + const before = stations[cleanName]; + const wasNew = !before; + const result = upsertStationNode(cleanName, ws, { ...data, syncedAt: Date.now() }); + if (!result) return; + + if (wasNew) { + log('INFO', `Station auto-registered via schedule sync: "${cleanName}" (upcoming: ${Number(data.upcomingCount) || 0}) nodeId=${result.node?.nodeId || ''}`); } - stations[cleanName].lastSeen = Date.now(); - if (data.stationCode) stations[cleanName].code = data.stationCode; - if (data.stationPlatform) stations[cleanName].platform = data.stationPlatform; - if (stations[cleanName].ws !== ws) stations[cleanName].ws = ws; - assignDeviceFields(stations[cleanName], data); - normalizeStationRecord(stations[cleanName]); + sendStationGroupState(cleanName); broadcastUpdate(); } @@ -2071,26 +2535,23 @@ function handleStationRegister(ws, data) { const { stationName, stationCode, stationPlatform, stationVersion } = data; const cleanName = stationName.trim(); - stations[cleanName] = { - name: cleanName, - code: stationCode, - platform: stationPlatform || getPlatformNumberFromName(cleanName), - version: stationVersion || '', - lastSeen: Date.now(), - ws: ws - }; - assignDeviceFields(stations[cleanName], data); - normalizeStationRecord(stations[cleanName]); - - log('SUCCESS', `Station registered: "${cleanName}" (${stationCode})${stationVersion ? ` ${stationVersion}` : ''}`); - + const result = upsertStationNode(cleanName, ws, data); + if (!result) return; + const { st, node } = result; + const cableGrp = getCableGroupByStationName(cleanName); + + log('SUCCESS', `Station registered: "${cleanName}" (${stationCode})${stationVersion ? ` ${stationVersion}` : ''} nodeId=${node?.nodeId || ''}`); + ws.send(JSON.stringify({ type: 'station_register_ack', - stationName: cleanName + stationName: cleanName, + nodeId: node?.nodeId || '', + audioCableGroup: cableGrp?.id || st?.audioCableGroup || '' })); - + maybeSendStationUpdate(ws, cleanName, stationVersion); saveData(); + sendStationGroupState(cleanName); broadcastUpdate(); } @@ -2464,7 +2925,8 @@ function broadcastUpdate() { type: 'state_update', stations: list, activeTrains, - stationUpdate: getStationUpdateSummary() + stationUpdate: getStationUpdateSummary(), + audioCableGroups: Object.values(audioCableGroups) }; broadcastToConsole(state); } @@ -2814,12 +3276,18 @@ app.post('/api/stations', express.json(), (req, res) => { name, code, platform: getPlatformNumberFromName(name), - lastSeen: Date.now() + audioCableGroup: '', + version: '', + lastSeen: Date.now(), + upcomingCount: 0, + upcomingTrains: [], + currentTrain: null, + nodes: [] }; normalizeStationRecord(stations[name]); saveData(); broadcastUpdate(); - res.json({ success: true, station: { name, code, lastSeen: stations[name].lastSeen } }); + res.json({ success: true, station: { name, code, lastSeen: stations[name].lastSeen, audioCableGroup: stations[name].audioCableGroup, nodeCount: 0 } }); }); app.put('/api/stations/:name', express.json(), (req, res) => { @@ -2834,6 +3302,16 @@ app.put('/api/stations/:name', express.json(), (req, res) => { if (newName && newName !== name) { if (stations[newName]) return res.status(409).json({ success: false, message: 'New station name already exists' }); stations[newName] = { ...st, name: newName }; + if (!Array.isArray(stations[newName].nodes)) stations[newName].nodes = []; + if (stations[newName].audioCableGroup) { + const cgId = stations[newName].audioCableGroup; + if (audioCableGroups[cgId] && Array.isArray(audioCableGroups[cgId].stationNames)) { + const arr = audioCableGroups[cgId].stationNames; + const idx = arr.indexOf(name); + if (idx >= 0) arr[idx] = newName; + else if (!arr.includes(newName)) arr.push(newName); + } + } delete stations[name]; targetName = newName; } @@ -2841,21 +3319,26 @@ app.put('/api/stations/:name', express.json(), (req, res) => { if (nextCode !== undefined) stations[targetName].code = nextCode; stations[targetName].platform = getPlatformNumberFromName(stations[targetName].name); stations[targetName].lastSeen = stations[targetName].lastSeen || Date.now(); + if (!Array.isArray(stations[targetName].nodes)) stations[targetName].nodes = []; normalizeStationRecord(stations[targetName]); saveData(); + if (targetName !== name) sendStationGroupState(targetName); broadcastUpdate(); - res.json({ success: true, station: { name: stations[targetName].name, code: stations[targetName].code, lastSeen: stations[targetName].lastSeen } }); + res.json({ success: true, station: { name: stations[targetName].name, code: stations[targetName].code, lastSeen: stations[targetName].lastSeen, audioCableGroup: stations[targetName].audioCableGroup || '', nodeCount: stations[targetName].nodes.length } }); }); app.delete('/api/stations/:name', (req, res) => { const selector = String(req.params.name || '').trim(); - const group = resolveStationGroup(selector); - if (group && Array.isArray(group.memberNames) && group.memberNames.length > 0) { - for (const memberName of group.memberNames) { - delete stations[memberName]; + if (stations[selector]) { + delete stations[selector]; + for (const id of Object.keys(audioCableGroups)) { + const g = audioCableGroups[id]; + if (Array.isArray(g.stationNames)) { + g.stationNames = g.stationNames.filter(s => s !== selector); + } } - log('INFO', `Deleted station target: ${selector} (${group.memberNames.length} node(s))`); + log('INFO', `Deleted station: ${selector}`); saveData(); broadcastUpdate(); res.json({ success: true }); @@ -2864,6 +3347,203 @@ app.delete('/api/stations/:name', (req, res) => { } }); +app.put('/api/stations/:name/cable-group', express.json(), (req, res) => { + const name = String(req.params.name || '').trim(); + const cableGroupId = String(req.body?.cableGroupId || '').trim(); + if (!stations[name]) return res.status(404).json({ success: false, message: 'Station not found' }); + const st = stations[name]; + + if (!cableGroupId) { + const oldId = st.audioCableGroup; + st.audioCableGroup = ''; + if (oldId && audioCableGroups[oldId]) { + audioCableGroups[oldId].stationNames = (audioCableGroups[oldId].stationNames || []).filter(s => s !== name); + } + saveData(); + sendStationGroupState(name); + broadcastUpdate(); + return res.json({ success: true, station: { name, audioCableGroup: '' } }); + } + if (!audioCableGroups[cableGroupId]) { + return res.status(404).json({ success: false, message: 'Cable group not found' }); + } + const grp = audioCableGroups[cableGroupId]; + + const oldId = st.audioCableGroup; + if (oldId && oldId !== cableGroupId && audioCableGroups[oldId]) { + audioCableGroups[oldId].stationNames = (audioCableGroups[oldId].stationNames || []).filter(s => s !== name); + } + st.audioCableGroup = cableGroupId; + if (!Array.isArray(grp.stationNames)) grp.stationNames = []; + if (!grp.stationNames.includes(name)) grp.stationNames.push(name); + + saveData(); + sendStationGroupState(name); + broadcastUpdate(); + res.json({ success: true, station: { name, audioCableGroup: cableGroupId } }); +}); + +app.get('/api/audio-cable-groups', (req, res) => { + const list = Object.values(audioCableGroups).map(g => ({ + id: g.id, + name: g.name || '', + color: g.color || '', + description: g.description || '', + stationNames: Array.isArray(g.stationNames) ? g.stationNames.slice() : [], + memberCount: Array.isArray(g.stationNames) ? g.stationNames.length : 0, + tokenHolder: (() => { + const tok = audioCableTokens.get(g.id); + if (!tok) return null; + return { + holderStationName: tok.holderStationName, + holderAudioId: tok.holderAudioId, + holderPriority: tok.holderPriority, + acquiredAt: tok.acquiredAt + }; + })() + })); + res.json({ success: true, groups: list }); +}); + +app.post('/api/audio-cable-groups', express.json(), (req, res) => { + const name = String(req.body?.name || '').trim(); + const color = String(req.body?.color || '').trim(); + const description = String(req.body?.description || '').trim(); + const stationNamesRaw = req.body?.stationNames; + const stationNames = Array.isArray(stationNamesRaw) + ? stationNamesRaw.map(s => String(s || '').trim()).filter(Boolean) + : []; + + if (!name) return res.status(400).json({ success: false, message: 'Group name required' }); + const id = 'cg_' + Date.now().toString(36) + '_' + Math.random().toString(36).slice(2, 6); + + audioCableGroups[id] = { + id, + name, + color, + description, + stationNames: [] + }; + + const validNames = []; + for (const sn of stationNames) { + if (!stations[sn]) continue; + stations[sn].audioCableGroup = id; + validNames.push(sn); + } + audioCableGroups[id].stationNames = validNames; + + saveData(); + for (const sn of validNames) sendStationGroupState(sn); + broadcastUpdate(); + res.status(201).json({ success: true, group: audioCableGroups[id] }); +}); + +app.put('/api/audio-cable-groups/:id', express.json(), (req, res) => { + const id = String(req.params.id || '').trim(); + const grp = audioCableGroups[id]; + if (!grp) return res.status(404).json({ success: false, message: 'Cable group not found' }); + + if (req.body?.name !== undefined) grp.name = String(req.body.name || '').trim(); + if (req.body?.color !== undefined) grp.color = String(req.body.color || '').trim(); + if (req.body?.description !== undefined) grp.description = String(req.body.description || '').trim(); + + const affected = new Set(); + if (req.body?.stationNames !== undefined) { + const raw = Array.isArray(req.body.stationNames) ? req.body.stationNames : []; + const next = raw.map(s => String(s || '').trim()).filter(Boolean); + + const prevArr = Array.isArray(grp.stationNames) ? grp.stationNames.slice() : []; + for (const oldName of prevArr) { + if (stations[oldName] && !next.includes(oldName)) { + stations[oldName].audioCableGroup = ''; + affected.add(oldName); + } + } + const valid = []; + for (const sn of next) { + if (!stations[sn]) continue; + stations[sn].audioCableGroup = id; + valid.push(sn); + affected.add(sn); + } + grp.stationNames = valid; + } + + saveData(); + for (const n of Array.from(affected)) sendStationGroupState(n); + broadcastUpdate(); + res.json({ success: true, group: grp }); +}); + +app.delete('/api/audio-cable-groups/:id', (req, res) => { + const id = String(req.params.id || '').trim(); + const grp = audioCableGroups[id]; + if (!grp) return res.status(404).json({ success: false, message: 'Cable group not found' }); + + const affected = Array.isArray(grp.stationNames) ? grp.stationNames.slice() : []; + for (const sn of affected) { + if (stations[sn]) stations[sn].audioCableGroup = ''; + } + const tok = audioCableTokens.get(id); + if (tok) { + if (tok.cancelToken) tok.cancelToken.cancelled = true; + audioCableTokens.delete(id); + } + delete audioCableGroups[id]; + + saveData(); + for (const n of affected) sendStationGroupState(n); + broadcastUpdate(); + res.json({ success: true }); +}); + +app.post('/api/audio-cable-groups/:id/stations', express.json(), (req, res) => { + const id = String(req.params.id || '').trim(); + const grp = audioCableGroups[id]; + if (!grp) return res.status(404).json({ success: false, message: 'Cable group not found' }); + const raw = Array.isArray(req.body?.stationNames) ? req.body.stationNames : []; + if (!Array.isArray(grp.stationNames)) grp.stationNames = []; + + const added = []; + for (const s of raw) { + const sn = String(s || '').trim(); + if (!sn || !stations[sn]) continue; + if (grp.stationNames.includes(sn)) continue; + const prev = stations[sn].audioCableGroup; + if (prev && prev !== id && audioCableGroups[prev]) { + audioCableGroups[prev].stationNames = (audioCableGroups[prev].stationNames || []).filter(x => x !== sn); + } + stations[sn].audioCableGroup = id; + grp.stationNames.push(sn); + added.push(sn); + } + + saveData(); + for (const n of added) sendStationGroupState(n); + broadcastUpdate(); + res.json({ success: true, added, group: grp }); +}); + +app.delete('/api/audio-cable-groups/:id/stations/:stationName', (req, res) => { + const id = String(req.params.id || '').trim(); + const grp = audioCableGroups[id]; + if (!grp) return res.status(404).json({ success: false, message: 'Cable group not found' }); + const stationName = String(req.params.stationName || '').trim(); + + if (stations[stationName] && stations[stationName].audioCableGroup === id) { + stations[stationName].audioCableGroup = ''; + } + if (Array.isArray(grp.stationNames)) { + grp.stationNames = grp.stationNames.filter(s => s !== stationName); + } + + saveData(); + sendStationGroupState(stationName); + broadcastUpdate(); + res.json({ success: true }); +}); + app.delete('/api/streams/:streamName', (req, res) => { const streamName = String(req.params.streamName || '').trim(); if (!streamName) return res.status(400).json({ success: false, message: 'Stream name required' }); diff --git a/station_broadcast.lua b/station_broadcast.lua index f2b07dd..3d7f55a 100644 --- a/station_broadcast.lua +++ b/station_broadcast.lua @@ -1,6 +1,6 @@ -- 1. Configuration & Constants -local APP_VERSION_TEXT = "v1.0.98" +local APP_VERSION_TEXT = "v1.0.99" local CONFIG = { CLOUD_URL = "ws://stabadmin.fse-media.group", @@ -2539,6 +2539,7 @@ end function Network.sendScheduleSync() if not State.ws then return end + if not Utils.isLocalStationMaster() then return end local upcomingTrainsPayload = {} if State.upcomingTrains and type(State.upcomingTrains) == "table" then for _, t in ipairs(State.upcomingTrains) do @@ -3555,19 +3556,21 @@ function Main.loop() if isPendingAlertReady(alertKey, now) and not State.announcedAlerts[alertKey] then State.announcedAlerts[alertKey] = now Utils.log("INFO", string.format("Train Alert: %s (ETA: %ds, Delay: %dm, Cancelled: %s, Reason: %s)", trainName, etaSeconds, displayDelayMinutes, tostring(cancelled), tostring(reason or ""))) - Network.send({ - type = "train_alert", - stationName = State.stationName, - stationCode = State.stationCode, - stationPlatform = platform, - trainId = trainName, - destination = destination, - eta = tostring(etaSeconds) .. "s", - etaSeconds = etaSeconds, - delay = displayDelayMinutes, - cancelled = cancelled, - reason = reason - }) + if Utils.isLocalStationMaster() then + Network.send({ + type = "train_alert", + stationName = State.stationName, + stationCode = State.stationCode, + stationPlatform = platform, + trainId = trainName, + destination = destination, + eta = tostring(etaSeconds) .. "s", + etaSeconds = etaSeconds, + delay = displayDelayMinutes, + cancelled = cancelled, + reason = reason + }) + end end end @@ -3590,19 +3593,21 @@ function Main.loop() if currentPending and (now - (currentPending.since or now)) >= stableMs and not State.announcedApproaching[announceKey] then State.announcedApproaching[announceKey] = now Utils.log("INFO", string.format("Train Approaching: %s (ETA: %.1fs, Delay: %dm)", trainName, etaSeconds, delayMinutes)) - Network.send({ - type = "train_approaching", - stationName = State.stationName, - stationCode = State.stationCode, - stationPlatform = platform, - trainId = trainName, - destination = destination, - eta = string.format("%.1fs", etaSeconds), - etaSeconds = etaSeconds, - delay = displayDelayMinutes, - cancelled = cancelled, - reason = reason - }) + if Utils.isLocalStationMaster() then + Network.send({ + type = "train_approaching", + stationName = State.stationName, + stationCode = State.stationCode, + stationPlatform = platform, + trainId = trainName, + destination = destination, + eta = string.format("%.1fs", etaSeconds), + etaSeconds = etaSeconds, + delay = displayDelayMinutes, + cancelled = cancelled, + reason = reason + }) + end end end end @@ -3717,17 +3722,19 @@ function Main.loop() if isPendingAlertReady(alertKey, now) and not State.announcedAlerts[alertKey] then State.announcedAlerts[alertKey] = now Utils.log("INFO", string.format("Train Delay Alert: %s (ETA: %ds, Delay: %dm)", tName, etaSeconds, delay)) - Network.send({ - type = "train_alert", - stationName = State.stationName, - stationCode = State.stationCode, - stationPlatform = platform or State.stationPlatform or "1", - trainId = tName, - destination = dest, - eta = etaStr, - etaSeconds = etaSeconds, - delay = delay - }) + if Utils.isLocalStationMaster() then + Network.send({ + type = "train_alert", + stationName = State.stationName, + stationCode = State.stationCode, + stationPlatform = platform or State.stationPlatform or "1", + trainId = tName, + destination = dest, + eta = etaStr, + etaSeconds = etaSeconds, + delay = delay + }) + end end end @@ -3750,17 +3757,19 @@ function Main.loop() if currentPending and (now - (currentPending.since or now)) >= stableMs and not State.announcedApproaching[announceKey] then State.announcedApproaching[announceKey] = now Utils.log("INFO", string.format("Train Approaching: %s (ETA: %ds, Delay: %dm)", tName, etaSeconds, delay)) - Network.send({ - type = "train_approaching", - stationName = State.stationName, - stationCode = State.stationCode, - stationPlatform = platform or State.stationPlatform or "1", - trainId = tName, - destination = dest, - eta = etaStr, - etaSeconds = etaSeconds, - delay = delay - }) + if Utils.isLocalStationMaster() then + Network.send({ + type = "train_approaching", + stationName = State.stationName, + stationCode = State.stationCode, + stationPlatform = platform or State.stationPlatform or "1", + trainId = tName, + destination = dest, + eta = etaStr, + etaSeconds = etaSeconds, + delay = delay + }) + end end end end @@ -3829,39 +3838,43 @@ function Main.loop() departureSent = false } - Network.send({ - type = "train_arrival", - stationName = State.stationName, - stationCode = State.stationCode, - stationPlatform = State.stationPlatform, - trainId = id, - destination = dest, - destCode = destCode, - nextStation = nextStation, - nextCode = nextCode, - lineCode = lineCode, - departureTime = State.currentTrain.departureTime, - departureBeep = State.departureBeep - }) + if Utils.isLocalStationMaster() then + Network.send({ + type = "train_arrival", + stationName = State.stationName, + stationCode = State.stationCode, + stationPlatform = State.stationPlatform, + trainId = id, + destination = dest, + destCode = destCode, + nextStation = nextStation, + nextCode = nextCode, + lineCode = lineCode, + departureTime = State.currentTrain.departureTime, + departureBeep = State.departureBeep + }) + end else local now = os.epoch("utc") / 1000 -- 增加 1.5 秒网络与预缓冲补偿，确保音频结束时列车刚好发车 local triggerOffset = (State.currentTrain.departureDuration or 10) + 1.5 if not State.currentTrain.departureSent and (State.currentTrain.departureTime - now) <= triggerOffset then State.currentTrain.departureSent = true - Network.send({ - type = "train_departure", - stationName = State.stationName, - stationCode = State.stationCode, - stationPlatform = State.stationPlatform, - trainId = State.currentTrain.id, - destination = State.currentTrain.dest, - destCode = State.currentTrain.destCode, - nextStation = State.currentTrain.nextStation, - nextCode = State.currentTrain.nextCode, - lineCode = State.currentTrain.lineCode, - departureBeep = State.departureBeep - }) + if Utils.isLocalStationMaster() then + Network.send({ + type = "train_departure", + stationName = State.stationName, + stationCode = State.stationCode, + stationPlatform = State.stationPlatform, + trainId = State.currentTrain.id, + destination = State.currentTrain.dest, + destCode = State.currentTrain.destCode, + nextStation = State.currentTrain.nextStation, + nextCode = State.currentTrain.nextCode, + lineCode = State.currentTrain.lineCode, + departureBeep = State.departureBeep + }) + end end end else @@ -3945,6 +3958,7 @@ parallel.waitForAny( Network.heartbeat, Network.syncFiles, Input.listener, - Main.loop + Main.loop, + LocalBus.loop )