From 0aee662f7636e7b9df2e4ef9773fc81f63fcdeef Mon Sep 17 00:00:00 2001 From: Martijn Versluis Date: Tue, 7 Jul 2026 14:27:19 +0200 Subject: [PATCH 1/3] Route the RM to a configurable CEM endpoint The RM defaults to the bundled demo CEM stand-in inside rm-mqtt-ws-bridge/cem/. This change lets an operator point the RM at any CEM without editing compose files: * CEM_BASE_URL is now a docker-compose passthrough. demo.sh reads it and, if unset, probes cem-reference-1 on localhost:8003 and points the RM at ws://host.docker.internal:8003/ws when it answers. Only if there is no CEM_BASE_URL and no reference CEM running does it fall back to the bundled stand-in, which now sits under a "bundled-cem" compose profile so it stays offline in the other two modes. * MOSQUITTO_HOST_PORT and RESOURCE_ID were also promoted to passthroughs so they can be overridden when the defaults clash with something on the host. * The three-panel demo screen no longer hard-codes ws://localhost:9000/s2 or the CEM iframe. It fetches /traffic from the RM on load (and every 2s) and, when CEM_BASE_URL points outside the compose network, replaces the CEM iframe with an "S2 CEM external" tile that shows the live endpoint and link state. When the bundled stand-in is active the iframe is restored, so the original click-through demo still works. * TrafficController now sets Access-Control-Allow-Origin so the demo panel can read /traffic from a different origin. --- rm-mqtt-ws-bridge/demo.sh | 30 ++++- rm-mqtt-ws-bridge/demo/index.html | 127 +++++++++++++++++- rm-mqtt-ws-bridge/docker-compose.yml | 20 ++- .../rm/app/controllers/traffic_controller.rb | 5 + 4 files changed, 170 insertions(+), 12 deletions(-) diff --git a/rm-mqtt-ws-bridge/demo.sh b/rm-mqtt-ws-bridge/demo.sh index 65b9f68..fb9ec65 100755 --- a/rm-mqtt-ws-bridge/demo.sh +++ b/rm-mqtt-ws-bridge/demo.sh @@ -3,7 +3,31 @@ set -euo pipefail cd "$(dirname "$0")" -REQUIRED_PORTS=(8080 3000 9000 4000 1883) +REFERENCE_CEM_HEALTH_URL="${REFERENCE_CEM_HEALTH_URL:-http://localhost:8003/health}" +REFERENCE_CEM_WS_URL="${REFERENCE_CEM_WS_URL:-ws://host.docker.internal:8003/ws}" + +reference_cem_reachable() { + curl -fs -o /dev/null --max-time 2 "$REFERENCE_CEM_HEALTH_URL" 2>/dev/null +} + +COMPOSE_PROFILES_ARGS=() +if [ -n "${CEM_BASE_URL:-}" ]; then + echo "Using external CEM (via CEM_BASE_URL): $CEM_BASE_URL" + echo "The bundled CEM stand-in will not start." +elif reference_cem_reachable; then + export CEM_BASE_URL="$REFERENCE_CEM_WS_URL" + echo "Reference CEM detected at $REFERENCE_CEM_HEALTH_URL." + echo "Wiring the RM to $CEM_BASE_URL. The bundled CEM stand-in will not start." +else + echo "No CEM_BASE_URL supplied and no reference CEM answering at $REFERENCE_CEM_HEALTH_URL." + echo "Falling back to the bundled CEM stand-in (simulation mode)." + COMPOSE_PROFILES_ARGS=(--profile bundled-cem) +fi + +REQUIRED_PORTS=(8080 3000 4000 1883) +if [ "${#COMPOSE_PROFILES_ARGS[@]}" -gt 0 ]; then + REQUIRED_PORTS+=(9000) +fi if command -v lsof >/dev/null 2>&1; then for port in "${REQUIRED_PORTS[@]}"; do if lsof -iTCP:"$port" -sTCP:LISTEN >/dev/null 2>&1; then @@ -12,8 +36,8 @@ if command -v lsof >/dev/null 2>&1; then done fi -echo "Building and starting the three applications + the demo screen (this can take a few minutes the first time)..." -docker compose up --build -d +echo "Building and starting the applications + the demo screen (this can take a few minutes the first time)..." +docker compose "${COMPOSE_PROFILES_ARGS[@]}" up --build -d wait_for() { for _ in $(seq 1 60); do diff --git a/rm-mqtt-ws-bridge/demo/index.html b/rm-mqtt-ws-bridge/demo/index.html index fdcd709..2416979 100644 --- a/rm-mqtt-ws-bridge/demo/index.html +++ b/rm-mqtt-ws-bridge/demo/index.html @@ -61,9 +61,60 @@ .pane.rm { flex: 1; } .pane.mqtt { flex: 1; } .pane iframe { flex: 1; width: 100%; border: 0; background: var(--bg); } - .pane.cem iframe, .addr-cell.cem .addr-bar { background: #eaedf3; } + .pane.cem iframe, .pane.cem .external-cem, .addr-cell.cem .addr-bar { background: #eaedf3; } .pane.rm iframe, .addr-cell.rm .addr-bar { background: #e9f1ec; } .pane.mqtt iframe, .addr-cell.mqtt .addr-bar { background: #f2efe9; } + + .external-cem { + flex: 1; + display: flex; + flex-direction: column; + gap: 14px; + padding: 22px 24px; + font-family: var(--sans); + color: var(--text); + overflow: auto; + } + .external-cem .panel-title { + font-size: 15px; + font-weight: 700; + display: flex; + align-items: center; + gap: 8px; + } + .external-cem .panel-tag { + font-size: 10.5px; + font-weight: 600; + letter-spacing: 0.04em; + text-transform: uppercase; + color: #2f6b3c; + background: #d5ecd8; + border-radius: 999px; + padding: 2px 8px; + } + .external-cem .panel-role { color: var(--muted); font-size: 13px; line-height: 1.4; } + .external-cem .kv { + display: grid; + grid-template-columns: max-content 1fr; + column-gap: 12px; + row-gap: 6px; + font-size: 12.5px; + background: var(--panel); + border: 1px solid var(--border); + border-radius: 6px; + padding: 12px 14px; + } + .external-cem .kv .k { color: var(--muted); font-weight: 600; } + .external-cem .kv .v { font-family: var(--mono); word-break: break-all; } + .external-cem .live-dot { + display: inline-block; + width: 8px; + height: 8px; + border-radius: 50%; + background: #52a462; + box-shadow: 0 0 0 3px rgba(82, 164, 98, 0.18); + } + .external-cem .live-dot.offline { background: #b9b9b9; box-shadow: 0 0 0 3px rgba(0, 0, 0, 0.08); } @@ -71,7 +122,7 @@
- +
@@ -87,12 +138,81 @@
-
+
diff --git a/rm-mqtt-ws-bridge/docker-compose.yml b/rm-mqtt-ws-bridge/docker-compose.yml index 1f6772e..dd8690d 100644 --- a/rm-mqtt-ws-bridge/docker-compose.yml +++ b/rm-mqtt-ws-bridge/docker-compose.yml @@ -18,7 +18,7 @@ services: mosquitto: image: eclipse-mosquitto:2 ports: - - "1883:1883" + - "${MOSQUITTO_HOST_PORT:-1883}:1883" volumes: - ./mqtt/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro @@ -31,13 +31,16 @@ services: condition: service_started cem: condition: service_healthy + required: false + extra_hosts: + - "host.docker.internal:host-gateway" environment: RAILS_ENV: production RAILS_LOG_LEVEL: info SECRET_KEY_BASE: "demo-secret-key-base-for-standalone-reference-only" DATABASE_URL: "postgres://postgres:postgres@postgres:5432/resource_manager_mqtt_ruby" MQTT_URL: "mqtt://mosquitto:1883" - CEM_BASE_URL: "ws://cem:9000/s2" + CEM_BASE_URL: ${CEM_BASE_URL:-ws://cem:9000/s2} S2_PROVIDES_FORECAST: "true" SOLID_QUEUE_IN_PUMA: "true" DEMO_ALLOW_FRAMING: "1" @@ -47,6 +50,7 @@ services: cem: build: ./cem + profiles: ["bundled-cem"] environment: PORT: "9000" healthcheck: @@ -64,16 +68,20 @@ services: environment: PORT: "4000" MQTT_URL: "mqtt://mosquitto:1883" - RESOURCE_ID: "demo-resource-001" + RESOURCE_ID: ${RESOURCE_ID:-demo-resource-001} ports: - "4000:4000" demo: build: ./demo depends_on: - - rm - - cem - - mqtt-panel + rm: + condition: service_started + mqtt-panel: + condition: service_started + cem: + condition: service_started + required: false ports: - "8080:80" diff --git a/rm-mqtt-ws-bridge/rm/app/controllers/traffic_controller.rb b/rm-mqtt-ws-bridge/rm/app/controllers/traffic_controller.rb index 6f0af32..6c97121 100644 --- a/rm-mqtt-ws-bridge/rm/app/controllers/traffic_controller.rb +++ b/rm-mqtt-ws-bridge/rm/app/controllers/traffic_controller.rb @@ -1,5 +1,6 @@ class TrafficController < ApplicationController skip_forgery_protection + before_action :allow_cross_origin def reset TrafficLog.clear @@ -23,6 +24,10 @@ def index private + def allow_cross_origin + response.headers["Access-Control-Allow-Origin"] = "*" + end + def connection_state(resource) connection = Supervisor.instance.find_connection(resource.s2_identifier) return connection.status.to_s if connection From dc05a35f2a88a6b0725d608bddbf91789467cfdf Mon Sep 17 00:00:00 2001 From: Martijn Versluis Date: Tue, 7 Jul 2026 14:27:49 +0200 Subject: [PATCH 2/3] Keep FRBC system description IDs stable across reconnects FRBCSystemDescriptionFactory used SecureRandom.uuid for the actuator, both operation modes, and the transitions. Every reconnect (or RM restart) produced a fresh set, so any earlier FRBC.Instruction persisted on the CEM referenced IDs that no longer existed in the current system description. The wepositive CEM surfaces that as a warning triangle next to the actuator_id and operation_mode fields on stored instructions. The structural IDs are now derived deterministically via UUIDv5, namespaced per role and per resource: Digest::UUID.uuid_v5(NAMESPACE, "/") Same resource + same role -> same UUID, across reconnects and restarts, with no persistent state to maintain. Only message_id stays random per emission, as the S2 spec requires. --- .../lib/frbc_system_description_factory.rb | 24 +++++++++++++------ 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/rm-mqtt-ws-bridge/rm/app/lib/frbc_system_description_factory.rb b/rm-mqtt-ws-bridge/rm/app/lib/frbc_system_description_factory.rb index 199bfaf..3220299 100644 --- a/rm-mqtt-ws-bridge/rm/app/lib/frbc_system_description_factory.rb +++ b/rm-mqtt-ws-bridge/rm/app/lib/frbc_system_description_factory.rb @@ -2,6 +2,11 @@ class FRBCSystemDescriptionFactory < BaseFactory OPERATION_MODE_OFF_LABEL = "Off".freeze OPERATION_MODE_CHARGING_LABEL = "Charging".freeze + # Namespace so structural UUIDs (actuator, operation modes, transitions) stay stable across + # reconnects and restarts. Otherwise every new system description would carry fresh IDs and any + # earlier CEM-side instruction would reference actuators/operation_modes that no longer exist. + STRUCTURAL_UUID_NAMESPACE = "f5b1a51b-8f7f-4c78-a1c3-3f2d2b3e2f7c".freeze + # Workarounds to simplify EV control for CEM interoperability: # - Extremely low fill_rate effectively disables the buffer, so the CEM bases control only on power # - Extended fill_level_range (-10 to 110) provides headroom at SoC 0% and 100% @@ -11,8 +16,9 @@ class FRBCSystemDescriptionFactory < BaseFactory class << self def create(resource) # rubocop:disable Metrics/MethodLength - off_id = SecureRandom.uuid - charging_id = SecureRandom.uuid + actuator_id = structural_id(resource, "actuator") + off_id = structural_id(resource, "operation_mode:off") + charging_id = structural_id(resource, "operation_mode:charging") commodity_quantity = commodity_quantity_for(resource) minimum_power = calc_minimum_power(resource) maximum_power = calc_maximum_power(resource) @@ -23,7 +29,7 @@ def create(resource) # rubocop:disable Metrics/MethodLength "valid_from" => Time.current.iso8601, "actuators" => [ { - "id" => SecureRandom.uuid, + "id" => actuator_id, "diagnostic_label" => "Electric Vehicle Charger", "supported_commodities" => [ S2::Messages::Commodity::Electricity, @@ -57,8 +63,8 @@ def create(resource) # rubocop:disable Metrics/MethodLength ), ], "transitions" => [ - create_transition(from: off_id, to: charging_id), - create_transition(from: charging_id, to: off_id), + create_transition(from: off_id, to: charging_id, seed: structural_id(resource, "transition:off_to_charging")), + create_transition(from: charging_id, to: off_id, seed: structural_id(resource, "transition:charging_to_off")), ], "timers" => [], }, @@ -79,6 +85,10 @@ def create(resource) # rubocop:disable Metrics/MethodLength private + def structural_id(resource, role) + Digest::UUID.uuid_v5(STRUCTURAL_UUID_NAMESPACE, "#{resource.s2_identifier}/#{role}") + end + def create_operation_mode(id:, label:, fill_level_range:, fill_rate:, power_ranges:) { "id" => id, @@ -94,9 +104,9 @@ def create_operation_mode(id:, label:, fill_level_range:, fill_rate:, power_rang } end - def create_transition(from:, to:) + def create_transition(from:, to:, seed:) { - "id" => SecureRandom.uuid, + "id" => seed, "from" => from, "to" => to, "start_timers" => [], From 4768877c9bb078232ee4973d1fc6f4118726e1e4 Mon Sep 17 00:00:00 2001 From: Martijn Versluis Date: Tue, 7 Jul 2026 14:28:03 +0200 Subject: [PATCH 3/3] Fold reception statuses and raw bodies into the RM dashboard feed The RM dashboard feed treated ReceptionStatus as an ordinary row, gave S2 setup messages no direction indicator, hid the raw payload of S2 messages entirely, and dropped the RM->CEM InstructionStatusUpdate that follows the reverse path. * TrafficLog now stores the parsed body plus message_id / subject_message_id / status for each entry. S2MessageLogger and MQTTMessageLogger extract those fields from the payload and pass them through. * ReceptionStatus is no longer its own row. Every S2 message row shows a small badge next to its message type: green tick for status=OK, red ! for any other status, grey ... while no ack has arrived yet. The badge is a toggle: clicking it opens the raw ReceptionStatus JSON inline. * Every S2 and MQTT message row is clickable and expands to reveal its raw JSON body, matching the "> raw message" affordance the MQTT panel already provides. Open/closed state is preserved across the 1.2s poll cycle via a stable id hashed from peer/direction/timestamp/message_id/subject_message_id. * Setup rows (Handshake, HandshakeResponse, ResourceManagerDetails, SelectControlType, FRBC.SystemDescription) now carry a direction arrow so RM->CEM and CEM->RM are distinguishable at a glance. * The FWD map now includes instruction_update -> InstructionStatusUpdate, so the RM->CEM acknowledgement of an FRBC.Instruction shows up on its own row (with the CEM's ack badge) instead of being silently dropped. --- .../rm/app/lib/mqtt_message_logger.rb | 10 +- .../rm/app/lib/s2_message_logger.rb | 28 +-- rm-mqtt-ws-bridge/rm/app/lib/traffic_log.rb | 14 +- .../rm/app/views/resources/index.html.erb | 163 ++++++++++++++++-- 4 files changed, 187 insertions(+), 28 deletions(-) diff --git a/rm-mqtt-ws-bridge/rm/app/lib/mqtt_message_logger.rb b/rm-mqtt-ws-bridge/rm/app/lib/mqtt_message_logger.rb index 09e726e..d39813d 100644 --- a/rm-mqtt-ws-bridge/rm/app/lib/mqtt_message_logger.rb +++ b/rm-mqtt-ws-bridge/rm/app/lib/mqtt_message_logger.rb @@ -6,12 +6,12 @@ def initialize(logger) end def log_inbound(topic:, message:) - TrafficLog.record(peer: "mqtt", direction: "in", message_type: message_type_from(message), detail: detail_from(message)) + TrafficLog.record(peer: "mqtt", direction: "in", message_type: message_type_from(message), detail: detail_from(message), body: body_from(message)) log("Inbound MQTT message", topic:, message:) end def log_outbound(topic:, message:) - TrafficLog.record(peer: "mqtt", direction: "out", message_type: message_type_from(message), detail: detail_from(message)) + TrafficLog.record(peer: "mqtt", direction: "out", message_type: message_type_from(message), detail: detail_from(message), body: body_from(message)) log("Outbound MQTT message", topic:, message:) end @@ -24,6 +24,12 @@ def message_type_from(message) nil end + def body_from(message) + message.is_a?(String) ? JSON.parse(message) : message + rescue StandardError + nil + end + def detail_from(message) parsed = message.is_a?(String) ? JSON.parse(message) : message data = parsed["data"] diff --git a/rm-mqtt-ws-bridge/rm/app/lib/s2_message_logger.rb b/rm-mqtt-ws-bridge/rm/app/lib/s2_message_logger.rb index bd1402a..f8cebdf 100644 --- a/rm-mqtt-ws-bridge/rm/app/lib/s2_message_logger.rb +++ b/rm-mqtt-ws-bridge/rm/app/lib/s2_message_logger.rb @@ -4,15 +4,15 @@ def initialize(logger) end def log_inbound(resource_id:, message:) - message_type = extract_message_type(message) - TrafficLog.record(peer: "cem", direction: "in", message_type:) - log("Inbound S2 message", resource_id:, message:, message_type:) + parsed = parse(message) + TrafficLog.record(peer: "cem", direction: "in", body: parsed, **traffic_attrs(parsed)) + log("Inbound S2 message", resource_id:, message:, message_type: parsed["message_type"]) end def log_outbound(resource_id:, message:) - message_type = extract_message_type(message) - TrafficLog.record(peer: "cem", direction: "out", message_type:) - log("Outbound S2 message", resource_id:, message:, message_type:) + parsed = parse(message) + TrafficLog.record(peer: "cem", direction: "out", body: parsed, **traffic_attrs(parsed)) + log("Outbound S2 message", resource_id:, message:, message_type: parsed["message_type"]) end private @@ -32,10 +32,18 @@ def log(log_message, resource_id:, message:, message_type:) raise if Rails.env.test? end - def extract_message_type(message) - parsed = JSON.parse(message) - parsed["message_type"] + def parse(message) + JSON.parse(message) rescue JSON::ParserError - nil + {} + end + + def traffic_attrs(parsed) + { + message_type: parsed["message_type"], + message_id: parsed["message_id"], + subject_message_id: parsed["subject_message_id"], + status: parsed["status"], + } end end diff --git a/rm-mqtt-ws-bridge/rm/app/lib/traffic_log.rb b/rm-mqtt-ws-bridge/rm/app/lib/traffic_log.rb index 02baefd..3e81e9d 100644 --- a/rm-mqtt-ws-bridge/rm/app/lib/traffic_log.rb +++ b/rm-mqtt-ws-bridge/rm/app/lib/traffic_log.rb @@ -3,10 +3,20 @@ class TrafficLog PATH = Rails.root.join("tmp", "traffic.jsonl") class << self - def record(peer:, direction:, message_type:, detail: nil) + def record(peer:, direction:, message_type:, detail: nil, message_id: nil, subject_message_id: nil, status: nil, body: nil) return if message_type.blank? - entry = JSON.generate({ peer: peer, direction: direction, message_type: message_type, detail: detail, at: Time.current.utc.iso8601 }.compact) + entry = JSON.generate({ + peer: peer, + direction: direction, + message_type: message_type, + detail: detail, + message_id: message_id, + subject_message_id: subject_message_id, + status: status, + body: body, + at: Time.current.utc.iso8601, + }.compact) File.open(PATH, File::RDWR | File::CREAT, 0o644) do |file| file.flock(File::LOCK_EX) kept = file.read.split("\n").last(MAX - 1) diff --git a/rm-mqtt-ws-bridge/rm/app/views/resources/index.html.erb b/rm-mqtt-ws-bridge/rm/app/views/resources/index.html.erb index 943418e..19ef42c 100644 --- a/rm-mqtt-ws-bridge/rm/app/views/resources/index.html.erb +++ b/rm-mqtt-ws-bridge/rm/app/views/resources/index.html.erb @@ -34,6 +34,46 @@ .trow .pipe .arr.flip svg { transform: scaleX(-1); } .trow.is-setup .what { font-weight: 500; color: var(--muted); } + .ack { + display: inline-flex; + align-items: center; + justify-content: center; + width: 16px; + height: 16px; + margin-left: 6px; + font-size: 10px; + font-weight: 700; + line-height: 1; + border-radius: 50%; + vertical-align: middle; + border: 0; + padding: 0; + cursor: pointer; + font-family: inherit; + } + .ack.ok { background: #d5ecd8; color: #2f6b3c; } + .ack.err { background: #f4d3d3; color: #a12727; } + .ack.pending { background: #ececec; color: #888; cursor: default; } + + .trow.expandable { cursor: pointer; } + .trow.expandable:hover { background: rgba(0,0,0,0.02); } + .raw { + grid-column: 1 / -1; + display: none; + margin: 4px 0 8px 62px; + padding: 8px 10px; + background: #fafafa; + border: 1px solid var(--border); + border-radius: 4px; + font-family: var(--mono); + font-size: 11px; + color: var(--text); + white-space: pre; + overflow-x: auto; + } + .raw.open { display: block; } + .raw.rs { border-left: 3px solid #2f6b3c; background: #f2faf3; } +