Metadata-Version: 2.4
Name: pigeon-transmission
Version: 0.0.9
Summary: This library provides a high-level interface for communicating with IMT Analytics devices via RS232
Author: Bachatero18, Dennis Schramm
License: GPL-3.0-or-later
Keywords: pigeon-transmission,pigeon,imt,imt-analytics,citrex,H4,H5,flow,analyser,serial,transmission
Classifier: Development Status :: 5 - Production/Stable
Classifier: License :: OSI Approved :: GNU General Public License v3 (GPLv3)
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: End Users/Desktop
Classifier: Intended Audience :: Education
Classifier: Operating System :: OS Independent
Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: POSIX :: Linux
Classifier: Operating System :: MacOS :: MacOS X
Classifier: Operating System :: Unix
Classifier: Operating System :: POSIX :: BSD
Classifier: Operating System :: POSIX :: BSD :: FreeBSD
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pyserial
Dynamic: license-file

# 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:

```python
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`:

```python
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:

```python
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
```python
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
```python
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)
```python
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
```python
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)
```python
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
```python
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
