forcen_bonappetit_api 1.0.0

A Python API designed to streamline integration, control, and testing of Forcen force-torque sensors.

forcen_bonappetit_api

forcen_bonappetit_api is the Python communication layer for interacting with Forcen devices through the Forcen API Master and the BonAppetit WebSocket/UDP protocol stack.

It provides:

• A BonAppetitMasterClient for communicating with the Forcen API Master.
• A ProtocolDriverInterface defining the full device-level API.
• A BonAppetitClient for issuing commands and receiving realtime sensor data.

---

Table of Contents

• Architecture Overview
• Installation
• Quickstart Workflow
  • Start the Forcen API Master
  • Connect to the Master using BonAppetitMasterClient
  • Spawn a server for a device
  • Connect to the BonAppetitServer using BonAppetitClient
  • Send commands / read configuration
  • Getting Sensor Data
• Checked Return Types
• Queues and Callbacks
• Subscribing to Data, Replies, and Errors
• Logging
• ConsoleLogger
• Full API Reference
• License

---

Architecture Overview

Forcen API uses a multi-stage communication setup involving:

• Forcen API Master — detects devices and manage servers

• BonAppetitMasterClient — communicates with the Forcen API Master

• BonAppetitServer — one per device, communicate with the physical Forcen sensors

• BonAppetitClient — connects your application to the device server

• Forcen Physical Device — sensor connected over Serial/TCP/UDP

             ┌──────────────────────────┐
             │     User Application     │
             │  (your Python program)   │
             ├──────────────────────────┤
             │  BonAppetitMasterClient  │───▶ Talks to Forcen API Master
             ├──────────────────────────┤
             │  BonAppetitClient        │───▶ Talks to BonAppetit server
             └──────────────▲───────────┘
                            │
                 Python WebSocket APIs
                            │
       ┌────────────────────▼────────────────────┐
       │          Forcen API Master              │
       │  - Detects connected devices            │
       │  - Spawns/manages Bon Appetit servers   │
       └────────────────────▲────────────────────┘
                            │
                   spawning servers
                            │
   ┌────────────────────────▼───────────────────────┐
   │                 BonAppetitServer               │
   │     - One server per Forcen device             │
   │     - Exposes:                                 │
   │         - commands/replies                     │
   │         - sensor data                          │
   │         - error messages                       │
   └────────────────────────▲───────────────────────┘
                            │
                    TCP/UDP/Serial
                            │
                            ▼
                 Forcen Physical Device
  

Installation

Quickstart Workflow

1. Start the Forcen API Master

Get the Forcen API Master executable and launch the Master service on your system by running the executable:

./forcen_api_master_v1_0_0 <IP_ADDRESS> <COMMAND_SRV_PORT> <ERROR_PUB_PORT>

For local connection, <IP_ADDRESS> can be localhost. <COMMAND_SRV_PORT> <ERROR_PUB_PORT> could be automatically assigned if left as 0. For more info, refer to the README.md for Forcen API Master.

---

2. Connect to the Master using BonAppetitMasterClient

from forcen_bonappetit_api.bonappetit_master_client import BonAppetitMasterClient

with BonAppetitMasterClient("localhost", 65535, 65536) as masterclient:
    detected = masterclient.get_detected_devices(rerun_detection=True)
    if not detected.has_error():
	    print("Detected devices:", detected.value()) # list of DetectedSerialDevice

sample output

[DetectedSerialDevice(port='/dev/ttyACM0')]

---

3. Spawn a server for a device

For serial devices

device = detected[0]
baudrate = 115200 # default baudrate

serial = master.spawn_server(
    server_ip_address="localhost", # "localhost" for local connection
    transport_type="serial",
    device_info={
		"baudrate": 115200, #default baudrate
		"port": device.port,
	},
    enable_sensor_data_logging=False,
    enable_comms_logging=True,
)

server_uuid = resp.value()
server_info = master.wait_for_partially_initialized_server(server_uuid, timeout=10).value()
print("Server ready:", server_info)

For TCP/UPD devices

device = detected[0]

resp = master.spawn_server(
    server_ip_address="localhost",
    transport_type="tcp", # or "udp" for UDP devices
    device_info={
        "ip_address": "192.168.0.10",  # example
        "port": 2000,
    },
    enable_sensor_data_logging=False,
    enable_comms_logging=False,
)

server_uuid = resp.value()
server_info = master.wait_for_partially_initialized_server(server_uuid, timeout=10).value()
print("Server ready:", server_info)

spawn_server returns the UUID of the server that the Master has begun creating, but this does not mean the server is fully initialized. The Master starts the server asynchronously so it can continue handling requests from multiple clients without blocking.

