GR-Power

1. Overview

GR-Power is the DDS publishing module for robot power status. The runtime fourierpower publishes power and battery status as DDS topics for upper-layer modules to subscribe to.

Element	Value
Runtime name	fourierpower
Domain ID	123
Main DDS topics	hal_battery_monitor_status, hal_bms_error_code

Runtime environment

fourierpower runs as a system user on the robot and starts at boot. Subscription scripts just need to run as a regular user — no root/sudo required. Communicating with the root service on the same host requires the following before creating any DDS participant:

Bash

export FASTDDS_BUILTIN_TRANSPORTS=UDPv4

Otherwise Fast DDS may pick the shared-memory transport on the same host and you will see matched endpoints but no incoming data — or data arriving only intermittently. The Python examples below already embed this setting.

2. DDS Interface

2.1 Published Topics

Topic	Message Type	Description
hal_battery_monitor_status	fourier_msgs/msg/BmsStatusMsg	Main and sub-battery status list.
hal_bms_error_code	fourier_msgs/msg/ErrorCodes	Power / battery fault-code list.

Publishing cadence: hal_battery_monitor_status is published continuously at about 5 Hz; hal_bms_error_code is only published when fault codes exist (no callbacks on a healthy robot is normal), and updates at about 1 Hz while faults are present.

The default QoS (Best Effort + Volatile) is sufficient for subscribers; do not use Reliable QoS — the publisher is Best Effort, a Reliable subscription cannot match and receives no data.

2.2 Subscribed Topics

This module currently has no subscription entry points.

3. Message Definitions

3.1 BmsStatusMsg

Idl

struct BmsStatus {
    double voltage;
    double current;
    double temperature;
    uint8  level;
    uint8  level_cap;
    double mos_temperature;
    boolean failure;
    boolean battery_comm;
    boolean output_mos_open;
    boolean input_mos_open;
};

struct BmsStatusMsg {
    std_msgs::msg::Header header;
    sequence<BmsStatus> bms_state_list;
};

bms_state_list[0] is the main battery and bms_state_list[1] is the sub battery.

GR-W field notes

GR-W carries two batteries, so bms_state_list has length 2. The current battery source fills only level, voltage, current, and battery_comm (battery_comm is always true); temperature, failure, mos_temperature, and the MOS status fields keep their default values and must not be used as battery-health criteria on GR-W.

3.2 ErrorCodes

Idl

struct ErrorCode {
    uint32 high32;
    uint32 low32;
};

struct ErrorCodes {
    std_msgs::msg::Header header;
    uint8 source; // 4 = power / battery
    sequence<ErrorCode> error_codes;
};

GR-Power encodes high32 as follows:

Text

level     = high32 & 0xF
code      = (high32 >> 12) & 0xFFFF
type_flag = (high32 >> 28) & 0xF

type_flag	Meaning
0	BMS1 battery fault
1	BMS2 battery fault

caution

type_flag values beyond the table above exist (e.g. 2); their meaning depends on the actual power-module firmware. Only the values listed above are defined.

4. Basic Examples

4.1 Python - Subscribe to Battery Status

Python

import os
import signal
import time

os.environ.setdefault("FASTDDS_BUILTIN_TRANSPORTS", "UDPv4")  # required on the robot

from fourierdds_py import DDSInterface, SubscriberQosProfile
import fourier_msgs.msg.BmsState as BmsState

dds = DDSInterface(domain_id=123)


def seq_items(seq):
    size = seq.size() if hasattr(seq, "size") else len(seq)
    return [seq[i] for i in range(size)]


def on_bms(msg):
    states = seq_items(msg.bms_state_list())
    if not states:
        print("[bms] empty state list")
        return

    for index, state in enumerate(states):
        name = "main" if index == 0 else f"sub{index}"
        print(
            f"[bms:{name}] "
            f"level={state.level()}% "
            f"voltage={state.voltage():.3f}V "
            f"current={state.current():.3f}A "
            f"temperature={state.temperature():.1f}C "
            f"failure={state.failure()} "
            f"comm={state.battery_comm()}"
        )


running = {"value": True}


def _stop(signum, frame):
    running["value"] = False
    print(f"\nsignal {signum} received, stopping ...", flush=True)


# fastdds installs a native signal handler that swallows Ctrl+C;
# override it explicitly so the script can exit normally
signal.signal(signal.SIGINT, _stop)
signal.signal(signal.SIGTERM, _stop)

dds.create_subscription(
    BmsState.BmsStatusMsgPubSubType,
    "hal_battery_monitor_status",
    on_bms,
    qos_profile=SubscriberQosProfile.default(),
)

try:
    while running["value"]:
        time.sleep(0.1)
finally:
    dds.close()   # release DDS resources; skipping this may error at exit

Run the example (Ctrl+C to stop):

Bash

python3 watch_grpower_status.py

The subscription script runs as a regular user. If you must run it as root, keep the FASTDDS_BUILTIN_TRANSPORTS=UDPv4 environment variable:

Bash

sudo FASTDDS_BUILTIN_TRANSPORTS=UDPv4 python3 -u watch_grpower_status.py

4.2 Python - Subscribe to Fault Codes

Python

import os
import signal
import time

