The BIR HTTP interface allows you to interact with the module from external applications running on your PC or other devices in the same network.
You can read sensor values, write commands, and use the module as a real-time control backend for your custom logic implemented in Python, JavaScript, or any language that supports HTTP requests.
Neural Network example
This demo shows how you can use the BIR HTTP interface as a real-time control backend while running
higher-level logic on your PC. The script first trains a tiny neural network on synthetic temperature samples
(fast, offline, reproducible), and then switches into a live control loop where it continuously reads three
temperature channels from the BIR module and writes a servo angle back to the device.
What the demo does
- Train: Generate synthetic
TMP_6/TMP_7/TMP_8 triplets and train a tiny NN (offline).
- Teacher: Discrete targets
0° / 90° / 180° from a strict “winner” rule (ties skipped).
- Read: Fetch live temperatures in one request via
/peer_multi.
- Predict: NN outputs a continuous servo angle.
- Write: Send the angle to
SRV_4 via /command.
Training phase (synthetic “teacher”)
Instead of collecting real data first, the script generates synthetic temperature triplets inside a configurable range.
A simple “teacher” rule assigns only three target angles: 0°, 90°, or 180°
depending on which temperature is strictly the highest. If there is a tie for the maximum value, that sample is skipped.
This creates a clean supervised dataset and avoids ambiguous labels.
Live control loop (BIR as the actuator endpoint)
After training, the script enters an infinite loop:
it fetches the live temperatures, predicts a continuous servo angle, clamps it to 0..180,
and writes it back to the BIR module. The console output highlights the “winner” temperature channel and shows
the predicted angle versus the teacher reference (when there is no tie), so you can see how the NN behaves in real time.
WATCH THE VIDEO EXAMPLE
# =============================================================================
# BIR NN DEMO (REGRESSION, 3 inputs -> 1 continuous output 0..180)
#
# Goal:
# - Inputs: TMP_6, TMP_7, TMP_8 (temperatures)
# - "Teacher" outputs ONLY 3 discrete targets: 0 / 90 / 180 degrees
# * if T6 is highest -> 180
# * if T7 is highest -> 90
# * if T8 is highest -> 0
# - If there is a tie for maximum: teacher returns None (skip training sample)
#
# Key idea:
# NN input uses temps relative to MIN (offset-invariant).
# features = [(T6-min)/DIFF_SCALE, (T7-min)/DIFF_SCALE, (T8-min)/DIFF_SCALE]
# =============================================================================
import time
import random
from typing import List, Tuple, Optional
import requests
import sys
# -----------------------------
# Configuration (edit here)
# -----------------------------
# BIR
BIR_IP = "192.168.0.202" # IP address of your BIR (edit this)
TIMEOUT_S = 0.5 # HTTP request timeout (seconds) - adjust if you have network issues
PERIOD_S = 1.0 # control loop period (seconds) - adjust as needed (e.g. 0.5s or 2s)
TMP_IDS = ["TMP_6", "TMP_7", "TMP_8"] # temperature sensor IDs (edit if your sensors have different IDs)
SERVO_ID = "SRV_4" # servo ID to control (edit if you want to use a different servo)
BASE = f"http://{BIR_IP}" # base URL for BIR API
# Synthetic training temperature ranges
TRAIN_T_MIN = 20.0 # wide range min
TRAIN_T_MAX = 30.0 # wide range max
TRAIN_Tmicro_MIN = 24.0 # micro range min (more near-ties)
TRAIN_Tmicro_MAX = 26.0 # micro range max
TRAIN_RATIO = 0.1 # probability of drawing a sample from micro range
TEMP_STEP = 0.25 # quantization step (matches sensor resolution)
# NN + training
SEED = 1234 # random seed for reproducibility (tune this or set to None for random seed)
LR = 0.001 # learning rate (tune this)
"""
# Lite model
HIDDEN = 16 # number of hidden neurons (tune this)
EPOCHS = 10 # number of training epochs (tune this)
STEPS_PER_EPOCH = 20_000 # number of training steps per epoch (tune this)
"""
# Bigger model
HIDDEN = 32
EPOCHS = 30
STEPS_PER_EPOCH = 100_000
#"""
# Feature scaling for temperature differences
DIFF_SCALE = 10.0 # scale factor for input features (tune this, e.g. to match typical temp differences)
# Console progress update
PROGRESS_EVERY = 5000 # update live line each N steps
# -----------------------------
# Terminal colors (ANSI)
# -----------------------------
C_RESET = "\033[0m"
C_GREEN = "\033[32m"
C_ORANGE = "\033[38;5;208m"
C_TMP_WIN = "\033[96m"
C_TMP_OTHER = "\033[90m"
# =============================================================================
# BIR API helpers
# =============================================================================
class BirApiError(Exception):
pass
def bir_peer_multi(node_base: str, ids: List[str], timeout_s: float = TIMEOUT_S) -> List[float]:
"""
GET /peer_multi?ids=TMP_6,TMP_7,TMP_8
Response body: "v1;v2;v3"
"""
r = requests.get(f"{node_base}/peer_multi", params={"ids": ",".join(ids)}, timeout=timeout_s)
if r.status_code == 503:
raise BirApiError("NOT_READY (503)")
if r.status_code in (400, 404):
raise BirApiError(f"{r.status_code} {r.text.strip()}")
r.raise_for_status()
parts = r.text.strip().split(";")
if len(parts) != len(ids):
raise BirApiError(f"Parse mismatch: got '{r.text.strip()}' for ids={ids}")
# Tolerate decimal comma
return [float(p.strip().replace(",", ".")) for p in parts]
def bir_set_value(node_base: str, sensor_id: str, value: float, timeout_s: float = TIMEOUT_S) -> None:
"""
POST /command
Body: cmd:value=ID,value#
"""
body = f"cmd:value={sensor_id},{value}#"
r = requests.post(f"{node_base}/command", data=body, timeout=timeout_s)
r.raise_for_status()
# =============================================================================
# Small utility helpers
# =============================================================================
def clamp(x: float, lo: float, hi: float) -> float:
return lo if x < lo else hi if x > hi else x
def lrelu(x: float, a: float = 0.05) -> float:
return x if x >= 0.0 else a * x
def dlrelu(x: float, a: float = 0.05) -> float:
return 1.0 if x >= 0.0 else a
def quantize_temp(t: float, step: float = TEMP_STEP) -> float:
"""Round to nearest multiple of TEMP_STEP (e.g. 0.25°C)."""
return round(t / step) * step
def live_line(s: str) -> None:
"""Overwrite current console line (ANSI)."""
sys.stdout.write("\r\033[2K" + s)
sys.stdout.flush()
# =============================================================================
# Teacher (stateless, tie -> None)
# =============================================================================
def teacher_target_deg(t6: float, t7: float, t8: float) -> Optional[float]:
"""
Stateless teacher:
- if there is a UNIQUE winner (strictly highest), returns 180/90/0
- if there is a tie for highest, returns None (skip training)
"""
m = max(t6, t7, t8)
winners = (t6 == m) + (t7 == m) + (t8 == m)
if winners != 1:
return None
if t6 == m:
return 180.0
elif t7 == m:
return 90.0
else:
return 0.0
def color_tmp_triplet(t6: float, t7: float, t8: float) -> str:
ref = teacher_target_deg(t6, t7, t8) # 180/90/0 or None
vals = [t6, t7, t8]
if ref is None:
return ", ".join(f"{C_TMP_OTHER}{v:5.2f}{C_RESET}" for v in vals)
if ref == 180.0:
win = 0
elif ref == 90.0:
win = 1
else:
win = 2
out = []
for i, v in enumerate(vals):
c = C_TMP_WIN if i == win else C_TMP_OTHER
out.append(f"{c}{v:5.2f}{C_RESET}")
return ", ".join(out)
# =============================================================================
# Output normalization
# =============================================================================
def norm_y(y_deg: float) -> float:
"""0..180 -> -1..+1"""
return (y_deg - 90.0) / 90.0
def denorm_y(y_unit: float) -> float:
"""-1..+1 -> 0..180"""
return 90.0 + 90.0 * y_unit
# =============================================================================
# Feature engineering: temperature differences
# =============================================================================
def make_features(t6: float, t7: float, t8: float) -> List[float]:
mn = min(t6, t7, t8)
return [(t6 - mn) / DIFF_SCALE, (t7 - mn) / DIFF_SCALE, (t8 - mn) / DIFF_SCALE]
# =============================================================================
# Tiny MLP Regression: 3 -> H -> 1 (linear output)
# =============================================================================
class TinyRegMLP:
def __init__(self, hidden: int, lr: float, seed: int) -> None:
rnd = random.Random(seed)
self.h = hidden
self.lr = lr
# Layer 1: 3 -> H
self.W1 = [[(rnd.random() * 2 - 1) * 0.25 for _ in range(3)] for _ in range(hidden)]
self.b1 = [(rnd.random() * 2 - 1) * 0.05 for _ in range(hidden)]
# Layer 2: H -> 1
self.W2 = [(rnd.random() * 2 - 1) * 0.25 for _ in range(hidden)]
self.b2 = 0.0 # output in unit space (-1..+1-ish)
self.grad_clip = 5.0
def forward_unit(self, t6: float, t7: float, t8: float) -> Tuple[float, List[float], List[float], List[float]]:
x = make_features(t6, t7, t8)
pre = [0.0] * self.h
h = [0.0] * self.h
for i in range(self.h):
s = self.b1[i]
s += self.W1[i][0] * x[0]
s += self.W1[i][1] * x[1]
s += self.W1[i][2] * x[2]
pre[i] = s
h[i] = lrelu(s)
y = self.b2
for i in range(self.h):
y += self.W2[i] * h[i]
return y, x, pre, h
def predict_deg(self, t6: float, t7: float, t8: float) -> float:
y_unit, *_ = self.forward_unit(t6, t7, t8)
return denorm_y(y_unit)
def train_step(self, t6: float, t7: float, t8: float, target_deg: float) -> float:
target_unit = norm_y(target_deg)
y_unit, x, pre, h = self.forward_unit(t6, t7, t8)
err = (y_unit - target_unit)
loss = err * err
dL_dy = 2.0 * err
if dL_dy > self.grad_clip:
dL_dy = self.grad_clip
elif dL_dy < -self.grad_clip:
dL_dy = -self.grad_clip
W2_old = self.W2[:]
# Update output layer
for i in range(self.h):
self.W2[i] -= self.lr * (dL_dy * h[i])
self.b2 -= self.lr * dL_dy
# Backprop into hidden layer
for i in range(self.h):
dL_dh = dL_dy * W2_old[i]
dL_dpre = dL_dh * dlrelu(pre[i])
self.W1[i][0] -= self.lr * (dL_dpre * x[0])
self.W1[i][1] -= self.lr * (dL_dpre * x[1])
self.W1[i][2] -= self.lr * (dL_dpre * x[2])
self.b1[i] -= self.lr * dL_dpre
return loss
# =============================================================================
# Main
# =============================================================================
def print_settings() -> None:
print("\n" + "=" * 78)
print("BIR NN DEMO (REGRESSION) | TMP_6, TMP_7, TMP_8 -> SRV_4")
print("-" * 78)
print("Teacher rule (targets):")
print(" max(T6,T7,T8) == T6 -> 180°")
print(" max(T6,T7,T8) == T7 -> 90°")
print(" max(T6,T7,T8) == T8 -> 0°")
print(" tie for highest -> None (skip training sample)")
print("-" * 78)
print("Key idea:")
print(" NN input uses temps relative to MIN (offset-invariant).")
print(f" features = [(T6-min)/{DIFF_SCALE}, (T7-min)/{DIFF_SCALE}, (T8-min)/{DIFF_SCALE}]")
print("-" * 78)
print("BIR:")
print(f" IP={BIR_IP} TMP_IDS={TMP_IDS} SERVO={SERVO_ID} period={PERIOD_S}s timeout={TIMEOUT_S}s")
print("Training:")
print(
f" synthetic temps = mix: {TRAIN_RATIO*100:.0f}% micro({TRAIN_Tmicro_MIN:.1f}..{TRAIN_Tmicro_MAX:.1f}) + "
f"{(1-TRAIN_RATIO)*100:.0f}% normal({TRAIN_T_MIN:.1f}..{TRAIN_T_MAX:.1f}) °C"
)
print(f" epochs={EPOCHS} steps/epoch={STEPS_PER_EPOCH} hidden={HIDDEN} lr={LR} seed={SEED}")
print("=" * 78 + "\n")
def main() -> None:
rnd = random.Random(SEED)
model = TinyRegMLP(hidden=HIDDEN, lr=LR, seed=SEED)
print_settings()
# ---- Training ----
print("Training on synthetic data...\n")
for ep in range(1, EPOCHS + 1):
loss_sum = 0.0
trained = 0
last_y = 90.0
for step_in_ep in range(1, STEPS_PER_EPOCH + 1):
# Mixture sampling: micro range (near ties) vs normal wide range
if rnd.random() < TRAIN_RATIO:
lo, hi = TRAIN_Tmicro_MIN, TRAIN_Tmicro_MAX
else:
lo, hi = TRAIN_T_MIN, TRAIN_T_MAX
# Quantized temperatures (sensor-like)
t6 = quantize_temp(rnd.uniform(lo, hi))
t7 = quantize_temp(rnd.uniform(lo, hi))
t8 = quantize_temp(rnd.uniform(lo, hi))
# Teacher (skip tie)
y = teacher_target_deg(t6, t7, t8)
if y is None:
# still show progress sometimes, but do not train on this sample
if (step_in_ep % PROGRESS_EVERY) == 0 or step_in_ep == STEPS_PER_EPOCH:
mse = (loss_sum / trained) if trained > 0 else 0.0
nn = model.predict_deg(t6, t7, t8)
err = nn - last_y
live_line(
f"ep {ep:02d}/{EPOCHS} step {step_in_ep:6d}/{STEPS_PER_EPOCH} "
f"trained={trained:6d} mse={mse:.6f} "
f"T={t6:5.2f}, {t7:5.2f}, {t8:5.2f} "
f"REF={C_GREEN}{last_y:5.0f}{C_RESET} "
f"NN={C_ORANGE}{nn:6.1f}{C_RESET} "
f"err={err:6.1f} {C_TMP_OTHER}(tie-skip){C_RESET}"
)
continue
# Train on valid sample
last_y = y
loss = model.train_step(t6, t7, t8, y)
loss_sum += loss
trained += 1
# Live progress line
if (step_in_ep % PROGRESS_EVERY) == 0 or step_in_ep == STEPS_PER_EPOCH:
mse = (loss_sum / trained) if trained > 0 else 0.0
nn = model.predict_deg(t6, t7, t8)
err = nn - last_y
live_line(
f"ep {ep:02d}/{EPOCHS} step {step_in_ep:6d}/{STEPS_PER_EPOCH} "
f"trained={trained:6d} mse={mse:.6f} "
f"T={t6:5.2f}, {t7:5.2f}, {t8:5.2f} "
f"REF={C_GREEN}{last_y:5.0f}{C_RESET} "
f"NN={C_ORANGE}{nn:6.1f}{C_RESET} "
f"err={err:6.1f}"
)
sys.stdout.write("\n")
print("\nTraining done.")
print("Now running on REAL BIR: SRV_4 = NN output (continuous). Ctrl+C to stop.\n")
# Initialize servo to neutral position
try:
bir_set_value(BASE, SERVO_ID, 90.0)
except Exception as e:
print(f"[WARN] init servo failed: {e}")
step = 0
next_tick = time.monotonic()
# ---- Real-time loop ----
while True:
next_tick += PERIOD_S
step += 1
try:
t6, t7, t8 = bir_peer_multi(BASE, TMP_IDS)
ref = teacher_target_deg(t6, t7, t8) # 0/90/180 or None
nn = model.predict_deg(t6, t7, t8) # continuous (may overshoot)
ref_str = " tie " if ref is None else f"{ref:6.1f}"
err_str = " tie" if ref is None else f"{(nn - ref):7.1f}"
# Servo command is clamped and rounded to integer degrees
srv = int(clamp(round(nn), 0, 180))
bir_set_value(BASE, SERVO_ID, float(srv))
print(
f"[{step:05d}] TMP= {color_tmp_triplet(t6, t7, t8)} | "
f"REF={C_GREEN}{ref_str}{C_RESET} "
f"NN={C_ORANGE}{nn:6.1f}{C_RESET} err={err_str} | "
f"SRV_4={srv:3d}"
)
except (requests.RequestException, BirApiError, ValueError) as e:
print(f"[ERROR] {e}")
# Keep constant period
dt = next_tick - time.monotonic()
if dt > 0:
time.sleep(dt)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nStopped.")
Fast value exchange example
If you just want to quickly test the BIR HTTP API for reading and writing values in a fixed-period loop, here is a minimal example that does not use any ML, but still keeps a persistent session and handles errors gracefully.
The script continuously reads a set of sensor values, generates random values for a set of outputs, and writes them back to the BIR module. The console output shows the read values, the written values, and the time taken for each exchange.
Value exchange is limited to 10 inputs and 10 outputs. Testing showed that on a typical Wi-Fi connection, exchanging ten inputs and ten outputs can reach a transfer time of around 50 ms.
The demo also uses the integrated BASIC compute layer to generate random values for VAL_40 through VAL_44. This makes it possible to combine both programmable layers.
WATCH THE VIDEO EXAMPLE
# =============================================================================
# BIR FAST EXCHANGE DEMO (READ + WRITE in one call, fixed-period loop)
#
# Goal:
# - Quickly test the BIR HTTP API "exchange" endpoint in a tight loop.
# - One HTTP request does:
# * READ : a fixed list of sensor IDs
# * WRITE : a fixed list of output IDs (random values)
#
# Endpoint:
# POST {BASE}/exchange?read=ID1,ID2,ID3,...
# Body: "OUT_ID,val;OUT_ID,val;..."
#
# What you see in console:
# - One-line status with:
# * exchange time (ms)
# * READ values
# * WRITE values
# - Errors are printed as ERROR: ...
#
# Notes:
# - Uses requests.Session() to keep TCP connection alive (faster, lower overhead).
# - 503 is treated as NOT_READY (device busy / not ready).
# - Fixed-period scheduling tries to keep PERIOD_S constant (resync on overrun).
# =============================================================================
import time
import random
from typing import List, Tuple
import requests
# =========================
# USER MACROS / SETTINGS
# =========================
BASE = "http://192.168.0.202" # <-- Set your BIR / HUB base URL (IP or hostname)
TIMEOUT_S = 2.0
PERIOD_S = 0.1 # Fixed loop period (seconds). 0.1 = 100 ms
READ_IDS: List[str] = ["TMP_6", "TMP_7", "TMP_8", "INP_3", "DAY_SYS", "VAL_40", "VAL_41", "VAL_42", "VAL_43", "VAL_44"]
WRITE_IDS: List[str] = ["SRV_4", "SRV_5", "OUT_0", "OUT_1", "OUT_2", "VAL_35", "VAL_36", "VAL_37", "VAL_38", "VAL_39"]
WRITE_RANGES = {
"SRV_4": (0.0, 180.0, 1),
"SRV_5": (0.0, 180.0, 1),
"OUT_0": (0.0, 1.0, 0),
"OUT_1": (0.0, 1.0, 0),
"OUT_2": (0.0, 1.0, 0),
"VAL_35": (-100000.0, 100000.0, 2),
"VAL_36": (-100000.0, 100000.0, 2),
"VAL_37": (-100000.0, 100000.0, 2),
"VAL_38": (-100000.0, 100000.0, 2),
"VAL_39": (-100000.0, 100000.0, 2),
}
DEFAULT_RANGE = (0.0, 1.0, 2)
# =========================
# BIR EXCHANGE (keep-alive)
# =========================
class BirApiError(RuntimeError):
pass
sess = requests.Session()
def bir_exchange(
node_base: str,
read_ids: List[str],
write_pairs: List[Tuple[str, float]],
timeout_s: float = TIMEOUT_S,
) -> List[float]:
params = {"read": ",".join(read_ids)}
body = ";".join(f"{sid},{val}" for sid, val in write_pairs)
r = sess.post(f"{node_base}/exchange", params=params, data=body, timeout=timeout_s)
if r.status_code == 503:
raise BirApiError("NOT_READY (503)")
if r.status_code in (400, 404, 413):
raise BirApiError(f"{r.status_code} {r.text.strip()}")
r.raise_for_status()
txt = r.text.strip()
parts = [p.strip() for p in txt.split(";")] if txt else []
if len(parts) != len(read_ids):
raise BirApiError(f"Parse mismatch: got {len(parts)} values, expected {len(read_ids)}. Raw='{txt}'")
return [float(p.replace(",", ".")) for p in parts]
def gen_random_value(sensor_id: str) -> float:
mn, mx, dec = WRITE_RANGES.get(sensor_id, DEFAULT_RANGE)
v = random.uniform(mn, mx)
if dec <= 0:
return float(int(round(v)))
return round(v, dec)
def main():
print("BIR exchange demo (fixed period loop, keep-alive session)")
print("BASE:", BASE)
print("READ:", READ_IDS)
print("WRITE:", WRITE_IDS)
print("PERIOD_S:", PERIOD_S)
print("----")
last_len = 0
try:
next_t = time.perf_counter() # scheduled time of the next iteration
try:
while True:
loop_start = time.perf_counter()
write_pairs = [(sid, gen_random_value(sid)) for sid in WRITE_IDS]
try:
read_vals = bir_exchange(BASE, READ_IDS, write_pairs)
ok = True
err_txt = ""
except BirApiError as e:
ok = False
err_txt = str(e)
read_vals = []
loop_end = time.perf_counter()
dt_ms = (loop_end - loop_start) * 1000.0
# Prepare message
if ok:
read_str = ", ".join(f"{sid}={val}" for sid, val in zip(READ_IDS, read_vals))
write_str = ", ".join(f"{sid}={val}" for sid, val in write_pairs)
msg = f"[one exchange{dt_ms:7.1f} ms] READ: {read_str} | WRITE: {write_str}"
else:
msg = f"[one exchange{dt_ms:7.1f} ms] ERROR: {err_txt}"
# Single-line print
pad = " " * max(0, last_len - len(msg))
print("\r" + msg + pad, end="", flush=True)
last_len = len(msg)
# Fixed-period scheduling (sleep only the remaining time)
next_t += PERIOD_S
sleep_s = next_t - time.perf_counter()
if sleep_s > 0:
time.sleep(sleep_s)
else:
# Overrun: we're late. Resync to "now" to avoid drift.
next_t = time.perf_counter()
except KeyboardInterrupt:
print("\nStopped by user (Ctrl+C).")
finally:
sess.close()
if __name__ == "__main__":
main()