To obtain the information required to connect a BonAppetitClient to the new server, call wait_for_partially_initialized_server with the UUID returned by spawn_server. This helper repeatedly checks get_active_servers() until the Master reports a fully initialized ServerInfo object for that UUID.
While initialization is still in progress, the Master returns a PartialServerInfo, indicating that the server exists but is not yet ready for use.

---

4. Connect to the BonAppetitServer using BonAppetitClient

from forcen_bonappetit_api.bonappetit_websocket_client import BonAppetitClient

with BonAppetitClient(
            server_ip_address=server_info.ip_address,
            server_command_port=server_info.commands_srv_port,
            server_data_pub_port=server_info.sensor_data_pub_port,
            server_error_pub_port=server_info.errors_pub_port,
            server_replies_pub_port=server_info.replies_pub_port,
        ) as client:

    fw = client.get_firmware_version().value()
    print("Firmware version:", fw)

---

5. Send commands / read configuration

Here are some samples of the most often used functions

Start/Stop Device From Streaming Sensor Data

from forcen_bonappetit_api.protocol.protocol_data_types import DeviceMode

# start streaming data
client.set_mode(DeviceMode.RUNNING)

# stop streaming data
client.set_mode(DeviceMode.SLEEP)

# verify mode
mode_rly = client.get_mode()
if mode_rly.has_value():
	print(f"The sensor is in {mode_rly.value()} mode.")

Get/Set Sensor Data Rate

# set data rate
client.set_data_rate(data_rate_hz=100) # 100 sps

# get data rate
data_rate_rly = client.get_data_rate()
if data_rate_rly.has_value():
	print(f"The sensor is set to {data_rate_rly.value()} sps.")

Taring and Asynchronous Reply Messages

When a device enters taring mode, the firmware performs an internal procedure. Depending on the device model and firmware version, the taring process may generate multiple asynchronous reply messages indicating progress, intermediate status, or completion.

These reply packets are not part of the sensor data stream. To receive them, the application must subscribe to reply packets before entering taring mode.

The example below demonstrates:

• Subscribing to reply messages
• Setting the device into taring mode
• Handling asynchronous replies as they arrive

from forcen_bonappetit_api.protocol.protocol_data_types import DeviceMode

# Subscribe to reply packets
reply_sub = client.subscribe_to_reply_packets(max_buffer_size=64)
if reply_sub.has_error():
    raise RuntimeError("Failed to subscribe to reply packets: " + str(reply_sub.error()))

reply_queue = reply_sub.value()

# Enter taring mode
set_res = client.set_mode(DeviceMode.CALIBRATION)
if set_res.has_error():
    raise RuntimeError("Failed to enter taring mode: " + str(set_res.error()))

print("Taring started... waiting for asynchronous replies")

# Read reply packets until completion or timeout
while True:
    replies = reply_queue.sync_get_all(timeout=1.0)
    if not replies:
        print("No more replies (timeout).")
        break

    for pkt in replies:
        print(f"[TARE REPLY] {pkt.timestamp}: {pkt.data}")

---

6. Getting Sensor Data

There are two ways of getting sensor data: subscribe_to_sensor_packets and attach_sensor_data_callback

subscribe_to_sensor_packets(max_buffer_size, realtime_stream=False)

Creates a new subscription to the device’s sensor data stream and returns a queue (RTDQueueT) that will automatically be filled with incoming realtime data packets.

When this function is called, a WebSocket subscription is opened between your application and the BonAppetit Server. Incoming sensor packets are delivered to an internal callback, which pushes them into the returned queue.

Realtime vs Batched Streams

• realtime_stream=False (default) Sensor packets are grouped into batches before being delivered to the queue. This mode reduces network traffic and server load and should be used for most applications.

• realtime_stream=True Each packet is pushed individually as soon as it arrives. Use this mode only when strict low-latency streaming is required, since opening many realtime streams can reduce throughput and increase server load.

Arguments

Name	Type	Description
max_buffer_size	int	Maximum number of packets to store in the queue. Old packets may be dropped once full.
realtime_stream	bool	Whether to receive data packet-by-packet (True) or in batches (False).

Returns

A BonAppetitChecked[RTDQueueT] containing a queue that the user can get() packets from.

Example

Subscribe to sensor packets and process them in a loop

sub = client.subscribe_to_sensor_packets(
    max_buffer_size=1024,
    realtime_stream=False,
)

if sub.has_error():
    raise RuntimeError("Failed to subscribe: " + str(sub.error()))

queue = sub.value()

print("Waiting for sensor packets...")

while True:
    queue_data = queue.sync_get_all(timeout=1.0)
    for packet in queue_data:
	    print(f"Received packet at {packet.timestamp}: {packet.data}")