os.environ.setdefault("FASTDDS_BUILTIN_TRANSPORTS", "UDPv4")  # required on the robot

from fourierdds_py import DDSInterface, SubscriberQosProfile
import fourier_msgs.msg.ErrorCodes as ErrorCodes

dds = DDSInterface(domain_id=123)


def seq_items(seq):
    size = seq.size() if hasattr(seq, "size") else len(seq)
    return [seq[i] for i in range(size)]


def decode_high32(high32):
    return {
        "level": high32 & 0xF,
        "code": (high32 >> 12) & 0xFFFF,
        "type_flag": (high32 >> 28) & 0xF,
    }


def on_error(msg):
    print(f"[power_error] source={msg.source()} count={len(seq_items(msg.error_codes()))}")
    for index, item in enumerate(seq_items(msg.error_codes())):
        decoded = decode_high32(item.high32())
        print(
            f"  [{index}] "
            f"type_flag={decoded['type_flag']} "
            f"level={decoded['level']} "
            f"code=0x{decoded['code']:04X} "
            f"high32=0x{item.high32():08X} "
            f"low32=0x{item.low32():08X}"
        )


running = {"value": True}


def _stop(signum, frame):
    running["value"] = False
    print(f"\nsignal {signum} received, stopping ...", flush=True)


# fastdds installs a native signal handler that swallows Ctrl+C;
# override it explicitly so the script can exit normally
signal.signal(signal.SIGINT, _stop)
signal.signal(signal.SIGTERM, _stop)

dds.create_subscription(
    ErrorCodes.ErrorCodesPubSubType,
    "hal_bms_error_code",
    on_error,
    qos_profile=SubscriberQosProfile.default(),
)

try:
    while running["value"]:
        time.sleep(0.1)
finally:
    dds.close()

A healthy robot has no fault codes, so this script produces no callback output at all — that is expected. When fault codes exist they arrive at about 1 Hz.

4.3 C++ - Subscribe to Battery Status

C++

#include <BmsState/BmsStatePubSubTypes.hpp>
#include "fourier_dds/dds_node.hpp"
#include "fourier_dds/dds_subscription.hpp"

#include <chrono>
#include <iostream>
#include <memory>
#include <thread>

int main() {
    auto node = std::make_shared<fourier_dds::DdsNode>(123);

    auto bms_sub = std::make_shared<
        fourier_dds::DdsSubscription<fourier_msgs::msg::BmsStatusMsgPubSubType>>(
        node,
        "hal_battery_monitor_status",
        [](const fourier_msgs::msg::BmsStatusMsg& msg) {
            const auto& states = msg.bms_state_list();
            for (size_t i = 0; i < states.size(); ++i) {
                const auto& state = states[i];
                std::cout << "[bms:" << i << "] "
                          << "level=" << static_cast<int>(state.level()) << "% "
                          << "voltage=" << state.voltage() << "V "
                          << "current=" << state.current() << "A "
                          << "failure=" << state.failure()
                          << std::endl;
            }
        });

    while (true) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

Compile (note that fourierdds_core must be linked; otherwise the build fails with undefined reference to fourier_dds::...):

Bash

g++ -std=c++17 watch_grpower_status.cpp -o watch_grpower_status \
  -I/opt/fftai/include/fourier_dds \
  -I/opt/fftai/fourier_dds_msgs/include/fourier_dds_msgs \
  -L/opt/fftai/fourier_dds_msgs/lib \
  -Wl,-rpath,/opt/fftai/fourier_dds_msgs/lib \
  -lBmsState \
  -lfourierdds_core \
  -lfastdds \
  -lfastcdr \
  -pthread

Run the example (regular user is sufficient):

Bash

FASTDDS_BUILTIN_TRANSPORTS=UDPv4 ./watch_grpower_status

5. DDS Integration Workflow

5.1 Observing Battery Status

1. Confirm fourierpower is running on the robot (it starts at boot).
2. Run the status subscription script:
   Bash

   python3 watch_grpower_status.py

3. Confirm the subscriber receives hal_battery_monitor_status: GR-W carries two batteries, so bms_state_list should contain 2 elements with battery_comm=true and plausible voltage/level values (negative current means discharging).

5.2 Observing Fault Codes

1. Run the fault-code subscription script from section 4.2:
   Bash

   python3 watch_grpower_errors.py

2. No output on a healthy robot is expected; when fault codes exist you will receive hal_bms_error_code messages with source=4.
3. Use type_flag, level, code, and low32 to interpret the fault source and the specific fault bits.

6. Troubleshooting

No Battery Status Received

Confirm fourierpower is running (ps -C fourierpower), the DDS domain is 123, the subscriber uses the topic name and message type from the table, and FASTDDS_BUILTIN_TRANSPORTS=UDPv4 is set. Without it, endpoints may match but deliver no data — or deliver it only intermittently; sudo is not the right fix for this.

No Fault Codes Received

Confirm the subscriber uses hal_bms_error_code and fourier_msgs/msg/ErrorCodes, and that the DDS domain matches fourierpower. ErrorCodes is published only when GR-Power has fault codes: no callbacks on a healthy robot is normal, not a fault.

ErrorCodes.source=4

GR-Power sets ErrorCodes.source to 4, meaning the fault source is the power / battery module.