Metadata-Version: 2.1
Name: forcen_bonappetit_api
Version: 1.0.0
Summary: A Python API designed to streamline integration, control, and testing of Forcen force-torque sensors.
Author: Forcen
Author-email: lexie.zhang@forcen.tech
Requires-Python: >=3.8,<4.0
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Dist: dataclasses_json (>=0.6.7,<0.7.0)
Requires-Dist: forcen_public_utils (==1.0.0)
Requires-Dist: pycryptodome (>=3.10,<4.0)
Requires-Dist: websockets (>=13.1,<14.0)
Description-Content-Type: text/markdown

# 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](#architecture-overview)
- [Installation](#installation)
- [Quickstart Workflow](#quickstart-workflow)
  - [Start the Forcen API Master](#start-the-forcen-api-master)
  - [Connect to the Master using BonAppetitMasterClient](#connect-to-the-master-using-bonappetitmasterclient)
  - [Spawn a server for a device](#spawn-a-server-for-a-device)
  - [Connect to the BonAppetitServer using BonAppetitClient](#connect-to-the-bonappetitserver-using-bonappetitclient)
  - [Send commands / read configuration](#send-commands--read-configuration)
  - [Getting Sensor Data](#getting-sensor-data)
- [Checked Return Types](#checked-return-types)
- [Queues and Callbacks](#queues-and-callbacks)
- [Subscribing to Data, Replies, and Errors](#subscribing-to-data-replies-and-errors)
- [Logging](#logging)
- [ConsoleLogger](#ConsoleLogger)
- [Full API Reference](#full-api-reference)
- [License](#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:
```bash
./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
```python
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
```bash
[DetectedSerialDevice(port='/dev/ttyACM0')]
```
----

### 3. Spawn a server for a device

**For serial devices**
```python
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**

```python
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
```python
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**
```python
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**
```python
# 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

```python
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
```python
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
```python
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:
```python
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:
```python
BonAppetitChecked[T]  # alias for ch.Checked[T, BonAppetitCode]
```
### Consuming Checked Results Safely
To avoid exceptions, always test for success before accessing the value:
```python
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:
```python
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:
```python
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
```python
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](#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](#taring-and-asynchronous-reply-messages)
```python
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.
```python
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**
```bash
/HOME/<username>/.local/state/Forcen/bonappetit_websocket_master
```
**For Windows**
```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:
```python
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:
```python
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:
```python
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:
```python
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.