---

attach_sensor_data_callback(callback, realtime_stream=False)

Registers a callback that is invoked automatically whenever the device sends sensor data. Unlike subscribe_to_sensor_packets(), which returns a queue, this method pushes data directly into your callback.

When the function is called:

• A separate WebSocket (batched mode) or UDP (realtime mode) subscription is created.
• The callback is executed whenever new packets arrive.
• Multiple callbacks may be attached, and each will receive incoming data.

Realtime vs Batched Callback Behavior

• realtime_stream=False (default) The server batches packets and delivers them to the callback in groups. This reduces network overhead and should be the default for most use cases.

• realtime_stream=True The callback is called as soon as each packet arrives, minimizing latency. Using many realtime subscriptions can reduce network throughput and increase server load, so use sparingly.

Arguments

Name	Type	Description
callback	Callable[[List[RTDPacketStamped]], None]	Function invoked when sensor data is received. It always receives a list of packets.
realtime_stream	bool	Whether to receive packets immediately (True) or in batches (False).

Returns

A BonAppetitChecked[None]. If .is_ok() returns True, the subscription is active.

Example

def on_sensor_data(packets):
    for pkt in packets:
        print(f"[{pkt.timestamp}] Values: {pkt.data}")

res = client.attach_sensor_data_callback(on_sensor_data, realtime_stream=False)
if res.has_error():
    raise RuntimeError("Failed to attach callback: " + str(res.error()))

print("Callback attached. Waiting for sensor data...")

Compare subscribe_to_sensor_packets() and attach_sensor_data_callback()

Use attach_sensor_data_callback() when:

• You want event-driven processing
• You do not want a queue
• You want to handle data in your own thread or async loop
• You want immediate processing of packets

Use subscribe_to_sensor_packets() when:

• You want pull-based access to data
• You want to buffer data safely
• You prefer to call queue.sync_get_all() manually

Checked Return Types

The Forcen APIs use a checked result type instead of exceptions for most operations. The utility lives in forcen_public_utils.checked and is typically imported as:

import forcen_public_utils.checked as ch

A ch.Checked[ValueT, ErrorT] contains either:

• a value of type ValueT, or
• an error of type ErrorT plus a human-readable message,

but never both at the same time.

most driver methods in BonAppetitClient return something like:

BonAppetitChecked[T]  # alias for ch.Checked[T, BonAppetitCode]

Consuming Checked Results Safely

To avoid exceptions, always test for success before accessing the value:

