Small Python helper for streaming heart-rate measurements from Polar BLE devices.
polar-py is designed to run heart-rate streaming on the side while your main application keeps
doing other work. The threaded API keeps BLE connection and notification handling in background
threads, and the optional CSV logger writes from its own background thread too.
From this repository:
pip install .For editable development:
pip install -e ".[dev]"To run the live plotting demo:
pip install -e ".[plot]"
python examples/live_plot.pyUse connect_threaded() to pre-connect to the Polar sensor, then start_stream_threaded() to
begin forwarding heart-rate samples. Your main thread remains free for other real-time work.
import time
from polar_py import PolarPy
polar = PolarPy()
print("Connecting to Polar sensor...")
polar.connect_threaded()
if not polar.wait_until_connected(timeout=15.0):
polar.disconnect_threaded()
raise SystemExit("Failed to connect.")
t0 = time.perf_counter()
def shared_ts():
return time.perf_counter() - t0
def on_sample(ts_iso, hr, rr, shared):
# This callback runs from a background thread. Keep it fast.
print(ts_iso, hr, rr, shared)
polar.start_csv_logger("hr_log.csv")
polar.start_stream_threaded(
on_sample=on_sample,
log_to_csv=True,
print_samples=True,
shared_ts_provider=shared_ts,
)
try:
while True:
# Your main application work continues here while Polar streaming runs
# in the background.
time.sleep(0.1)
finally:
polar.stop_stream_threaded()
polar.stop_csv_logger()
polar.disconnect_threaded()For lowest latency, avoid heavy work inside on_sample. Copy values into a thread-safe queue or
shared state, then process them from your main application loop.
For multiple back-to-back sessions, reuse the same PolarPy instance and keep the BLE connection
open. Start a new CSV logger for each session, stop streaming before stopping the logger, then
start the next session with a different file.
from polar_py import PolarPy
polar = PolarPy()
polar.connect_threaded()
if not polar.wait_until_connected(timeout=15.0):
polar.disconnect_threaded()
raise SystemExit("Failed to connect.")
try:
polar.start_csv_logger("session_1.csv")
polar.start_stream_threaded(log_to_csv=True)
run_session_1()
polar.stop_stream_threaded()
polar.stop_csv_logger()
polar.start_csv_logger("session_2.csv")
polar.start_stream_threaded(log_to_csv=True)
run_session_2()
polar.stop_stream_threaded()
polar.stop_csv_logger()
finally:
polar.disconnect_threaded()You do not need to create a separate logger object. PolarPy owns one internally and can reopen
it for a new file after the previous logger has stopped.
addressmay be a MAC address on Windows/Linux or a CoreBluetooth UUID on macOS.- If no address is provided,
PolarPyscans and selects the first discovered device with "polar" in its advertised name. connect_threaded()keeps the BLE connection alive untildisconnect_threaded()is called.stop_stream_threaded()stops forwarding samples but leaves a pre-connected BLE session open; passdisconnect=Trueor calldisconnect_threaded()to close the connection.