diff --git a/README.md b/README.md index 3b3b37b..1fd9c7c 100644 --- a/README.md +++ b/README.md @@ -58,9 +58,9 @@ device support. For now, the supported interfaces are the following: [these](https://github.com/eclipse-paho/paho.mqtt.cpp?tab=readme-ov-file#build-the-paho-c-and-paho-c-libraries-together) instructions. - > Note: AutoparamDriver explicitly requires EPICS >= 7.0, see - > [reference](https://github.com/Cosylab/autoparamDriver/blob/main/autoparamDriverSup/src/Makefile#L15). For this reason, - > this module requires EPICS 7.0 or later to be built and used. +> Note: AutoparamDriver explicitly requires EPICS >= 7.0, see +> [reference](https://github.com/Cosylab/autoparamDriver/blob/main/autoparamDriverSup/src/Makefile#L15). For this +> reason, this module requires EPICS 7.0 or later to be built and used. 2. Clone this repository: @@ -117,9 +117,12 @@ Where: - `` is the format of the payload: `FLAT` or `JSON`. - `` is the general type of the expected value [`INT|FLOAT|DIGITAL|STRING|INTARRAY|FLOATARRAY`]. - `` is the MQTT topic to which the record will be subscribed/published. -- `` is JSON pointer to extract value from a JSON payload (e.g. `/sensor/temperature`). If empty, JSON root is used. +- `` is JSON pointer to extract value from a JSON payload (e.g. `/sensor/temperature`). If empty, JSON root is + used. -> **Note on JSON write support:** Writing to JSON-formatted topics is currently **not supported**. At the moment the driver has no way of knowing the JSON structure expected by the broker ahead of time for write records. For this reason, only `FLAT` format can be used for output records. +> **Note on JSON write:** JSON write currently publishes the record value as a plain JSON scalar/array. Composing JSON +> objects within the driver is not yet supported (see [#14](https://github.com/epics-modules/mqtt/issues/14)). To write +> a structured JSON payload, compose it at the application level using a `FLAT:STRING` record and a `scalcout` record. **Important: Due to the pub/sub nature of MQTT, ALL input records are expected to be `I/O Intr`.** @@ -166,7 +169,6 @@ Example: Below are the supported interfaces and their implementation status. - | Message type | Asyn Parameter Type | `FORMAT:TYPE` string to use | Direction | Status | | ------------------- | -------------------------------------- | --------------------------- | ------------ | --------- | | Integer | asynInt32 | `FLAT:INT` | Read / Write | Supported | @@ -175,22 +177,23 @@ Below are the supported interfaces and their implementation status. | Strings | asynOctetRead/asynOctetWrite | `FLAT:STRING` | Read / Write | Supported | | Integer Array | asynInt32ArrayIn/asynInt32ArrayOut | `FLAT:INTARRAY` | Read / Write | Supported | | Float Array | asynFloat64ArrayIn/asynFloat64ArrayOut | `FLAT:FLOATARRAY` | Read / Write | Supported | -| Integer | asynInt32 | `JSON:INT` | Read only | Supported | -| Float | asynFloat64 | `JSON:FLOAT` | Read only | Supported | -| Bit masked | asynUInt32Digital | `JSON:DIGITAL` | Read only | Supported | -| String | asynOctetRead | `JSON:STRING` | Read only | Supported | +| Integer | asynInt32 | `JSON:INT` | Read / Write | Supported | +| Float | asynFloat64 | `JSON:FLOAT` | Read / Write | Supported | +| Bit masked | asynUInt32Digital | `JSON:DIGITAL` | Read / Write | Supported | +| String | asynOctetRead/asynOctetWrite | `JSON:STRING` | Read / Write | Supported | +| Integer Array | asynInt32ArrayIn/asynInt32ArrayOut | `JSON:INTARRAY` | Read / Write | Supported | +| Float Array | asynFloat64ArrayIn/asynFloat64ArrayOut | `JSON:FLOATARRAY` | Read / Write | Supported | ## Licensing Terms Copyright (C) 2026 André Favoto -This program is free software: you can redistribute it and/or modify it under the terms of the GNU -General Public License as published by the Free Software Foundation, either version 3 of the -License, or (at your option) any later version. +This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public +License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later +version. -This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without -even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -General Public License for more details. +This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied +warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. -You should have received a copy of the GNU General Public License along with this program. If not, -see . +You should have received a copy of the GNU General Public License along with this program. If not, see +. diff --git a/mqttSup/src/drvMqtt.cpp b/mqttSup/src/drvMqtt.cpp index 9d9853f..3c1ac32 100644 --- a/mqttSup/src/drvMqtt.cpp +++ b/mqttSup/src/drvMqtt.cpp @@ -258,7 +258,13 @@ void MqttDriver::onMessageCb(Autoparam::Driver* driver, const std::string& topic if (addr.format == MqttTopicAddr::JSON) { try { json root = json::parse(payload); - val = to_string(root.at(json::json_pointer(addr.jsonField))); + auto jsonVal = root.at(json::json_pointer(addr.jsonField)); + val = jsonVal.is_string() ? jsonVal.get() : to_string(jsonVal); + } + catch (const json::out_of_range& e) { + // there can be multiple records pointing to same topic + // topicName is not enough to differentiate (json pointer may differ) + continue; } catch (const std::exception& e) { asynPrint(pself->pasynUserSelf, ASYN_TRACE_ERROR, @@ -557,8 +563,10 @@ WriteResult MqttDriver::integerWrite(DeviceVariable& deviceVar, epicsInt32 value status = asynSuccess; } else if (addr.format == MqttTopicAddr::TopicFormat::JSON) { - // TODO: implement JSON support for integer values - throw std::logic_error("JSON support not implemented"); + // TODO: Compose JSON from driver instead of publishing a plain JSON scalar. + // See https://github.com/epics-modules/mqtt/issues/14 + driver->mqttClient.publish(addr.topicName, to_string(value)); + status = asynSuccess; } } catch (const std::exception& exc) { @@ -599,8 +607,10 @@ WriteResult MqttDriver::digitalWrite(DeviceVariable& deviceVar, epicsUInt32 cons status = asynSuccess; } else if (addr.format == MqttTopicAddr::TopicFormat::JSON) { - // TODO: implement JSON support for digital values - throw std::logic_error("JSON support not implemented"); + // TODO: Compose JSON from driver instead of publishing a plain JSON scalar. + // See https://github.com/epics-modules/mqtt/issues/14 + driver->mqttClient.publish(addr.topicName, to_string(value)); + status = asynSuccess; } } catch (const std::exception& exc) { @@ -626,8 +636,10 @@ WriteResult MqttDriver::floatWrite(DeviceVariable& deviceVar, epicsFloat64 value status = asynSuccess; } else if (addr.format == MqttTopicAddr::TopicFormat::JSON) { - // TODO: implement JSON support for float values - throw std::logic_error("JSON support not implemented"); + // TODO: Compose JSON from driver instead of publishing a plain JSON scalar. + // See https://github.com/epics-modules/mqtt/issues/14 + driver->mqttClient.publish(addr.topicName, to_string(value)); + status = asynSuccess; } } catch (const std::exception& exc) { @@ -649,10 +661,9 @@ WriteResult MqttDriver::arrayWrite(DeviceVariable& deviceVar, Array(deviceVar).driver; try { + const epicsDataType* arrayData = reinterpret_cast(value.data()); + size_t count = value.size(); if (addr.format == MqttTopicAddr::TopicFormat::FLAT) { - const epicsDataType* arrayData = reinterpret_cast(value.data()); - size_t count = value.size(); - std::ostringstream oss; for (size_t i = 0; i < count; ++i) { if (i > 0) oss << ","; @@ -662,8 +673,14 @@ WriteResult MqttDriver::arrayWrite(DeviceVariable& deviceVar, ArraymqttClient.publish(addr.topicName, data.dump()); + status = asynSuccess; } } catch (const std::exception& exc) { @@ -683,17 +700,18 @@ WriteResult MqttDriver::stringWrite(DeviceVariable& deviceVar, Octet const& valu MqttTopicAddr const& addr = static_cast(deviceVar.address()); std::string topicName = addr.topicName; MqttDriver* driver = static_cast(deviceVar).driver; + std::vector stringData(value.maxSize()); + value.writeTo(stringData.data(), stringData.size()); try { if (addr.format == MqttTopicAddr::TopicFormat::FLAT) { - std::vector stringData(value.maxSize()); - if (value.writeTo(stringData.data(), stringData.size())) { - driver->mqttClient.publish(addr.topicName, stringData.data()); - status = asynSuccess; - } - } - else if (addr.format == MqttTopicAddr::TopicFormat::JSON) { - // TODO: implement JSON support for string values - throw std::logic_error("JSON support not implemented"); + driver->mqttClient.publish(addr.topicName, stringData.data()); + status = asynSuccess; + } else if (addr.format == MqttTopicAddr::TopicFormat::JSON) { + // TODO: Compose JSON from driver instead of publishing a plain JSON scalar. + // See https://github.com/epics-modules/mqtt/issues/14 + json j = std::string(stringData.data()); + driver->mqttClient.publish(addr.topicName, j.dump()); + status = asynSuccess; } } catch (const std::exception& exc) { diff --git a/testApp/Db/mqttTest.db b/testApp/Db/mqttTest.db index 82aa914..743d097 100644 --- a/testApp/Db/mqttTest.db +++ b/testApp/Db/mqttTest.db @@ -70,3 +70,102 @@ record(aao, "$(P)$(R)FloatArrayOutput") { field(NELM, "16") field(OUT, "@asyn($(PORT)) FLAT:FLOATARRAY $(TOPIC_ROOT)/floatarray") } + +record(ai, "$(P)$(R)JsonInt32Input") { + field(DESC, "CI Int32 Input") + field(DTYP, "asynInt32") + field(SCAN, "I/O Intr") + field(INP, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/int") +} + +record(ao, "$(P)$(R)JsonInt32Output") { + field(DESC, "CI Int32 Output") + field(DTYP, "asynInt32") + field(OUT, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/int") +} + +record(ai, "$(P)$(R)JsonFloat64Input") { + field(DESC, "CI Float64 Input") + field(DTYP, "asynFloat64") + field(SCAN, "I/O Intr") + field(INP, "@asyn($(PORT)) JSON:FLOAT $(TOPIC_ROOT)/json/float") +} + +record(ao, "$(P)$(R)JsonFloat64Output") { + field(DESC, "CI Float64 Output") + field(DTYP, "asynFloat64") + field(OUT, "@asyn($(PORT)) JSON:FLOAT $(TOPIC_ROOT)/json/float") +} + +record(stringin, "$(P)$(R)JsonStringInput") { + field(DESC, "CI String Input") + field(DTYP, "asynOctetRead") + field(SCAN, "I/O Intr") + field(INP, "@asyn($(PORT)) JSON:STRING $(TOPIC_ROOT)/json/string") +} + +record(stringout, "$(P)$(R)JsonStringOutput") { + field(DESC, "CI String Output") + field(DTYP, "asynOctetWrite") + field(OUT, "@asyn($(PORT)) JSON:STRING $(TOPIC_ROOT)/json/string") +} + +record(aai, "$(P)$(R)JsonIntArrayInput") { + field(DESC, "CI Int Array Input") + field(DTYP, "asynInt32ArrayIn") + field(SCAN, "I/O Intr") + field(FTVL, "LONG") + field(NELM, "16") + field(INP, "@asyn($(PORT)) JSON:INTARRAY $(TOPIC_ROOT)/json/intarray") +} + +record(aao, "$(P)$(R)JsonIntArrayOutput") { + field(DESC, "CI Int Array Output") + field(DTYP, "asynInt32ArrayOut") + field(FTVL, "LONG") + field(NELM, "16") + field(OUT, "@asyn($(PORT)) JSON:INTARRAY $(TOPIC_ROOT)/json/intarray") +} + +record(aai, "$(P)$(R)JsonFloatArrayInput") { + field(DESC, "CI Float Array Input") + field(DTYP, "asynFloat64ArrayIn") + field(SCAN, "I/O Intr") + field(FTVL, "DOUBLE") + field(NELM, "16") + field(INP, "@asyn($(PORT)) JSON:FLOATARRAY $(TOPIC_ROOT)/json/floatarray") +} + +record(aao, "$(P)$(R)JsonFloatArrayOutput") { + field(DESC, "CI Float Array Output") + field(DTYP, "asynFloat64ArrayOut") + field(FTVL, "DOUBLE") + field(NELM, "16") + field(OUT, "@asyn($(PORT)) JSON:FLOATARRAY $(TOPIC_ROOT)/json/floatarray") +} + +record(ai, "$(P)$(R)JsonAltAddressInput") { + field(DESC, "CI Alt Address Input") + field(DTYP, "asynInt32") + field(SCAN, "I/O Intr") + field(INP, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/altAddress") +} + +record(ao, "$(P)$(R)JsonAltAddressOutput") { + field(DESC, "CI Alt Address Output") + field(DTYP, "asynInt32") + field(OUT, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/altAddress") +} + +record(ai, "$(P)$(R)JsonTemplateInput") { + field(DESC, "CI Template Input") + field(DTYP, "asynInt32") + field(SCAN, "I/O Intr") + field(INP, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/writeTemplate") +} + +record(ao, "$(P)$(R)JsonTemplateOutput") { + field(DESC, "CI Template Output") + field(DTYP, "asynInt32") + field(OUT, "@asyn($(PORT)) JSON:INT $(TOPIC_ROOT)/json/writeTemplate") +} diff --git a/tests/README.md b/tests/README.md index b1a9899..3b9a31f 100644 --- a/tests/README.md +++ b/tests/README.md @@ -1,21 +1,34 @@ # MQTT Round-Trip Tests -This folder contains an integration test for the EPICS MQTT driver using `pytest` + `mosquitto` + `p4p`. +This folder contains an integration test for the EPICS MQTT driver using `pytest` + `mosquitto` + +`p4p`. ## What this test does - Starts a local Mosquitto broker in port **18830**. - Starts the test IOC by running `./st.cmd` in `iocBoot/ioctest`, pointed at the local broker. - Uses a unique MQTT client ID and topic root per test session to avoid crosstalk. -- Writes values to output PVs and waits for the corresponding input PVs to receive the same value back. - - This ensures the basic functionality of the driver, including MQTT connectivity, topic subscription, and record - updates, is working correctly. +- Writes values to output PVs and waits for the corresponding input PVs to receive the same value + back. + - This ensures the basic functionality of the driver, including MQTT connectivity, topic + subscription, and record updates, is working correctly. ## Run locally -To run the tests locally, from the root of the repository: +To run the tests locally, install the required dependencies: ```bash +sudo apt-install mosquitto pip3 install p4p pytest +``` + +> By default, `mosquitto` creates a broker service in systemd that runs on `1883`, which is why the +> test uses `18830` to avoid conflicts. You may disable the system service if you want to use the +> default port by running `sudo systemctl disable --now mosquitto`. + +Then, from the root of the repository: + +```bash +make # build the epics module pytest -q tests ``` diff --git a/tests/test_mqtt_roundtrip.py b/tests/test_mqtt_roundtrip.py index 7100e3c..ae8f4d8 100644 --- a/tests/test_mqtt_roundtrip.py +++ b/tests/test_mqtt_roundtrip.py @@ -73,7 +73,14 @@ def _put_and_wait(context, output_pv, input_pv, value, timeout=10.0): ("mqtt:test:StringOutput", "mqtt:test:StringInput", "epicsMQTT-ci"), ("mqtt:test:IntArrayOutput", "mqtt:test:IntArrayInput", [1, 2, 3, 4, 5]), ("mqtt:test:FloatArrayOutput", "mqtt:test:FloatArrayInput", [1.1, 2.2, 3.3, 4.4, 5.5]), - ], + ("mqtt:test:JsonInt32Output", "mqtt:test:JsonInt32Input", 42), + ("mqtt:test:JsonFloat64Output", "mqtt:test:JsonFloat64Input", 3.14159), + ("mqtt:test:JsonStringOutput", "mqtt:test:JsonStringInput", "epicsMQTT-ci"), + ("mqtt:test:JsonIntArrayOutput", "mqtt:test:JsonIntArrayInput", [1, 2, 3, 4, 5]), + ("mqtt:test:JsonFloatArrayOutput", "mqtt:test:JsonFloatArrayInput", [1.1, 2.2, 3.3, 4.4, 5.5]), + ("mqtt:test:JsonAltAddressOutput", "mqtt:test:JsonAltAddressInput", 43), + ("mqtt:test:JsonTemplateOutput", "mqtt:test:JsonTemplateInput", 44), + ], ) def test_round_trip_via_broker(pva_context, output_pv, input_pv, value): _put_and_wait(pva_context, output_pv, input_pv, value)