res = client.get_serial_number()
if res.is_ok():
    temp = res.value() print("Temperature:", temp) else:
    err = res.error() print("Error:", err.code, err.msg)`

Helper methods:

• has_value() / has_error() – indicate which variant is present
• value() – returns the stored value or raises CheckedBadAccess if there is an error
• error() – returns the stored error or raises CheckedBadAccess if there is a value
• full_error() – returns a FullError containing error and message

Queues and Callbacks

The api uses typed queues and callbacks for asynchronous data:

from forcen_public_utils.sync_queue import SyncQueue
from forcen_bonappetit_api.driver.protocol_driver_interface import (
    RTDPacketStamped,
    ReplyPacketStamped,
    PacketStamped,
)
import forcen_public_utils.checked as ch from forcen_bonappetit_api.common.error_types
import BonAppetitCode

RTDQueueT   = SyncQueue[RTDPacketStamped]
ReplyQueueT = SyncQueue[ReplyPacketStamped]
MixedQueueT = SyncQueue[PacketStamped]
ErrorQueueT = SyncQueue[ch.FullError[BonAppetitCode]]

Callbacks:

RTDCallbackT   = Callable[[List[RTDPacketStamped]], None]
ReplyCallbackT = Callable[[ReplyPacketStamped], None]
MixedCallbackT = Callable[[PacketStamped], None]
ErrorCallbackT = Callable[[ch.FullError[BonAppetitCode]], None]

Subscribing to Data, Replies, and Errors

The BonAppetitClient provides several subscription mechanisms for receiving asynchronous information from the device server. Each subscription returns a typed queue, or triggers user-defined callbacks, depending on how the stream is configured.

There are three types of subscribable streams:

1. Sensor Data (RTD)

Sensor packets from the device's realtime data stream. Use this for reading force/torque measurements

client.subscribe_to_sensor_packets(...)
client.attach_sensor_data_callback(...)

Sensor data subscriptions support batched WebSocket mode or low-latency UDP mode, depending on the realtime_stream option. Explanation in details can be found in Getting Sensor Data.

---

2. Reply Packets

Reply packets are messages sent by the device in response to commands. These include confirmation replies and multi-step command outputs, e.g. taring progress. Replies are not part of the sensor data stream; they are separate logical messages generated by device firmware. To get example on usage, refer to Taring and Asynchronous Reply Messages

client.subscribe_to_reply_packets(...)
client.attach_reply_callback(...)

---

3. Error Notifications

The server may publish transport errors, device warnings, runtime faults, or communication-layer issues. These appear as typed FullError[BonAppetitCode] objects.

client.subscribe_to_errors(...)
client.attach_error_callback(...)

Use this to:

• Monitor device health
• Report warnings or hardware faults
• Track network/transport errors
• Surface runtime issues in UI dashboards

Error channels operate independently of reply and sensor data streams.

Logging

The Forcen API Master, BonAppetit servers, and BonAppetit client all produce logs to assist with debugging, device bring-up, and system validation. Logging is enabled automatically at the Master level and can be configured when spawning device servers.

---

Master Logging

When running the Forcen API Master executable:

./forcen_api_master_v1_0_0 <IP_ADDRESS> <COMMAND_SRV_PORT> <ERROR_PUB_PORT>

the Master automatically enables its internal logging subsystem. These logs begin as soon as the Master starts, before any device or client connections occur. It includes everything that gets printed on the terminal:

• Device detection events
• Server lifecycle events (spawn, connect, disconnect)
• Transport activity
• Error and warning messages

Automatic Log Cleanup

The Master automatically deletes log files older than 14 days, preventing unbounded disk growth. Log retention requires no user configuration.

Log File Location

For Ubuntu

/HOME/<username>/.local/state/Forcen/bonappetit_websocket_master

For Windows

C:\Users\<username>\AppData\Local\Forcen\console_logger

---

Server Logging

When the Master spawns a BonAppetit server for a device, logging can be enabled per-server using:

resp = masterclient.spawn_server(
    server_ip_address="localhost",
    transport_type="serial",
    device_info={
        "baudrate": 115200,
        "port": device.port,
    },
    enable_sensor_data_logging=True,
    enable_comms_logging=True,
)

server_uuid = resp.value()
server_info = master.wait_for_partially_initialized_server(server_uuid, timeout=10).value()
print("Server ready:", server_info)

Two independent logging channels can be enabled:

Option	Description
enable_sensor_data_logging	Logs raw sensor data packets processed by the server. Useful for calibration debugging or data validation.
enable_comms_logging	Logs command packets, replies, protocol events, and transport-level details. Helpful for diagnosing configuration issues.

These logs appear under the same Forcen directory used by the Master.

For production systems, it is common to disable both logging flags, or only enable enable_comms_logging = True selectively during debugging.

Client Logging

BonAppetitClient uses ConsoleLogger to emit client-side messages such as:

• Connection setup or failure
• UDP/WebSocket subscriber startup
• Forwarded device errors
• Callback or subscription behavior

By default, logs are printed to stdout, but applications may reconfigure ConsoleLogger to use different sinks.

ConsoleLogger

ConsoleLogger is a public logging utility included in forcen_public_utils.
It provides unified, thread-safe logging for any Python application using the Forcen API.
The logger outputs messages to both:

1. The console, and
2. A log file, with rotation and configurable settings.

Each log line contains:

• Timestamp
• Process ID or thread ID
• Log level
• File and line number
• Human-readable message

This format makes ConsoleLogger suitable for debugging device communication, multi-threaded applications, and long-running data collection systems.

Using ConsoleLogger

Since the ConsoleLogger is a singleton, you can only construct it once.

Retrieve the active logger with:

from forcen_public_utils.loggers.console_logger import ConsoleLogger

logger = ConsoleLogger.get()
logger.info("System initialized")
logger.error("Something went wrong")

Convenience methods exist for all logging levels:

logger.trace("trace message")
logger.debug("debug details")
logger.info("starting...")
logger.success("operation completed")
logger.warning("unexpected condition")
logger.error("operation failed")
logger.critical("fatal error")

For more detailed example, you can refer to /forcen_public_utils/scripts/console_logger_demo.py

Full API Reference

The Forcen BonAppetit API exposes a large number of device commands and configuration functions. Descriptions of all device features include data streaming, calibration, ADC configuration, CAN/UART networking, user script upload, firmware queries, error handling, and more.

To explore the full API interactively:

from forcen_bonappetit_api.bonappetit_websocket_client import BonAppetitClient

help(BonAppetitClient)

License

Copyright (c) 2019 Forcen Inc. All Rights Reserved. Restriction on Use, Publication or Disclosure of Proprietary Information. Unauthorized copying, distribution, or use is strictly prohibited.