Index: CMakeLists.txt =================================================================== diff -u -rcfc0df719cb5033078d0cac45ce0f6243810f2e7 -r6f9f24886c73d92b380e4c3afd82c0b19559d42d --- CMakeLists.txt (.../CMakeLists.txt) (revision cfc0df719cb5033078d0cac45ce0f6243810f2e7) +++ CMakeLists.txt (.../CMakeLists.txt) (revision 6f9f24886c73d92b380e4c3afd82c0b19559d42d) @@ -1,6 +1,7 @@ cmake_minimum_required(VERSION 3.16) -project(CloudSync +project(RtCDT + DESCRIPTION "Real-time Cloud Data Transmission" LANGUAGES CXX ) @@ -40,6 +41,6 @@ add_subdirectory(lib) add_subdirectory(tools) -add_subdirectory(CloudSyncRt) +add_subdirectory(LeahiRt) # dump_cmake_variables() Fisheye: Tag 6f9f24886c73d92b380e4c3afd82c0b19559d42d refers to a dead (removed) revision in file `CloudSyncRt/CMakeLists.txt'. Fisheye: No comparison available. Pass `N' to diff? Fisheye: Tag 6f9f24886c73d92b380e4c3afd82c0b19559d42d refers to a dead (removed) revision in file `CloudSyncRt/CloudSyncRtController.cpp'. Fisheye: No comparison available. Pass `N' to diff? Fisheye: Tag 6f9f24886c73d92b380e4c3afd82c0b19559d42d refers to a dead (removed) revision in file `CloudSyncRt/CloudSyncRtController.h'. Fisheye: No comparison available. Pass `N' to diff? Fisheye: Tag 6f9f24886c73d92b380e4c3afd82c0b19559d42d refers to a dead (removed) revision in file `CloudSyncRt/main.cpp'. Fisheye: No comparison available. Pass `N' to diff? Index: LeahiRt/CMakeLists.txt =================================================================== diff -u -rccc40a7e73e9ee5d2a5fb56f3f2bea4f8294900f -r6f9f24886c73d92b380e4c3afd82c0b19559d42d --- LeahiRt/CMakeLists.txt (.../CMakeLists.txt) (revision ccc40a7e73e9ee5d2a5fb56f3f2bea4f8294900f) +++ LeahiRt/CMakeLists.txt (.../CMakeLists.txt) (revision 6f9f24886c73d92b380e4c3afd82c0b19559d42d) @@ -7,17 +7,16 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_INCLUDE_CURRENT_DIR ON) +set(CMAKE_VERBOSE_MAKEFILE ON) set(CMAKE_AUTOMOC ON) -find_package(QT NAMES Qt6 Qt5 REQUIRED COMPONENTS Core Network) -find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Core Network) +find_package(QT NAMES Qt6 Qt5 REQUIRED COMPONENTS Core WebSockets) +find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Core WebSockets) find_package(Comms HINTS ${CMAKE_CURRENT_SOURCE_DIR}/../lib/Comms REQUIRED) find_package(MsgUtils HINTS ${CMAKE_CURRENT_SOURCE_DIR}/../lib/MsgUtils REQUIRED) -set(LEAHI_MSG_CONF ${CMAKE_CURRENT_SOURCE_DIR}/../data/LeahiUnhandled.conf) - set(INCLUDES LeahiRtController.h ) @@ -27,21 +26,18 @@ main.cpp ) -generate_msg_handling_ini(LEAHI_MSG_CONF ${CMAKE_CURRENT_SOURCE_DIR}/config/LeahiRtMsgHandling.ini generate_msg_handling_ini) - add_executable(${PROJECT_NAME}) -add_dependencies(${PROJECT_NAME} generate_msg_handling_ini) - target_sources(${PROJECT_NAME} PRIVATE ${INCLUDES} ${SRCS}) set_target_properties(${PROJECT_NAME} PROPERTIES RUNTIME_OUTPUT_DIRECTORY "${CMAKE_CURRENT_SOURCE_DIR}/../bin" + ADDITIONAL_CLEAN_FILES "${CMAKE_CURRENT_SOURCE_DIR}/../bin" ) target_link_libraries(${PROJECT_NAME} PRIVATE Comms MsgUtils + Qt${QT_VERSION_MAJOR}::WebSockets Qt${QT_VERSION_MAJOR}::Core - Qt${QT_VERSION_MAJOR}::Network ) Index: LeahiRt/LeahiRtController.cpp =================================================================== diff -u -rccc40a7e73e9ee5d2a5fb56f3f2bea4f8294900f -r6f9f24886c73d92b380e4c3afd82c0b19559d42d --- LeahiRt/LeahiRtController.cpp (.../LeahiRtController.cpp) (revision ccc40a7e73e9ee5d2a5fb56f3f2bea4f8294900f) +++ LeahiRt/LeahiRtController.cpp (.../LeahiRtController.cpp) (revision 6f9f24886c73d92b380e4c3afd82c0b19559d42d) @@ -1,185 +1,52 @@ -/*! - * - * Copyright (c) 2024-2026 Diality Inc. - All Rights Reserved. - * \copyright - * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN - * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. - * - * \file LeahiRtController.cpp - * \author (original) Stephen Quong - * \date (original) 24-May-2026 - * - */ -#include +#include +#include // SQ #include +#include "AlphaOmega.h" // SQ #include "LeahiRtController.h" -#include "LeahiMsgProtoUtils.h" +#include "LeahiMsgDefs.h" -/*! - * \brief LeahiRtController::LeahiRtController - * \details Constructor. Starts the CAN interface and wires the frame→dispatcher→ - * completed-message pipeline. - * \param configPath - path to the settings file - * \param msgHandlingPath - path to the message handling INI - * \param parent - optional QObject parent - */ -LeahiRtController::LeahiRtController(const QString &configPath, const QString &msgHandlingPath, QObject *parent) : +LeahiRtController::LeahiRtController(QObject *parent) : QObject(parent), - _settings(configPath, QSettings::IniFormat), _canInterface(), _canThread(this), - _dispatcher(this) + _clientSocket(QString(), QWebSocketProtocol::VersionLatest, this), + _msgBuilder(this) + // SQ _protoInterface(), + // SQ _protoThread(this) { - loadMsgHandling(msgHandlingPath); - _canInterface.init(_canThread); connect(&_canInterface, &Can::CanInterface::didFrameReceive, this, &LeahiRtController::onFrameReceive); - connect(&_dispatcher, &Can::MessageDispatcher::didActionReceive, this, &LeahiRtController::onMessageReceive); - connect(&_agentInterface, &AgentInterface::didDisconnect, this, &LeahiRtController::onAgentDisconnect); -} -/*! - * \brief LeahiRtController::~LeahiRtController - * \details Destructor. Stops and joins the CAN and Agent worker threads. - */ -LeahiRtController::~LeahiRtController() -{ - _canThread.quit(); - _canThread.wait(); + // SQ _protoInterface.init(_protoThread); + // SQ connect(this, &LeahiRtController::didCanMessageReceive, &_protoInterface, &proto::ProtoInterface::onCanMessageReceive); - _agentThread.quit(); - _agentThread.wait(); + connect(&_clientSocket, &QWebSocket::connected, [&]() { qDebug().noquote() << "Socket connected"; }); } -/*! - * \brief LeahiRtController::connectToAgent - * \details Initialises the AgentInterface using the socket path and reconnect - * interval from the settings file. - */ -void LeahiRtController::connectToAgent() +LeahiRtController::~LeahiRtController() { - const QString socketPath = _settings.value("Socket/LocalSocketName", "/tmp/leahi_rt.sock").toString(); - const int reconnectIntervalMs = _settings.value("Socket/ReconnectIntervalMs", 5000).toInt(); - _agentInterface.init(socketPath, reconnectIntervalMs, _agentThread); + _clientSocket.close(); } -/*! - * \brief LeahiRtController::loadMsgHandling - * \details Parses message handling INI and populates _msgHandling. - * \note Unknown msg action values default to Drop; unknown topic strings default to ClinicalData. - * \param msgHandlingPath - path to the message handling INI file - */ -void LeahiRtController::loadMsgHandling(const QString &msgHandlingPath) +void LeahiRtController::openSocket(const QString host, const unsigned int port) { - static const QHash actionMap = { - { QStringLiteral("send_always"), MsgAction::SendAlways }, - { QStringLiteral("send_delta"), MsgAction::SendDelta }, - { QStringLiteral("drop"), MsgAction::Drop }, - }; - static const QHash topicMap = { - { QStringLiteral("ClinicalData"), AgentMessage::MsgId::ClinicalData }, - { QStringLiteral("Diagnostic"), AgentMessage::MsgId::Diagnostic }, - { QStringLiteral("Ack"), AgentMessage::MsgId::Ack }, - { QStringLiteral("Alarms"), AgentMessage::MsgId::Alarms }, - { QStringLiteral("Audit"), AgentMessage::MsgId::Audit }, - { QStringLiteral("DeviceLogFile"), AgentMessage::MsgId::DeviceLogFile }, - { QStringLiteral("TreatmentLogFile"), AgentMessage::MsgId::TreatmentLogFile }, - { QStringLiteral("CloudSyncLogFile"), AgentMessage::MsgId::CloudSyncLogFile }, - }; - - QSettings msgHandlingIni(msgHandlingPath, QSettings::IniFormat); - if (msgHandlingIni.status() != QSettings::NoError) { - qWarning().noquote() << "LeahiRt: could not read message handling INI" << msgHandlingPath << "— all messages will be dropped"; - return; - } - - int loaded = 0; - for (const QString &group : msgHandlingIni.childGroups()) { - bool ok = false; - const Can::MsgId msgId = static_cast(group.toUInt(&ok, 16)); - if (!ok) { - continue; - } - - msgHandlingIni.beginGroup(group); - const QString actionStr = msgHandlingIni.value(QStringLiteral("action"), QStringLiteral("drop")).toString().trimmed(); - const QString topicStr = msgHandlingIni.value(QStringLiteral("topic")).toString().trimmed(); - msgHandlingIni.endGroup(); - - MsgHandling msgHandling; - msgHandling.action = actionMap.value(actionStr, MsgAction::Drop); - msgHandling.topic = topicMap.value(topicStr, AgentMessage::MsgId::ClinicalData); - - if (!actionMap.contains(actionStr)) { - qWarning().noquote() << QString("LeahiRt: unknown message action \"%1\" for msgId=0x%2 — defaulting to drop") - .arg(actionStr).arg(msgId, 4, 16, QChar('0')); - } - - _msgHandling.insert(msgId, msgHandling); - loaded++; - } - qInfo().noquote() << QString("LeahiRt: loaded message handling %1 (%2 entries)").arg(msgHandlingPath).arg(loaded); + AO_BEGIN_END_CLASS(ao) // SQ + _clientSocket.close(); + _clientSocket.open(QUrl(QString("ws://%1:%2").arg(host).arg(port))); } -/*! - * \brief LeahiRtController::onFrameReceive - * \details Unpacks a CAN frame and feeds it to the dispatcher, which reassembles - * multi-frame messages per CAN id. - * \param frame - the received CAN frame - */ void LeahiRtController::onFrameReceive(const QCanBusFrame frame) { - _dispatcher.onFrameReceive(Can::CanId(frame.frameId()), frame.payload()); -} - -/*! - * \brief LeahiRtController::onMessageReceive - * \details Applies the message handling policy from LeahiRtMsgHandling.ini: drops, - * forwards unconditionally, or forwards only on payload change. Uses the - * section's topic to set the AgentMessage frame msg_id. - * \param msg - the reassembled message - */ -void LeahiRtController::onMessageReceive(const Can::Message &msg) -{ - const auto it = _msgHandling.constFind(msg.msgId); - if (it == _msgHandling.constEnd()) { - qInfo().noquote() << QString("LeahiRt: no action defined for %1 (0x%2), dropping") - .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); - return; + const Can::CanId canId = Can::CanId(frame.frameId()); + Can::Message &msg = _messages[canId]; + // construct the message from the received frame and determine if a complete message has been received + if (_msgBuilder.buildMessage(frame.payload(), msg, canId) && msg.isComplete()) { + // SQ Q_EMIT didCanMessageReceive(QDateTime::currentDateTime(), msg); + // if (_clientSocket.isValid()) { + qDebug().noquote() << QString("Received message with MsgId=0x%1").arg(QString("%1").arg(msg.msgId, 4, 16, QChar('0')).toUpper()); + _clientSocket.sendBinaryMessage(leahi::canMessageToProtobufByteArray(QDateTime::currentDateTime(), QStringLiteral("test_device"), msg)); + // } + msg.clear(); } - - if (it->action == MsgAction::Drop) { - qInfo().noquote() << QString("LeahiRt: dropping message %1 (0x%2)") - .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); - return; - } - - auto &[received, cachedMsg] = _msgCache[msg.msgId]; - if (it->action == MsgAction::SendDelta && received && - cachedMsg.data.chopped(1) == msg.data.chopped(1)) { - qInfo().noquote() << QString("LeahiRt: received message %1 (0x%2) did not changed from previous, dropping") - .arg(leahi::msgIdString(leahi::MsgId(msg.msgId))).arg(msg.msgId, 4, 16, QChar('0')); - return; - } - - const QByteArray payload = leahi::canMessageToProtobufByteArray( - QDateTime::currentDateTime(), - QStringLiteral("test_device"), - msg); - _agentInterface.send(it->topic, _txSequence++, payload); - received = true; - cachedMsg = msg; } - -/*! - * \brief LeahiRtController::onAgentDisconnect - * \details Resets the received flag on all cache entries so that send_delta - * messages are treated as new on the next connection. - */ -void LeahiRtController::onAgentDisconnect() -{ - for (auto &[received, msg] : _msgCache) { - received = false; - } -} Index: LeahiRt/LeahiRtController.h =================================================================== diff -u -rccc40a7e73e9ee5d2a5fb56f3f2bea4f8294900f -r6f9f24886c73d92b380e4c3afd82c0b19559d42d --- LeahiRt/LeahiRtController.h (.../LeahiRtController.h) (revision ccc40a7e73e9ee5d2a5fb56f3f2bea4f8294900f) +++ LeahiRt/LeahiRtController.h (.../LeahiRtController.h) (revision 6f9f24886c73d92b380e4c3afd82c0b19559d42d) @@ -1,116 +1,41 @@ -/*! - * - * Copyright (c) 2024-2026 Diality Inc. - All Rights Reserved. - * \copyright - * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN - * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. - * - * \file LeahiRtController.h - * \author (original) Stephen Quong - * \date (original) 24-May-2026 - * - */ #pragma once #include #include -#include #include #include -#include -#include #include -#include +#include -#include "AgentInterface.h" -#include "AgentMessage.h" #include "CanInterface.h" #include "CanMessage.h" -#include "MessageDispatcher.h" +#include "MessageBuilder.h" +// SQ #include "ProtoInterface.h" using namespace Can; -/*! - * \brief Real-time CAN to cloud controller - * \details Owns the CAN→cloud pipeline. CanInterface receives frames, the - * MessageDispatcher reassembles them per CAN id, and completed messages - * are converted to protobuf and forwarded to the Connectivity Agent - * over the AgentInterface. - */ class LeahiRtController : public QObject { Q_OBJECT public: - explicit LeahiRtController(const QString &configPath, const QString &msgHandlingPath, QObject *parent = nullptr); + explicit LeahiRtController(QObject *parent = nullptr); ~LeahiRtController(); - /*! - * \brief connectToAgent - * \details Initialises the AgentInterface using the socket path and reconnect - * interval from the settings file. - */ - void connectToAgent(); + void openSocket(const QString host, const unsigned int port=80); private: - /*! - * \brief Send message handling policy for a CAN message. - */ - enum class MsgAction { - Drop, ///< Discard the message; do not forward to the Agent. - SendAlways, ///< Forward unconditionally. - SendDelta, ///< Forward only when the payload has changed since the last send. - }; - - /*! - * \brief Message handling entry loaded from message handling INI. - */ - struct MsgHandling { - MsgAction action = MsgAction::Drop; - AgentMessage::MsgId topic = AgentMessage::MsgId::ClinicalData; - }; - - void loadMsgHandling(const QString &msgHandlingPath); - - QSettings _settings; Can::CanInterface _canInterface; QThread _canThread; - Can::MessageDispatcher _dispatcher; - QMap> _msgCache; - QHash _msgHandling; - AgentInterface _agentInterface; - QThread _agentThread; - quint16 _txSequence = 0; + QWebSocket _clientSocket; + Can::MessageBuilder _msgBuilder; + QMap _messages; + // SQ proto::ProtoInterface _protoInterface; + // SQ QThread _protoThread; Q_SIGNALS: - /*! - * \brief didCanMessageReceive - * \details Emitted when a complete CAN message has been reassembled. - * \param timestamp - time the message was completed - * \param msg - the completed message - */ void didCanMessageReceive(const QDateTime timestamp, const Can::Message msg); private Q_SLOTS: - /*! - * \brief onFrameReceive - * \details Unpacks a CAN frame and feeds it to the dispatcher for reassembly. - * \param frame - the received CAN frame - */ void onFrameReceive(const QCanBusFrame frame); - - /*! - * \brief onMessageReceive - * \details Applies the message handling policy: drops, forwards unconditionally, - * or forwards only on payload change, using the INI-configured topic. - * \param msg - the reassembled message - */ - void onMessageReceive(const Can::Message &msg); - - /*! - * \brief onAgentDisconnect - * \details Resets the received flag on all cache entries so that send_delta - * messages are treated as new on the next connection. - */ - void onAgentDisconnect(); }; Index: LeahiRt/main.cpp =================================================================== diff -u -r0359f9239be3218d0b972a54d444e253cc3fc5d9 -r6f9f24886c73d92b380e4c3afd82c0b19559d42d --- LeahiRt/main.cpp (.../main.cpp) (revision 0359f9239be3218d0b972a54d444e253cc3fc5d9) +++ LeahiRt/main.cpp (.../main.cpp) (revision 6f9f24886c73d92b380e4c3afd82c0b19559d42d) @@ -1,71 +1,41 @@ -#include -#include -#include - -#include -#include +#include #include #include -#include -#include +#include +#include #include "LeahiRtController.h" #define log qDebug().noquote() -int main(int argc, char *argv[]) +/*! + * \brief signalHandler + * \details When application terminates it quits gracefully. + * \param[in] sig - The Linux signal causes the termination. + */ +void signalHandler(int sig) { - // Block SIGINT and SIGTERM from normal delivery; redirect them to a fd. - sigset_t mask; - sigemptyset(&mask); - sigaddset(&mask, SIGINT); - sigaddset(&mask, SIGTERM); - sigprocmask(SIG_BLOCK, &mask, nullptr); - - const int sfd = signalfd(-1, &mask, SFD_CLOEXEC); - if (sfd == -1) { - qCritical("Failed to create signalfd"); - return 1; + if (sig == SIGINT) { + log << "Application terminated by SIGINT"; + qApp->quit(); } + else if (sig == SIGTERM) { + log << "Application terminated by SIGTERM"; + qApp->quit(); + } +} +int main(int argc, char *argv[]) +{ + signal(SIGINT , signalHandler); + signal(SIGTERM, signalHandler); + QCoreApplication app(argc, argv); app.setApplicationName("LeahiRt"); app.setApplicationVersion("1.0"); - // Notifier runs on the main thread inside the event loop — safe to call quit(). - QSocketNotifier notifier(sfd, QSocketNotifier::Read); - QObject::connect(¬ifier, &QSocketNotifier::activated, [&](int fd) { - struct signalfd_siginfo info{}; - (void)read(fd, &info, sizeof(info)); - if (info.ssi_signo == SIGINT) { - log << "Application terminated by SIGINT"; - } - else if (info.ssi_signo == SIGTERM) { - log << "Application terminated by SIGTERM"; - } - app.quit(); - }); + LeahiRtController rtController; + rtController.openSocket("localhost", 80); - QCommandLineParser parser; - parser.setApplicationDescription("Leahi Real-time Cloud Data Transmission daemon."); - parser.addHelpOption(); - parser.addVersionOption(); - - QCommandLineOption configOption( - {"c", "config"}, "Path to the configuration INI file.", "config", - QDir(app.applicationDirPath()).filePath("config/LeahiRt.ini") - ); - QCommandLineOption msgHandlingOption( - {"m", "msg_handling"}, "Path to the message handling INI file.", "msg_handling", - QDir(app.applicationDirPath()).filePath("config/LeahiRtMsgHandling.ini") - ); - parser.addOption(configOption); - parser.addOption(msgHandlingOption); - parser.process(app); - - LeahiRtController rtController(parser.value(configOption), - parser.value(msgHandlingOption)); - rtController.connectToAgent(); - return app.exec(); }