pigeon-transmission 0.0.6

This library provides a high-level interface for communicating with IMT Analytics devices via RS232

General information

pigeon‑transmission is an independently developed Python library that provides simple and reliable serial communication with measurement devices from IMT Analytics. It is designed for users who want to automate data acquisition, integrate measurement workflows, or control devices programmatically.

This is an unofficial project and is not developed, reviewed, or supported by IMT Analytics. The development of this library was carried out entirely independently of IMT Analytics.

⚙️ Installation

Python 3.10 or higher is required

pip install pigeon-transmission

🚀 Example Usage

First Example - with statement

from pigeon_transmission import DeviceH4

with DeviceH4(port="com2") as device:
    print(device.write_setting(operation_name="gas type", value="air"))  # WS

Second Example - Explicit Resource Management

from pigeon_transmission import DeviceH4

device = DeviceH4()
device.open_serial_interface(port="com2")  # open
print(device.write_setting(operation_name="gas type", value="air"))  # WS
device.close_serial_interface()  # close

Third Example - Manual Resource Management (The instance does not need to stay alive)

from pigeon_transmission import DeviceH4

device = DeviceH4()
serial_object = device.open_serial_interface(port="com2")  # open
print(device.execute_command(operation_name="switch echo", value="off", serial_object=serial_object))  # CM
device.close_serial_interface(serial_object=serial_object)  # close

Data stream (read 3 values)

import queue  # only for data stream
import threading  # only for data stream
from pigeon_transmission import DeviceH4

device = DeviceH4()
device.open_serial_interface(port="com2")  # open serial connection

data_queue, stop_queue = queue.Queue(), queue.Queue()

# Start background thread for data stream
thread_stream = threading.Thread(
    target=device.start_stream,
    args=("high flow", "differential pressure", "temperature",
          data_queue, stop_queue, 5),
    daemon=True
)
thread_stream.start()

timeout_counter_data_queue = 0  # initialize the counter
# Main loop: receive streamed measurement data
while stop_queue.empty():
    try:
        data = data_queue.get(timeout=1)
        timeout_counter_data_queue = 0  # reset the counter
        print(data)
        # --- Handle incoming data here ---
        # Process, log, visualize, or store the measurement values.
        # To stop streaming, call:
        #     stop_queue.put(True)
        # This will end the loop and signal the thread to stop.
        # ---------------------------------
    except queue.Empty:
        timeout_counter_data_queue += 1  # increment by 1

# Wait for the streaming thread to finish
thread_stream.join()
device.close_serial_interface()  # close serial connection

# After the thread has finished, check whether the device timed out
if timeout_counter_data_queue >= 10:
    # Note: The device has an internal timeout of 10 seconds.
    # The data queue timeout is 1 second, so if this occurs more than 10 times,
    # the device is considered timed out. Note: Depending on how slow the while loop
    # is, the 10-second timeout may vary
    print("The device timed out")  # Handle the timeout here

Fast data stream (read 12 values) - only for DeviceH5

import queue  # only for data stream
import threading  # only for data stream
from pigeon_transmission import DeviceH5

device = DeviceH5()
device.open_serial_interface(port="com2", baudrate=115200)  # open serial connection

data_queue, stop_queue = queue.Queue(), queue.Queue()

# Start background thread for data stream
thread_stream = threading.Thread(
    target=device.start_stream,
    args=("high flow", "differential pressure", "temperature",  # 12 values
          "mean pressure", "peep", "peak flow inspiration",
          "ambient pressure", "plateau pressure", "ipap",
          "current breath phase", "oxygen", "I:E Ratio",
          data_queue, stop_queue, 5),
    daemon=True
)
thread_stream.start()

timeout_counter_data_queue = 0  # initialize the counter
# Main loop: receive streamed measurement data
while stop_queue.empty():
    try:
        data = data_queue.get(timeout=1)
        timeout_counter_data_queue = 0  # init the counter
        print(data)
        # --- Handle incoming data here ---
        # Process, log, visualize, or store the measurement values.
        # To stop streaming, call:
        #     stop_queue.put(True)
        # This will end the loop and signal the thread to stop.
        # ---------------------------------
    except queue.Empty:
        timeout_counter_data_queue += 1  # increment by 1

# Wait for the streaming thread to finish
thread_stream.join()
device.close_serial_interface()  # close serial connection

# After the thread has finished, check whether the device timed out
if timeout_counter_data_queue >= 10:
    # Note: The device has an internal timeout of 10 seconds.
    # The data queue timeout is 1 second, so if this occurs more than 10 times,
    # the device is considered timed out. Note: Depending on how slow the while loop
    # is, the 10-second timeout may vary
    print("The device timed out")  # Handle the timeout here

Notes

• Measurement of I:E ratio – Use Ti/Te if Ti > Te, otherwise Te/Ti