pigeon-transmission 0.0.3

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()

# Main loop: receive streamed measurement data
while stop_queue.empty():
    try:
        data = data_queue.get(timeout=1)
        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:
        pass

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

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()

# Main loop: receive streamed measurement data
while stop_queue.empty():
    try:
        data = data_queue.get(timeout=1)
        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:
        pass

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

Notes

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