Eduardo_Scripniciuc/st-ten-1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fco730 dev

Browse Source
This commit is contained in:
neo-dl 2023-10-11 09:46:38 +02:00
parent f854eff774
commit 656b56eb62
5 changed files with 118 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
config/machine_settings/st-ten-9.ini
View File

@ -15,7 +15,7 @@ hold_time: 1
[furness_controls]
port: COM4
model: t3l
model: fco730
[label_printer]
platform: windows

3
src/components/furness_controls_fco730_registers.py
View File

@ -1,3 +1,6 @@
registers = {
}
settings = {
}

3
src/components/furness_controls_fco780_registers.py
View File

@ -1,3 +1,6 @@
registers = {
}
settings = {
}

168
src/components/furness_controls_leak_tester.py
View File

@ -1,3 +1,5 @@
import sys
from lib.db import Recipes, Steps, db
from PyQt5.QtCore import QSemaphore, pyqtSignal
from PyQt5.QtWidgets import QMessageBox
@ -6,101 +8,66 @@ from .component import Component
from .modbus_component import ModbusComponent
from .furness_controls_fco730_registers import registers as fco730_registers
from .furness_controls_fco780_registers import registers as fco780_registers
from .furness_controls_fco730_registers import settings as fco730_settings
from .furness_controls_fco780_registers import settings as fco780_settings
if "--sim-furness-controls" in sys.argv:
from components.dummies.serial import serial
else:
import serial
ETX=0x03
EOT=0x04
ENQ=0x05
ACK=0X06
NACK=0X15
class FurnessControlsLeakTester(Component):
def __init__(self, config=None, name=None, period=1, lazy=True, paused=False, threaded=True):
super().__init__(config=config, name=name, period=period, lazy=lazy, paused=paused, threaded=threaded, registers=None)
self.bytesize = None
self.timeout = None
self.parity = None
self.conn = None
self.stopbits = None
self.baudrate = None
self.port = None
self.model = None
self.settings = None
self.registers = None
self.id1=0
self.id2=1
def config_changed(self):
super().config_changed()
self.model = self.config[self.name]["model"].lower()
if self.model == "fco730":
self.registers = fco730_registers
self.settings = fco730_settings
elif self.model == "fco730":
self.registers = fco780_registers
self.settings = fco780_settings
else:
raise NotImplementedError(f"Furness Controls model {self.model!r} not implemented.")
self.set_measure_units()
self.units = self.get_measure_units()
self.max_program_number = self.read("Max number of programs")
self.model = self.config[self.name]["model"].lower()
self.log.info(f"units: {self.units}")
_pressure_units = {"mH2O": 0, "mbar": 1, "kPa": 2, "mmHg": 3, "inH2O": 4, "psi": 5, "mmH2O": 6, } # (se fondoscala <=6 bar)
_leak_units = {"mmH2O": 0, "mbar": 1, "Pa": 2, "mmHg": 3, "inH2O": 4, "psi": 5, }
_leak_flow_units = {"cm3/min": 0, "cm3/h": 1, }
_volume_units = {"litri": 0, "cm3": 1, }
_time_units = {"seconds": 0, }
_flow_units = {"liters/min": 0, "liters/h": 1, "m3/h": 2, "cfm": 3}
_pressure_units_map = {v: k for k, v in _pressure_units.items()}
_leak_units_map = {v: k for k, v in _leak_units.items()}
_leak_flow_units_map = {v: k for k, v in _leak_flow_units.items()}
_volume_units_map = {v: k for k, v in _volume_units.items()}
_time_units_map = {v: k for k, v in _time_units.items()}
_flow_units_map = {v: k for k, v in _flow_units.items()}
def set_measure_units(self):
if self.model == "t3p":
for register, [unit, decimals] in {
"MEASURE UNITS: Relative pressure": [self._pressure_units["mbar"], 0], # red, purple
"MEASURE UNITS: Differential (leak) pressure": [self._leak_units["mbar"], 0], # yellow
"MEASURE UNITS: Calculated leak flow rate": [self._leak_flow_units["cm3/min"], 0], # blue
"MEASURE UNITS: Volume": [self._volume_units["litri"], 0], # green
"MEASURE UNITS: Flow rate": [self._flow_units["liters/min"], 0], # orange
}.items():
self.write(register, (decimals << 8) + unit)
elif self.model == "t3l":
for register, [unit, decimals] in {
"MEASURE UNITS: Relative pressure": [self._pressure_units["mbar"], 0], # red, purple
"MEASURE UNITS: Differential (leak) pressure": [self._leak_units["mbar"], 0], # yellow
"MEASURE UNITS: Calculated leak flow rate": [self._leak_flow_units["cm3/min"], 0], # blue
"MEASURE UNITS: Volume": [self._volume_units["litri"], 0], # green
"MEASURE UNITS: Flow rate": [self._flow_units["liters/min"], 0], # orange
}.items():
self.write(register, unit) # (decimals << 8) + unit)
else:
raise NotImplementedError(f"tecna t3 model {self.model!r} not implemented.")
def get_measure_units(self):
units = {}
if self.model == "t3p":
for [register, unit_map, unit_names] in [
["Running test: relative pressure format", self._pressure_units_map, ["relative_pressure", "red", "r", 21, ]], # also by documentation color and register number
["Running test: differential pressure format", self._pressure_units_map, ["differential_pressure", "purple", "p", 22, ]], # also by documentation color and register number
["Running test: relative pressure format (low resolution)", self._leak_units_map, ["relative_pressure_lr", "yellow", "y", 23, ]], # also by documentation color and register number
["Running test: calculated leak flow rate format", self._leak_flow_units_map, ["leak_flow", "blue", "b", 24, ]], # also by documentation color and register number
["Running test: volume format", self._volume_units_map, ["volume", "green", "g", 25, ]], # also by documentation color and register number
["Running test: time format", self._time_units_map, ["time", "orange", "t", 26, ]], # also by documentation color and register number
["Running test: flow rate format", self._flow_units_map, ["flow", "white", "o", 27, ]], # also by documentation color and register number
]:
v = self.read(register)
unit_spec = [10**(-((v >> 8) & 0xff)), unit_map[v & 0xff]]
for unit_name in unit_names:
units[unit_name] = unit_spec
elif self.model == "t3l":
for [register, unit_map, unit_names] in [
[["Running test: relative pressure scale", "Running test: relative pressure decimals"], self._pressure_units_map, ["relative_pressure", "red", "r", 1501, ]], # also by documentation color and register number
[["Running test: differential pressure scale", "Running test: differential pressure decimals"], self._pressure_units_map, ["differential_pressure", "purple", "p", 1503, ]], # also by documentation color and register number
[["Running test: relative pressure scale (low resolution)", "Running test: relative pressure decimals (low resolution)"], self._leak_units_map, ["relative_pressure_lr", "yellow", "y", 1505, ]], # also by documentation color and register number
["Running test: calculated leak flow rate format", self._leak_flow_units_map, ["leak_flow", "blue", "b", 1507, ]], # also by documentation color and register number
["Running test: volume format", self._volume_units_map, ["volume", "green", "g", 1508, ]], # also by documentation color and register number
["Running test: time format", self._time_units_map, ["time", "orange", "t", 1509, ]], # also by documentation color and register number
["Running test: flow rate format", self._flow_units_map, ["flow", "white", "o", 1510, ]], # also by documentation color and register number
["Running test: line pressure format", self._pressure_units_map, ["line_pressure", "lp", "l", 1511, ]], # also by documentation color and register number
]:
if type(register) is list:
v = [self.read(r) for r in register]
unit_spec = [10**(-(v[1] & 0xff)), v[0]]
else:
v = self.read(register)
unit_spec = [10**(-((v >> 8) & 0xff)), unit_map[v & 0xff]]
for unit_name in unit_names:
units[unit_name] = unit_spec
else:
raise NotImplementedError(f"tecna t3 model {self.model!r} not implemented.")
return units
self.port = self.config[self.name]["port"]
self.baudrate = int(self.config[self.name]["baudrate"])
self.stopbits = getattr(serial, self.config[self.name].get("stopbits", "stopbits_one").upper())
self.parity = getattr(serial, self.config[self.name].get("parity", "parity_none").upper())
self.bytesize = getattr(serial, self.config[self.name].get("bytesize", "eightbits").upper())
self.timeout = float(self.config[self.name].get("read_timeout", 1))
if self.conn is not None:
self.conn.close()
self.conn = serial.Serial(
self.port,
baudrate=self.baudrate,
stopbits=self.stopbits,
parity=self.parity,
bytesize=self.bytesize,
timeout=self.timeout,
write_timeout=self.timeout,
#inter_byte_timeout=self.timeout,
)
def _convert_from_format(self, data, formatting=None, decoding_map=None):
if decoding_map is not None and data in decoding_map:
@ -296,16 +263,39 @@ class FurnessControlsLeakTester(Component):
"Various flags": 0b0000000000010000 if self.config["recipes_defaults"]["tecna_discharge_enable"]=="yes" else 0b0000000000000000
}
if self.model == "t3p":
pass
elif self.model == "t3l":
spec.update({
"Use programs or use products": 0,
"Nominal peak pressure": step.spec["test_pressure"],
"Pn - Nominal test pressure": step.spec["test_pressure"],
})
else:
raise NotImplementedError(f"tecna t3 model {self.model!r} not implemented.")
self.log.debug(str(spec))
for register, value in spec.items():
self.write(register, value)
def send_command(self,command):
out_bytes= bytearray()
out_bytes.append(EOT)
out_bytes.append(self.id1)
out_bytes.append(self.id2)
self.conn.write(out_bytes)
response = self.conn.read(1)
if response != ACK:
self.log.error(f"SEND COMMAND:{response}")
def send_enquiry(self,enquiry):
out_bytes=bytearray(EOT)
out_bytes.append(self.id1)
out_bytes.append(self.id2)
out_bytes.append(enquiry)
out_bytes.append(ENQ)
checksum=self.calc_checksum(out_bytes)
out_bytes.append(checksum)
self.conn.write(out_bytes)
response = self.conn.read(1)
if response != ACK:
self.log.error(f"SEND COMMAND:{response}")
def calc_checksum(self,data):
checksum=0
for i,data_byte in enumerate(data):
if i==0:
continue # skip EOT
checksum=checksum^data_byte
return checksum

32
src/test/furness_controls.py Normal file
View File

@ -0,0 +1,32 @@
import sys
from datetime import datetime
from PyQt5.QtWidgets import QApplication
from components import Os_Label_Printer
from furness_controls_leak_tester import FurnessControlsLeakTester
from lib.helpers import ConfigReader
config = ConfigReader(system_id="vm")
app = QApplication(sys.argv)
tester_component = FurnessControlsLeakTester(name="furness_controls")
command=bytearray(['F',0,5,0])
response=tester_component.send_command(command)
if response is not None:
print(f"RESPONSE:{response}")
else:
print("COMMAND ERROR")
enquiry=bytearray(['F',0,5,0])
response=tester_component.send_enquiry(enquiry)
if response is not None:
print(f"RESPONSE:{response}")
else:
print("INQUIRY ERROR")