pigeon-transmission 0.0.2

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

Fast data mode

import queue  # only for fast data
import threading  # only for fast data
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 fast data streaming
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

Notes

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