58 lines
1.5 KiB
Python
58 lines
1.5 KiB
Python
|
import librtd, os
|
||
|
|
|
||
|
|
|
||
|
|
stacks = [int(h) for h in os.getenv("MODULE_RTD8_LEVELS", "")]
|
||
|
|
if not stacks:
|
||
|
|
print("MODULE_RTD8_LEVELS is not set or invalid")
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def read_temp(stack: int, channel: int) -> float | str:
|
||
|
|
"""
|
||
|
|
Read specified channel in Celsius
|
||
|
|
"""
|
||
|
|
if stack in stacks and 1 <= channel <= 8:
|
||
|
|
try:
|
||
|
|
return librtd.get(stack, channel)
|
||
|
|
except Exception as e:
|
||
|
|
raise ValueError(f"Error read data from card {stack} on channel {channel} by error: {e}")
|
||
|
|
|
||
|
|
return -512
|
||
|
|
|
||
|
|
def read_temp_stack(stack: int) -> dict:
|
||
|
|
"""
|
||
|
|
Read all inputs data in Celsius
|
||
|
|
"""
|
||
|
|
return {ch: read_temp(stack=stack, channel=ch) for ch in range(1, 9)}
|
||
|
|
|
||
|
|
def read_temp_all() -> dict:
|
||
|
|
"""
|
||
|
|
Read all RTD8 cards inputs in Celsius
|
||
|
|
"""
|
||
|
|
return {stack: read_resistance_stack(stack=stack) for stack in stacks}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def read_resistance(stack: int, channel: int) -> float | str:
|
||
|
|
"""
|
||
|
|
Read specified channel in Ohm
|
||
|
|
"""
|
||
|
|
if stack in stacks and 1 <= channel <= 8:
|
||
|
|
try:
|
||
|
|
return librtd.getRes(stack, channel)
|
||
|
|
except Exception as e:
|
||
|
|
raise ValueError(f"Error read data from card {stack} on channel {channel} by error: {e}")
|
||
|
|
|
||
|
|
return -512
|
||
|
|
|
||
|
|
def read_resistance_stack(stack: int) -> dict:
|
||
|
|
"""
|
||
|
|
Read all inputs data in Ohm
|
||
|
|
"""
|
||
|
|
return {ch: read_resistance(stack=stack, channel=ch) for ch in range(1, 9)}
|
||
|
|
|
||
|
|
def read_resistance_all() -> dict:
|
||
|
|
"""
|
||
|
|
Read all RTD8 cards inputs in Ohm
|
||
|
|
"""
|
||
|
|
return {stack: read_resistance_stack(stack=stack) for stack in stacks}
|