From 656b56eb624cb51452b74934d8b3b3f29fd95245 Mon Sep 17 00:00:00 2001 From: neo-dl Date: Wed, 11 Oct 2023 09:46:38 +0200 Subject: [PATCH] fco730 dev --- config/machine_settings/st-ten-9.ini | 2 +- .../furness_controls_fco730_registers.py | 3 + .../furness_controls_fco780_registers.py | 3 + .../furness_controls_leak_tester.py | 168 ++++++++---------- src/test/furness_controls.py | 32 ++++ 5 files changed, 118 insertions(+), 90 deletions(-) create mode 100644 src/test/furness_controls.py diff --git a/config/machine_settings/st-ten-9.ini b/config/machine_settings/st-ten-9.ini index a19c0f2..63ca523 100644 --- a/config/machine_settings/st-ten-9.ini +++ b/config/machine_settings/st-ten-9.ini @@ -15,7 +15,7 @@ hold_time: 1 [furness_controls] port: COM4 -model: t3l +model: fco730 [label_printer] platform: windows diff --git a/src/components/furness_controls_fco730_registers.py b/src/components/furness_controls_fco730_registers.py index e804068..cb9cc6a 100644 --- a/src/components/furness_controls_fco730_registers.py +++ b/src/components/furness_controls_fco730_registers.py @@ -1,3 +1,6 @@ registers = { } +settings = { + +} diff --git a/src/components/furness_controls_fco780_registers.py b/src/components/furness_controls_fco780_registers.py index e804068..cb9cc6a 100644 --- a/src/components/furness_controls_fco780_registers.py +++ b/src/components/furness_controls_fco780_registers.py @@ -1,3 +1,6 @@ registers = { } +settings = { + +} diff --git a/src/components/furness_controls_leak_tester.py b/src/components/furness_controls_leak_tester.py index 96ef1d9..ffe5756 100644 --- a/src/components/furness_controls_leak_tester.py +++ b/src/components/furness_controls_leak_tester.py @@ -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 diff --git a/src/test/furness_controls.py b/src/test/furness_controls.py new file mode 100644 index 0000000..f2cac4c --- /dev/null +++ b/src/test/furness_controls.py @@ -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") \ No newline at end of file