pigeon-transmission 0.0.9

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

🔠 Enums

Instead of using plain strings, you can use the provided Enums for better IDE autocompletion and type safety:

from pigeon_transmission import DeviceH5, EnumH5

# Instead of:
device.write_setting(operation_name="gas type", value="air")

# You can use:
device.write_setting(operation_name=EnumH5.WriteSetting.GAS_TYPE, value=EnumH5.WriteSetting_GasType.AIR)

Both styles are fully supported and produce identical results. All available Enums can be explored directly in your IDE by typing EnumH5. and using autocompletion.

📏 Limits & Units

You can retrieve the valid limits and units for any measurement using get_operation_limit_and_unit:

from pigeon_transmission import DeviceH5, EnumH5

limits_and_units = device.get_operation_limit_and_unit(
    operation_names=[
        EnumH5.ReadMeasurement.HIGH_FLOW,
        EnumH5.ReadMeasurement.DIFFERENTIAL_PRESSURE,
        EnumH5.ReadMeasurement.TEMPERATURE
    ]
)

# Access limits and unit for a specific measurement
print(limits_and_units[EnumH5.ReadMeasurement.HIGH_FLOW]["min"])  # minimum value
print(limits_and_units[EnumH5.ReadMeasurement.HIGH_FLOW]["max"])  # maximum value
print(limits_and_units[EnumH5.ReadMeasurement.HIGH_FLOW]["unit"])  # unit e.g. "l/min"

🪵 Logging

Logging can be enabled by passing log_level to the constructor. All communication details, lifecycle events, and errors are then written to the console and optionally to a log file. Logging can also be changed or disabled at any time during an active session:

import logging
from pigeon_transmission import DeviceH4

device = DeviceH4(port="COM2", log_level=logging.DEBUG)
device.set_logging(enabled=False, level=logging.DEBUG)  # disable logging
device.set_logging(enabled=True, level=logging.WARNING)  # re-enable with higher level
device.set_logging(enabled=True, level=logging.DEBUG, log_file="device.log")  # add log file

Available log levels:

Level	Constant	Description
10	logging.DEBUG	Every sent/received byte, internal state changes
20	logging.INFO	Lifecycle events (port open/close, stream start/stop)
30	logging.WARNING	Retries, unexpected but recoverable situations
40	logging.ERROR	Failures directly before an exception is raised

🚀 Example Usage

First Example - with statement

from pigeon_transmission import DeviceH4, EnumH4
import logging

with DeviceH4(port="com2", log_level=logging.WARNING) as device:
    device.write_setting(operation_name="gas type", value="air")  # WS with string
    device.write_setting(operation_name=EnumH4.WriteSetting.GAS_TYPE, value=EnumH4.WriteSetting_GasType.AIR)

Second Example - Explicit Resource Management

from pigeon_transmission import DeviceH4, EnumH4

device = DeviceH4()
device.open_serial_interface(port="com2")  # open
device.write_setting(operation_name=EnumH4.WriteSetting.GAS_TYPE, value=EnumH4.WriteSetting_GasType.AIR)
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
device.execute_command(operation_name="switch echo", value="off", serial_object=serial_object)  # CM
device.close_serial_interface(serial_object=serial_object)  # close

Read multiple measurements - uses "read_measurement" method to acquire repeated measurements

from pigeon_transmission import DeviceH5, EnumH5

with DeviceH5(port="COM2", baudrate=19200) as device:
    values = device.read_multiple_measurements(
        operation_name=EnumH5.ReadMeasurement.HIGH_FLOW,
        sample_count=10,   # number of measurements
        sample_rate=2      # measurements per second (min: 0.01 | max: 20)
    )
    print(values)  # ['15.0', '15.1', '14.9', ...]

Note: The sample_rate is an approximate target value. Due to the nature of serial communication, the actual timing between measurements cannot be guaranteed with exact precision. For time-critical applications, use the data stream instead.

Data stream (read 3 values)

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

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

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

# Start background thread for data stream
# Tip: You can also use Enums and Strings here
thread_stream = threading.Thread(
    target=device.start_stream,
    args=(EnumH4.StartStream.HIGH_FLOW,
          EnumH4.StartStream.DIFFERENTIAL_PRESSURE,
          EnumH4.StartStream.TEMPERATURE,
          data_queue, stop_queue, error_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

# After the thread has finished, check whether the device timed out
if not error_queue.empty():
    raise error_queue.get(timeout=1)

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, error_queue = queue.Queue(), queue.Queue(), queue.Queue()

# Start background thread for data stream
# Tip: You can also use Enums here
thread_stream = threading.Thread(
    target=device.start_fast_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, error_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

# After the thread has finished, check whether the device timed out
if not error_queue.empty():
    raise error_queue.get(timeout=1)

Notes

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