diff --git a/apps/equipment_service.hpp b/apps/equipment_service.hpp index 17bf6ec..738c113 100644 --- a/apps/equipment_service.hpp +++ b/apps/equipment_service.hpp @@ -12,8 +12,13 @@ #include +#include +#include +#include #include #include +#include +#include #include #include #include @@ -155,8 +160,9 @@ inline pb::ControlState::State to_proto_state(gem::ControlState s) { class EquipmentService final : public pb::Equipment::Service { public: - // Snapshots the (immutable) name->id/format dictionaries. Construct before - // run_async() so the model is read while the io thread isn't running yet. + // Snapshots the (immutable) name->id/format dictionaries and registers the + // health observers. Construct before run_async() so the model is read (and + // observers land) while the io thread isn't running yet. explicit EquipmentService(gem::EquipmentRuntime& rt) : rt_(rt) { for (const auto& sv : rt.model().svids.all()) vars_.insert({sv.name, {sv.id, sv.value.format()}}); @@ -164,6 +170,21 @@ class EquipmentService final : public pb::Equipment::Service { vars_.insert({dv.name, {dv.id, dv.value.format()}}); for (const auto& ev : rt.model().events.all_events()) events_.insert({ev.name, ev.id}); + for (const auto& al : rt.model().alarms.all()) { + if (!al.name.empty()) alarms_.insert({al.name, al.id}); + alarms_.insert({std::to_string(al.id), al.id}); // always addressable by id + } + // Health signals: bump a version + wake WatchHealth streams whenever the + // HSMS link or the control state changes (observers fire on the io thread; + // add_ observers survive register_default_handlers' primary set_). + rt.add_link_observer([this](bool selected) { + link_selected_.store(selected, std::memory_order_relaxed); + bump_health(); + }); + rt.add_control_state_observer( + [this](gem::ControlState, gem::ControlState, gem::ControlEvent) { + bump_health(); + }); } grpc::Status SetVariables(grpc::ServerContext*, const pb::VariableUpdate* req, @@ -214,6 +235,16 @@ class EquipmentService final : public pb::Equipment::Service { return grpc::Status::OK; } + grpc::Status SetAlarm(grpc::ServerContext*, const pb::Alarm* req, + pb::Ack* resp) override { + return alarm_action(req->name(), /*set=*/true, resp); + } + + grpc::Status ClearAlarm(grpc::ServerContext*, const pb::Alarm* req, + pb::Ack* resp) override { + return alarm_action(req->name(), /*set=*/false, resp); + } + grpc::Status GetVariables(grpc::ServerContext*, const pb::VariableQuery* req, pb::VariableSnapshot* resp) override { // Resolve names against the snapshot maps (empty query = everything). @@ -258,7 +289,123 @@ class EquipmentService final : public pb::Equipment::Service { return grpc::Status::OK; } + // Operator-panel control-state transitions (e.g. "offline for maintenance"). + // Fires the operator events on the io thread and reports what the E30 table + // actually did: ACCEPT iff the equipment landed in the requested state, + // CANNOT_DO_NOW (naming the actual state) otherwise. Note the shipped table + // has no operator path to EQUIPMENT_OFFLINE — operator_offline lands + // HOST_OFFLINE — so honesty matters here. + grpc::Status RequestControlState(grpc::ServerContext*, + const pb::ControlStateRequest* req, + pb::Ack* resp) override { + using PS = pb::ControlState; + const auto desired = req->desired(); + if (desired == PS::ATTEMPT_ONLINE) { + resp->set_code(pb::Ack::PARAMETER_INVALID); + resp->set_message("ATTEMPT_ONLINE is transient; request a settled state"); + return grpc::Status::OK; + } + auto final_state = rt_.read_sync([this, desired]() { + auto& sm = rt_.control(); + switch (desired) { + case PS::ONLINE_LOCAL: + if (sm.state() == gem::ControlState::OnlineRemote) sm.operator_local(); + else if (!sm.online()) sm.operator_online(); // table chains to OnlineLocal + break; + case PS::ONLINE_REMOTE: + if (sm.state() == gem::ControlState::OnlineLocal) sm.operator_remote(); + else if (!sm.online()) { + sm.operator_online(); + sm.operator_remote(); + } + break; + case PS::HOST_OFFLINE: + case PS::EQUIPMENT_OFFLINE: + sm.operator_offline(); + break; + default: + break; + } + return sm.state(); + }); + if (!final_state) { + resp->set_code(pb::Ack::CANNOT_DO_NOW); + resp->set_message("engine io thread did not answer (not running?)"); + return grpc::Status::OK; + } + if (to_proto_state(*final_state) == desired) { + resp->set_code(pb::Ack::ACCEPT); + } else { + resp->set_code(pb::Ack::CANNOT_DO_NOW); + resp->set_message(std::string("equipment is in ") + + gem::control_state_name(*final_state)); + } + return grpc::Status::OK; + } + + // Streams a Health snapshot immediately, then again whenever the link or + // control state changes (and on spool-depth changes, sampled at the poll + // interval). Ends when the client cancels or the engine stops. + grpc::Status WatchHealth(grpc::ServerContext* ctx, const pb::Empty*, + grpc::ServerWriter* writer) override { + bool first = true; + pb::Health last; + while (!ctx->IsCancelled()) { + const uint64_t seen = health_version_.load(std::memory_order_relaxed); + auto snap = make_health(); + if (!snap) break; // engine stopped — end the stream + if (first || snap->link() != last.link() || + snap->control_state() != last.control_state() || + snap->spool_depth() != last.spool_depth()) { + if (!writer->Write(*snap)) break; + last = *snap; + first = false; + } + std::unique_lock lk(health_mu_); + health_cv_.wait_for(lk, std::chrono::milliseconds(500), [&] { + return health_version_.load(std::memory_order_relaxed) != seen; + }); + } + return grpc::Status::OK; + } + private: + void bump_health() { + health_version_.fetch_add(1, std::memory_order_relaxed); + health_cv_.notify_all(); + } + + // Build a Health snapshot. Control state + link come from atomics; spool + // depth is mutable engine state, read via read_sync. nullopt = engine down. + std::optional make_health() { + auto depth = rt_.read_sync( + [this]() { return static_cast(rt_.model().spool.size()); }); + if (!depth) return std::nullopt; + pb::Health h; + // CONNECTED (TCP up, not yet SELECTED) is reserved: the runtime's link + // observer fires on SELECTED/closed only. TODO(daemon): surface the + // intermediate state if a tool ever needs it. + h.set_link(link_selected_.load(std::memory_order_relaxed) + ? pb::Health::SELECTED + : pb::Health::DISCONNECTED); + h.set_spool_depth(*depth); + h.set_control_state(to_proto_state(rt_.control_state())); + return h; + } + + grpc::Status alarm_action(const std::string& name, bool set, pb::Ack* resp) { + auto it = alarms_.find(name); + if (it == alarms_.end()) { + resp->set_code(pb::Ack::PARAMETER_INVALID); + resp->set_message("no alarm named '" + name + "'"); + return grpc::Status::OK; + } + if (set) rt_.set_alarm(it->second); + else rt_.clear_alarm(it->second); + resp->set_code(pb::Ack::ACCEPT); + return grpc::Status::OK; + } + struct VarRef { uint32_t vid; s2::Format format; // declared wire format from equipment.yaml @@ -267,6 +414,14 @@ class EquipmentService final : public pb::Equipment::Service { gem::EquipmentRuntime& rt_; std::map vars_; // SVIDs + DVIDs (SVIDs win on clash) std::map events_; // CEID by name + std::map alarms_; // ALID by name AND stringified id + + // WatchHealth plumbing: observers (io thread) bump the version and wake the + // per-stream wait loops (gRPC threads). + std::atomic link_selected_{false}; + std::atomic health_version_{0}; + std::mutex health_mu_; + std::condition_variable health_cv_; }; } // namespace secsgem::daemon diff --git a/data/equipment.yaml b/data/equipment.yaml index ee74548..05e2721 100644 --- a/data/equipment.yaml +++ b/data/equipment.yaml @@ -56,10 +56,12 @@ ceids: - {id: 400, name: ControlJobExecuting} # E94 CJ entered Executing - {id: 401, name: ControlJobCompleted} # E94 CJ entered Completed -# Reported on S5F5 / S5F1. `category` is the lower-7 of ALCD. +# Reported on S5F5 / S5F1. `category` is the lower-7 of ALCD. `name` is an +# optional local key for the daemon's name-based API (gRPC SetAlarm/ClearAlarm); +# it never goes on the wire — SEMI only defines numeric ALID + freetext text. alarms: - - {id: 1, text: "Chiller Temp High", category: 4} - - {id: 2, text: "Door Open", category: 1} + - {id: 1, name: chiller_temp_high, text: "Chiller Temp High", category: 4} + - {id: 2, name: door_open, text: "Door Open", category: 1} # Reported on S7F19. Body served by S7F5/F6. recipes: diff --git a/docs/DAEMON_ROADMAP.md b/docs/DAEMON_ROADMAP.md index 1b04790..5bd1d3b 100644 --- a/docs/DAEMON_ROADMAP.md +++ b/docs/DAEMON_ROADMAP.md @@ -29,7 +29,7 @@ host and stays conformant while the tool software restarts/upgrades/crashes. | Daemon interop vs **secsgem-py** reference host | ✅ | `interop/daemon_interop.py` (via `gemd` compose service): gRPC `SetVariables(ChamberPressure=2.5)` + `FireEvent` → host receives `S6F11 CEID 300` carrying `` — value *and declared format* flow gRPC→engine→HSMS→host | | Daemon interop vs **secs4j** (Java) | ⬜ | mirror the secsgem-py harness against `interop/secs4j` | | `Subscribe` host→tool command stream | ⬜ | design settled (HCACK-4, see below); not implemented | -| Remaining universal RPCs (alarms, `RequestControlState`, `WatchHealth`) | ⬜ | see plan (Phase A) | +| Universal RPC surface complete (vars/events/alarms/control-state/health) | ✅ | Phase A done; daemon tests 101 assertions, interop 15 checks | | Python client package (the "beautiful API") | ⬜ | thin wrapper over generated stubs | ## Known issues (found in the 2026-06-10 audit; honest list) @@ -39,10 +39,9 @@ host and stays conformant while the tool software restarts/upgrades/crashes. observer (`HandlerSlot` primary+observers pattern), so the mirror survives `register_default_handlers` claiming the primary slot. `control_state()` is now safe from any thread. -- ⬜ **Alarms have no name key.** `equipment.yaml` alarms carry only numeric - `id` + freetext `text` (matches SEMI: ALID/ALTX; there is no standard short - name). The name-based `SetAlarm`/`ClearAlarm` RPCs need an optional local - `name:` field in the alarm config (fallback: stringified id). +- ✅ ~~**Alarms have no name key.**~~ Optional `name:` added to the alarm + config (loader + validator + shipped equipment.yaml); daemon RPCs accept + the name or the stringified ALID. - ⬜ **`pvd_tool` predates the behaviour hook.** It still hard-codes `if (rcmd=="START") recipe->start(...)` in a router handler. Migrate it to `commands.set_handler` so the flagship example showcases the intended seam. @@ -132,11 +131,20 @@ debts tax every later phase, and the most valuable tests aren't automated. through the API, read back through the API — plus empty-query-returns-all, INVALID_ARGUMENT on unknown names, and a live round-trip check in `daemon_interop.py`. -2. ⬜ Alarm `name:` config field + `SetAlarm`/`ClearAlarm` RPCs + tests. -3. ⬜ `RequestControlState` (operator online/offline) + control-state atomic - mirror (fixes the known race) + `WatchHealth` stream (link state from the - selected/closed handlers, spool depth, control state). -4. ⬜ Extend `test_daemon_service.cpp` + `daemon_interop.py` for all of the above. +2. ✅ Alarm `name:` config field (optional local key; `name` appended LAST on + the Alarm struct so existing brace-inits compile unchanged) + `SetAlarm`/ + `ClearAlarm` RPCs (addressable by config name AND stringified ALID). + Validated end-to-end: gRPC `SetAlarm(chiller_temp_high)` -> secsgem-py host + receives `S5F1 ALCD=0x84 ALID=1`. +3. ✅ `RequestControlState` — fires operator events on the io thread and + reports what the E30 table actually did (ACCEPT iff landed in the requested + state; the shipped table has NO operator path to EquipmentOffline and the + test pins that honesty). ✅ `WatchHealth` — initial snapshot + push on + link/control-state change (+ spool depth sampled at 500ms); unit-tested + incl. the change push; link state still SELECTED/DISCONNECTED only + (CONNECTED reserved, TODO in code). Interop covers RequestControlState; + WatchHealth external check rides with Phase B. +4. ✅ Done per-item above (daemon suite at 101 assertions; interop at 15 checks). ### Phase B — the command stream (the big one) 5. ⬜ Implement `Subscribe`/`CompleteCommand` per the design above, including diff --git a/include/secsgem/gem/store/alarms.hpp b/include/secsgem/gem/store/alarms.hpp index c609095..203e9cc 100644 --- a/include/secsgem/gem/store/alarms.hpp +++ b/include/secsgem/gem/store/alarms.hpp @@ -39,6 +39,11 @@ struct Alarm { // Lower 7 bits of ALCD: severity bitmap (see AlarmSeverity). Bit 7 // marks set vs cleared and is applied at emit time. uint8_t severity_category; + // Optional local key for name-based APIs (the gRPC daemon, future Python + // client). NOT on the wire — SEMI defines only numeric ALID + freetext + // ALTX. Last field so existing {id, text, category} brace-inits compile + // unchanged; empty = unnamed (address it by id). + std::string name; bool has(AlarmSeverity bit) const { return has_severity(severity_category, bit); } bool is_safety() const { diff --git a/interop/daemon_interop.py b/interop/daemon_interop.py index fe86568..eba230e 100644 --- a/interop/daemon_interop.py +++ b/interop/daemon_interop.py @@ -99,6 +99,8 @@ def run(grpc_addr: str, hsms_host: str, hsms_port: int, session_id: int, ceid300 = threading.Event() last_s6f11 = {} + s5f1_seen = threading.Event() + last_s5f1 = {} def on_s6f11(_handler, message): decoded = client.settings.streams_functions.decode(message) @@ -110,7 +112,17 @@ def run(grpc_addr: str, hsms_host: str, hsms_port: int, session_id: int, ceid300.set() client.send_response(F.SecsS06F12(0), message.header.system) + def on_s5f1(_handler, message): + decoded = client.settings.streams_functions.decode(message) + body = decoded.get() + LOG.info("[alm] S5F1 body=%r", body) + if isinstance(body, dict): + last_s5f1.update(body) + s5f1_seen.set() + client.send_response(F.SecsS05F02(0), message.header.system) + client.register_stream_function(6, 11, on_s6f11) + client.register_stream_function(5, 1, on_s5f1) client.enable() try: @@ -172,6 +184,32 @@ def run(grpc_addr: str, hsms_host: str, hsms_port: int, session_id: int, for v in vals) check("S6F11 report carries ChamberPressure=2.5 (value flowed end-to-end)", near, f"scalars={vals}") + + # --- alarms: host enables ALID 1 (S5F3), gRPC raises it BY NAME, + # host receives the unsolicited S5F1 with set-bit + ALID 1 --- + client.send_and_waitfor_response( + F.SecsS05F03({"ALED": 0x80, "ALID": 1})) + ack = stub.SetAlarm(pb.Alarm(name="chiller_temp_high")) + check("gRPC SetAlarm(chiller_temp_high) -> ACCEPT", + ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code)) + got_alarm = s5f1_seen.wait(timeout=10) + check("host received S5F1 (gRPC SetAlarm bridged to HSMS)", got_alarm) + if got_alarm: + check("S5F1 carries ALID 1 with the set bit", + last_s5f1.get("ALID") == 1 + and (int(last_s5f1.get("ALCD") or 0) & 0x80) != 0, + f"body={last_s5f1}") + stub.ClearAlarm(pb.Alarm(name="chiller_temp_high")) + + # --- operator-panel control state: take the tool offline via gRPC --- + ack = stub.RequestControlState(pb.ControlStateRequest( + desired=pb.ControlState.HOST_OFFLINE)) + check("gRPC RequestControlState(HOST_OFFLINE) -> ACCEPT", + ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code)) + cs = stub.GetControlState(pb.Empty()) + check("control state is HOST_OFFLINE after operator request", + cs.state == pb.ControlState.HOST_OFFLINE, + pb.ControlState.State.Name(cs.state)) finally: client.disable() diff --git a/src/config/loader.cpp b/src/config/loader.cpp index 959da0b..6d51644 100644 --- a/src/config/loader.cpp +++ b/src/config/loader.cpp @@ -206,6 +206,7 @@ EquipmentDescriptor load_equipment(const std::string& path, gem::EquipmentDataMo static_cast(a["id"].as()), a["text"].as(), static_cast(a["category"].as()), + a["name"] ? a["name"].as() : "", // optional local key }); } } diff --git a/src/config/validate.cpp b/src/config/validate.cpp index 302d911..be1037d 100644 --- a/src/config/validate.cpp +++ b/src/config/validate.cpp @@ -218,6 +218,10 @@ void validate_equipment_block(YAML::Node& root, Sink& sink) { // is the set/clear flag, set at emit time — must not be in YAML. as_int_in_range(als[i]["category"], sink, path + ".category", 0, 127); + // Optional local key for name-based APIs (not on the wire). If + // present it must be non-empty. + if (als[i]["name"]) + as_nonempty_string(als[i]["name"], sink, path + ".name"); if (id && !alarm_ids.insert(*id).second) { sink.error(path + ".id", "duplicate ALID " + std::to_string(*id), als[i]); diff --git a/tests/test_daemon_service.cpp b/tests/test_daemon_service.cpp index 2acaa28..2679cb3 100644 --- a/tests/test_daemon_service.cpp +++ b/tests/test_daemon_service.cpp @@ -125,6 +125,30 @@ TEST_CASE("Equipment gRPC service over an in-process channel") { check_all(rt.model().dvids.all()); } + SUBCASE("SetAlarm / ClearAlarm by config name, by stringified id, unknown rejected") { + auto call = [&](auto method, const std::string& name) { + grpc::ClientContext ctx; + pb::Alarm req; + pb::Ack resp; + req.set_name(name); + REQUIRE((stub.get()->*method)(&ctx, req, &resp).ok()); + return resp.code(); + }; + // By config name. + CHECK(call(&pb::Equipment::Stub::SetAlarm, "chiller_temp_high") == pb::Ack::ACCEPT); + rt.poll(); + CHECK(rt.model().alarms.active(1)); + CHECK(call(&pb::Equipment::Stub::ClearAlarm, "chiller_temp_high") == pb::Ack::ACCEPT); + rt.poll(); + CHECK_FALSE(rt.model().alarms.active(1)); + // By stringified ALID — always works, even for unnamed alarms. + CHECK(call(&pb::Equipment::Stub::SetAlarm, "2") == pb::Ack::ACCEPT); + rt.poll(); + CHECK(rt.model().alarms.active(2)); + // Unknown name. + CHECK(call(&pb::Equipment::Stub::SetAlarm, "no_such_alarm") == pb::Ack::PARAMETER_INVALID); + } + SUBCASE("FireEvent accepts a known event and rejects an unknown one") { { grpc::ClientContext ctx; @@ -205,6 +229,58 @@ TEST_CASE("GetVariables round-trip under run_async (production threading mode)") CHECK(st.error_message().find("definitely_not_a_var") != std::string::npos); } + SUBCASE("RequestControlState walks the E30 table; WatchHealth pushes the change") { + // Open the health stream first: the initial snapshot arrives immediately. + grpc::ClientContext health_ctx; + pb::Empty empty; + auto reader = stub->WatchHealth(&health_ctx, empty); + pb::Health h; + REQUIRE(reader->Read(&h)); + CHECK(h.link() == pb::Health::DISCONNECTED); // no HSMS host in this test + CHECK(h.control_state() == pb::ControlState::HOST_OFFLINE); + CHECK(h.spool_depth() == 0); + + auto request = [&](pb::ControlState::State desired) { + grpc::ClientContext ctx; + pb::ControlStateRequest req; + pb::Ack resp; + req.set_desired(desired); + REQUIRE(stub->RequestControlState(&ctx, req, &resp).ok()); + return resp; + }; + + // HostOffline --operator_online--> (AttemptOnline) --> OnlineLocal. + CHECK(request(pb::ControlState::ONLINE_LOCAL).code() == pb::Ack::ACCEPT); + CHECK(rt.control_state() == gem::ControlState::OnlineLocal); + + // The stream pushes the change (well inside its 500ms poll interval). + REQUIRE(reader->Read(&h)); + CHECK(h.control_state() == pb::ControlState::ONLINE_LOCAL); + + // OnlineLocal --operator_remote--> OnlineRemote. + CHECK(request(pb::ControlState::ONLINE_REMOTE).code() == pb::Ack::ACCEPT); + CHECK(rt.control_state() == gem::ControlState::OnlineRemote); + + // Transient state is not requestable. + CHECK(request(pb::ControlState::ATTEMPT_ONLINE).code() == pb::Ack::PARAMETER_INVALID); + + // The shipped table has NO operator path to EquipmentOffline: + // operator_offline lands HostOffline — the API must say so honestly. + auto resp = request(pb::ControlState::EQUIPMENT_OFFLINE); + CHECK(resp.code() == pb::Ack::CANNOT_DO_NOW); + CHECK(resp.message().find("HostOffline") != std::string::npos); + CHECK(rt.control_state() == gem::ControlState::HostOffline); + + // Requesting HOST_OFFLINE (the state operators actually get) succeeds — + // idempotently, since we're already there. + CHECK(request(pb::ControlState::HOST_OFFLINE).code() == pb::Ack::ACCEPT); + + health_ctx.TryCancel(); + pb::Health drain; + while (reader->Read(&drain)) {} + (void)reader->Finish(); // CANCELLED — expected + } + server->Shutdown(); rt.stop(); } diff --git a/tests/test_loader.cpp b/tests/test_loader.cpp index 2204775..5a327fd 100644 --- a/tests/test_loader.cpp +++ b/tests/test_loader.cpp @@ -1,5 +1,6 @@ #include +#include #include #include @@ -144,3 +145,24 @@ TEST_CASE("loader: control_job_state.yaml -> non-empty table") { REQUIRE(row->to.has_value()); CHECK(*row->to == gem::ControlJobState::Completed); } + +TEST_CASE("alarm optional name: parsed when present, empty when absent") { + gem::EquipmentDataModel m; + config::load_equipment(SECSGEM_DATA_DIR "/equipment.yaml", m); + // data/equipment.yaml names both alarms. + CHECK(m.alarms.get(1)->name == "chiller_temp_high"); + CHECK(m.alarms.get(2)->name == "door_open"); + + // A config without `name:` yields an empty name (back-compat). + const auto path = std::filesystem::temp_directory_path() / "alarm_noname.yaml"; + { + std::ofstream f(path); + f << "device: {id: 0, model_name: M, software_rev: R}\n" + "alarms:\n" + " - {id: 9, text: \"Unnamed\", category: 1}\n"; + } + gem::EquipmentDataModel m2; + config::load_equipment(path.string(), m2); + CHECK(m2.alarms.get(9)->name.empty()); + std::filesystem::remove(path); +}