########################################################################### # # Copyright (c) 2021-2024 Diality Inc. - All Rights Reserved. # # THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN # WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. # # @file fp_defs.py # # @author (last) Zoltan Miskolci # @date (last) 12-Jan-2026 # @author (original) Peter Lucia # @date (original) 22-Jun-2021 # ############################################################################ from aenum import extend_enum from typing import TYPE_CHECKING from .dd_defs import DDConductivitySensorsEnum, DDPressureNames, DDValveNames from .global_vars import GlobalVariables, test_config_change from ..utils.base import DialinEnum class FP_Defs: # IDE autocomplete suppor if TYPE_CHECKING: FPOpModes: DialinEnum FPFaultStates: DialinEnum FPServiceStates: DialinEnum FPPostStates: DialinEnum FPStandbyStates: DialinEnum FPPreGenPermeateStates: DialinEnum FPPreGenPDefStates: DialinEnum FPGenPermeateStates: DialinEnum FPGenPermeateDefStates: DialinEnum FPNotLegalStates: DialinEnum FPEventList: DialinEnum FPEventDataTypes: DialinEnum FPBoostPumpNames: DialinEnum FPConductivitySensorNames: DialinEnum FPRORejectionRatioNames: DialinEnum FPFlowSensorNames: DialinEnum FPFloaterLevelSensorNames: DialinEnum FPPressureSensorNames: DialinEnum FPROPumpNames: DialinEnum FPTemperatureSensorNames: DialinEnum FPValveNames: DialinEnum FPAllPumpNames: DialinEnum FPFloaterLevels: DialinEnum FPPumpAttributes: DialinEnum FPValveStates: DialinEnum # A Dictionary storing which Property(Enum) should be created by which function _properties = { # Operation / State enums 'FPOpModes': '_create_fp_operation_modes', 'FPFaultStates': '_create_fp_fault_states', 'FPServiceStates': '_create_fp_service_states', 'FPPostStates': '_create_fp_post_states', 'FPStandbyStates': '_create_fp_standby_states', 'FPPreGenPermeateStates': '_create_fp_pregen_states', 'FPPreGenPDefStates': '_create_fp_pregen_defeatured_states', 'FPGenPermeateStates': '_create_fp_gen_permeate_states', 'FPGenPermeateDefStates': '_create_fp_gen_permeate_defeatured_states', 'FPNotLegalStates': '_create_fp_not_legal_states', # Event enums 'FPEventList': '_create_fp_eventlist', 'FPEventDataTypes': '_create_fp_event_data_types', # Module Names enums 'FPBoostPumpNames': '_create_fp_boost_pump_names', 'FPConductivitySensorNames': '_create_fp_conductivity_sensor_names', 'FPRORejectionRatioNames': '_create_fp_ro_rejection_ratio_names', 'FPFlowSensorNames': '_create_fp_flow_sensor_names', 'FPFloaterLevelSensorNames': '_create_fp_floater_sensor_names', 'FPPressureSensorNames': '_create_fp_pressure_sensor_names', 'FPROPumpNames': '_create_fp_ro_pump_names', 'FPValveNames': '_create_fp_valve_names', 'FPAllPumpNames': '_create_fp_all_pump_names', 'FPTemperatureSensorNames': '_create_fp_temperature_sensor_names', # Support enums 'FPFloaterLevels': '_create_fp_floater_levels', 'FPPumpAttributes': '_create_fp_pump_attributes', 'FPValveStates': '_create_fp_valve_states', } def __init__(self): self.update_enums() # Dynamically creates all the properties in runtime def __getattr__(self, name) -> DialinEnum: if name in self._properties: value = getattr(self, self._properties[name])() setattr(self, name, value) return value raise AttributeError(name) # Attached to a listener to update the enums when the listener is triggered def update_enums(self) -> None: for prop_name in self._properties: # Cleans the cached value self.__dict__.pop(prop_name, None) # Reinitializing the enum as it's not in the cache self.__getattr__(prop_name) # ================================================== Enum Creators: Operations ================================================== def _create_fp_operation_modes(self): members = { 'MODE_FAUL': 0, # Fault mode 'MODE_SERV': 1, # Service mode 'MODE_INIT': 2, # Initialization & POST mode 'MODE_STAN': 3, # Standby mode 'MODE_PRE_GENP': 4, # Pre Generate Permeate mode 'MODE_GENP': 5, # Generate Permeate mode 'MODE_DPGP': 6, # Defeatured Pre-Generate Permeate mode 'MODE_DEGP': 7, # Defeatured Generate Permeate mode 'MODE_NLEG': 8, # Not legal - an illegal mode transition occurred 'NUM_OF_FP_MODES': 9, # Number of FP operation modes } return DialinEnum('FPOpModes', members) def _create_fp_fault_states(self): members = { 'FP_FAULT_STATE_START': 0, # FP Fault Start State 'FP_FAULT_DEENERGIZED_STATE': 1, # FP Fault De-energized State 'FP_FAULT_ENERGIZED_STATE': 2, # FP Fault Energized State 'NUM_OF_FP_FAULT_STATES': 3, # Number of Fault Mode State } return DialinEnum('FPFaultStates', members) def _create_fp_service_states(self): members = { 'NUM_OF_FP_SERV_STATES': 0, # Number of FP service states } return DialinEnum('FPServiceStates', members) def _create_fp_init_post_states(self): members = { 'FP_POST_STATE_START': 0, # Start initialize & POST mode state 'FP_POST_STATE_FW_INTEGRITY': 1, # Run firmware integrity test state 'FP_POST_STATE_NVDATAMGMT': 2, # Run NV Data Mgmt. test state 'FP_POST_STATE_WATCHDOG': 3, # Run watchdog test state 'FP_POST_STATE_SAFETY_SHUTDOWN': 4, # Run safety shutdown test state 'FP_POST_STATE_TEMPERATURES': 5, # Run temperatures POST state 'FP_POST_STATE_FW_COMPATIBILITY': 6, # Run firmware compatibility test state 'FP_POST_STATE_FPGA': 7, # Run FPGA test state 'FP_POST_STATE_COMPLETED': 8, # POST self-tests completed state 'FP_POST_STATE_FAILED': 9, # POST self-tests failed state 'NUM_OF_FP_POST_STATES': 10 # Number of initialize & POST mode states } return DialinEnum('FPPostStates', members) def _create_fp_standby_states(self): members = { 'FP_STANDBY_MODE_STATE_IDLE': 0, # Idle standby mode state 'NUM_OF_FP_STANDBY_MODE_STATES': 1, # Number of standby mode states } return DialinEnum('FPStandbyStates', members) def _create_fp_pregen_states(self): members = { 'FP_PRE_GENP_INLET_PRES_CHECK': 0, # FP Pre Gen Permeate Inlet Pressure Check State 'FP_PRE_GENP_FILTER_FLUSH': 1, # FP Pre Gen Permeate Filter Flush State 'FP_PRE_GENP_PERMEATE_FLUSH': 2, # FP Pre Gen Permeate Permeate Flush State 'FP_PRE_GENP_CONCENTRATE_FLUSH': 3, # FP Pre Gen Permeate Concentrate Flush State 'FP_PRE_GENP_VERIFY_WATER': 4, # FP Pre Gen Permeate Verify Water State 'FP_PRE_GENP_PAUSED': 5, # FP Pre Gen Permeate Paused State 'NUM_OF_FP_PRE_GENP_MODE_STATES': 6, # Number of Pre-Gen Permeate mode states } return DialinEnum('FPPreGenPermeateStates', members) def _create_fp_pregen_defeatured_states(self): members = { 'FP_PRE_GENP_DEF_FLUSH': 0, # Pre Gen Permeate Defeatured Flush state 'FP_PRE_GENP_DEF_INLET_WATER_CHECK': 1, # Pre Gen Permeate Defeatured Inlet Water Check state 'FP_PRE_GENP_DEF_PAUSED': 2, # Defeatured Pre Gen Permeate Paused state 'NUM_OF_FP_PRE_GENP_DEF_MODE_STATES': 3, # Number of Defeatured Pre Gen Permeate states } return DialinEnum('FPPreGenPDefStates', members) def _create_fp_gen_permeate_states(self): members = { 'FP_GENP_TANK_FILL_STATE': 0, # Gen Permeate Tank Fill low state 'FP_GENP_TANK_FULL_STATE': 1, # Gen Permeate Tank Full state 'NUM_OF_FP_GENP_MODE_STATES': 2, # Number of Gen permeate states } return DialinEnum('FPGenPermeateStates', members) def _create_fp_gen_permeate_defeatured_states(self): members = { 'FP_GENP_DEF_SUPPLY_WATER': 0, # Gen Permeate Defeatured Supply Water statee 'FP_GENP_DEF_PAUSED': 1, # Gen Permeate Defeatured Paused state 'NUM_OF_FP_GENP_DEF_MODE_STATES': 2, # Number of Defeatured Gen permeate states } return DialinEnum('FPGenPermeateDefStates', members) def _create_fp_not_legal_states(self): members = { 'NUM_OF_NOT_LEGAL_STATES': 0, # Number of Not Legal states } return DialinEnum('FPNotLegalStates', members) def _create_fp_eventlist(self): members = { 'FP_EVENT_STARTUP': 0, # FP startup event 'FP_EVENT_OP_MODE_CHANGE': 1, # FP Op mode change event 'FP_EVENT_SUB_MODE_CHANGE': 2, # FP Op sub-mode change event 'FP_EVENT_PRE_GEN_RO_SET_PWM': 3, # FP gen permeate ro set pwm event 'FP_EVENT_GENP_BOOST_SET_PWM': 4, # FP gen permeate boost set pwm event 'FP_EVENT_GENP_CHANGE': 5, # FP gen permeate state change 'FP_EVENT_PRE_GEN_CHANGE': 6, # FP pre gen state change 'FP_EVENT_PRE_GEN_DEF_CHANGE': 7, # FP defeatured pre gen state change 'FP_EVENT_GENP_DEF_CHANGE': 8, # FP defeatured gen permeate state change 'FP_EVENT_FAULT_ALARM_TRIGGER': 9, # FP event for alarms that would trigger 'NUM_OF_FP_EVENT_IDS': 10, # Number of FP events } return DialinEnum('FPEventList', members) def _create_fp_event_data_types(self): members = { 'EVENT_DATA_TYPE_NONE': 0, # No Event Data Type 'EVENT_DATA_TYPE_U32': 1, # Unsigned 32bit Event Data Type 'EVENT_DATA_TYPE_S32': 2, # Signed 32bit Event Data Type 'EVENT_DATA_TYPE_F32': 3, # Float 32bit Event Data Type 'EVENT_DATA_TYPE_BOOL': 4, # Boolean Event Data Type 'NUM_OF_EVENT_DATA_TYPES': 5, # Number of Event Data Types } return DialinEnum('FPEventDataTypes', members) # ================================================== Enum Creators: Names ================================================== def _create_fp_boost_pump_names(self): members = { 'P40_PUMP': 0, # FP Boost Pump 'NUM_OF_BOOST_PUMPS': 1, # Number of FP Boost Pumps } return DialinEnum('FPBoostPumpNames', members) def _create_fp_conductivity_sensor_names(self): members = { 'P9_COND': 0, # CPi (P9) conductivity sensor value 'P18_COND': 1, # CPo (P18) conductivity sensor value 'NUM_OF_FP_CONDUCTIVITY_SENSRORS': 2, # Number of Conductivity sensors } return DialinEnum('FPConductivitySensorNames', members) def _create_fp_ro_rejection_ratio_names(self): members = { 'RAW_RO_REJECTION_RATIO': 0, # Raw RO Rejection Ratio value 'RAW_RO_REJECTION_RATIO_TANK_FILL': 1, # Raw RO Rejection Ratio Tank Fill value 'AVG_RO_REJECTION_RATIO': 2, # Average RO Rejection Ratio value 'AVG_RO_REJECTION_RATIO_TANK_FILL': 3, # Average RO Rejection Ratio Tank Fill value 'GEN_PERMEATE_STATE': 4, # Generate Permeate State value 'NUM_OF_FP_RO_REJECTION_RATIO': 5, # Number of RO Rejection Ratio items } return DialinEnum('FPRORejectionRatioNames', members) def _create_fp_flow_sensor_names(self): members = { 'P7_FLOW': 0, # Water inlet flow sensor (FMS P7) 'P16_FLOW': 1, # RO outlet flow sensor (FMP P16) 'NUM_OF_FP_FLOW_SENSORS': 2, # Number of FP Flow Sensors } return DialinEnum('FPFlowSensorNames', members) def _create_fp_floater_sensor_names(self): members = { 'P25_LEVEL': 0, # P25 Level sensor 'NUM_OF_FP_LEVEL_SENSORS': 1, # Number of Level Sensors } return DialinEnum('FPFloaterLevelSensorNames', members) def _create_fp_pressure_sensor_names(self): if GlobalVariables().NEXT_GEN_HW == True: members = { 'M3_PRES' : 0, # Water inlet pressure before regulator 'P8_PRES' : 1, # Water inlet pressure after regulator 'P13_PRES' : 2, # Water inlet pressure before the conductivity sensor 'P17_PRES' : 3, # Pressure before the RO filter 'P46_PRES' : 4, # Pressure after the RO filter 'NUM_OF_FP_PRES_SENSORS' : 5, # Number of Pressure Sensors } else: members = { 'M3_PRES' : 0, # Water inlet pressure before regulator 'P8_PRES' : 1, # Water inlet pressure after regulator 'P13_PRES' : 2, # Water inlet pressure before the conductivity sensor 'P17_PRES' : 3, # Pressure before the RO filter 'NUM_OF_FP_PRES_SENSORS' : 4, # Number of Pressure Sensors } return DialinEnum('FPPressureSensorNames', members) def _create_fp_ro_pump_names(self): members = { 'P12_PUMP': 0, # P12 RO Pump 'NUM_OF_RO_PUMPS': 1, # Number of RO Pumps } return DialinEnum('FPROPumpNames', members) def _create_fp_temperature_sensor_names(self): if GlobalVariables().NEXT_GEN_HW == True: members = { 'M3_TEMP' : 0, # Temperature after pressure regulator (M3) 'P8_TEMP' : 1, # Temperature before inlet conductivity sensor (P8) 'P13_TEMP' : 2, # Temperature before RO filter (P13) 'P17_TEMP' : 3, # Temperature after RO filter (P17) 'P46_TEMP' : 4, # Temperature after sediment filter (P46) 'P10_TEMP' : 5, # P10 temperature from inlet conductivity sensor. 'P19_TEMP' : 6, # P19 temperature from outlet conductivity sensor. 'P7_TEMP' : 7, # Temperature at p7 flow meter 'P16_TEMP' : 8, # Temperature at p16 flow meter 'NUM_OF_TEMPERATURE_SENSORS' : 9, # Number of Temperature Sensors } else: members = { 'M3_TEMP' : 0, # Temperature after pressure regulator (M3) 'P8_TEMP' : 1, # Temperature before inlet conductivity sensor (P8) 'P13_TEMP' : 2, # Temperature before RO filter (P13) 'P17_TEMP' : 3, # Temperature after RO filter (P17) 'P10_TEMP' : 4, # P10 temperature from inlet conductivity sensor. 'P19_TEMP' : 5, # P19 temperature from outlet conductivity sensor. 'P7_TEMP' : 6, # Temperature at p7 flow meter 'P16_TEMP' : 7, # Temperature at p16 flow meter 'NUM_OF_TEMPERATURE_SENSORS' : 8, # Number of Temperature Sensors } # Create the enum created_enum = DialinEnum('FPTemperatureSensorNames', members) # Add aliases extend_enum(created_enum, 'FIRST_PRESSURE_SENSOR', created_enum.M3_TEMP.value) extend_enum(created_enum, 'LAST_PRESSURE_SENSOR', created_enum.P46_TEMP.value if GlobalVariables.NEXT_GEN_HW == True else created_enum.P17_TEMP.value) extend_enum(created_enum, 'FIRST_CONDUCTIVITY_SENSOR', created_enum.P10_TEMP.value) extend_enum(created_enum, 'LAST_CONDUCTIVITY_SENSOR', created_enum.P19_TEMP.value) extend_enum(created_enum, 'FIRST_FLOW_SENSOR', created_enum.P7_TEMP.value) extend_enum(created_enum, 'LAST_FLOW_SENSOR', created_enum.P16_TEMP.value) # Add aliases to be searchable (listed when pressed TAB) created_enum._member_names_.append('FIRST_PRESSURE_SENSOR') created_enum._member_names_.append('LAST_PRESSURE_SENSOR') created_enum._member_names_.append('FIRST_CONDUCTIVITY_SENSOR') created_enum._member_names_.append('LAST_CONDUCTIVITY_SENSOR') created_enum._member_names_.append('FIRST_FLOW_SENSOR') created_enum._member_names_.append('LAST_FLOW_SENSOR') return created_enum def _create_fp_valve_names(self): members = { 'M4_VALV': 0, # Valve (M4 VWi) 'M12_VALV': 1, # Valve (M12) 'P11_VALV': 2, # Valve (P11 VPi) 'P33_VALV': 3, # Valve (P33 VCr) 'P34_VALV': 4, # Valve (P34 VCb) 'P37_VALV': 5, # Valve (P37 VCd) 'P39_VALV': 6, # Valve (P39 VROd) 'P6_VALV': 7, # Valve (P6 VFF) 'NUM_OF_FP_VALVES': 8, # Number of valves } return DialinEnum('FPValveNames', members) def _create_fp_all_pump_names(self): members = { 'P12_PUMP_RO': 0, # P14 RO Pump 'P40_PUMP_BOOSTER': 1, # P40 Boost Pump } return DialinEnum('FPAllPumpNames', members) # ================================================== Enum Creators: Support ================================================== def _create_fp_floater_levels(self): members = { 'LEVEL_STATE_EMPTY': 0, # Empty level for the FP Floater Level Sensors 'LEVEL_STATE_LOW': 1, # Low level for the FP Floater Level Sensors 'LEVEL_STATE_MEDIUM': 2, # Medium level for the FP Floater Level Sensors 'LEVEL_STATE_HIGH': 3, # High level for the FP Floater Level Sensors 'LEVEL_STATE_ILLEGAL': 4, # Illegal level for FP Floater Level Sensor (indicates investigation) 'NUM_OF_LEVEL_STATES': 5, # Number of Level States } return DialinEnum('FPFloaterLevels', members) def _create_fp_pump_attributes(self): members = { 'STATE': 0, # State to command attribute for Pumps 'DUTY_CYCLE': 1, # Duty Cycle attribute for Pumps 'FB_DUTY_CYCLE': 2, # Feedback Duty Cycle attribute for Pumps 'SPEED': 3, # Current Speed attribute for pumps 'TARGET_PRES': 4, # Target Pressure attribute for pumps 'TARGET_FLOW': 5, # Target Flow attribute for pumps 'TARGET_DUTY_CYCLE': 6, # Target Duty Cycle attribute for pumps 'DUTY_CYCLE_PCT': 7, # Duty Cycle Percentage attribute for pumps 'FB_DUTY_CYCLE_PCT': 8, # Feedback Duty Cycle Percentage attribute for pumps 'NUM_OF_RO_PUMP_ATTRIBUTES': 9, # Number of pump attributes } return DialinEnum('FPPumpAttributes', members) def _create_fp_valve_states(self): members = { 'VALVE_STATE_CLOSED': 0, # Valve Closed State 'VALVE_STATE_OPEN': 1, # Valve Open State } return DialinEnum('FPValveStates', members) # ================================================== Creates a singleton ================================================== fp_enum_repository = FP_Defs() # Attach listener to update function test_config_change.add_listener(fp_enum_repository.update_enums) BETA_IOFP_COND_SENSOR_OFFSET = DDConductivitySensorsEnum.NUM_OF_CONDUCTIVITY_SENSORS.value # Offset to translate Dialin FPTemperaturesNames to DD firmware DD Conductivity Sensors enum BETA_IOFP_PRES_SENSOR_OFFSET = DDPressureNames.NUM_OF_PRESSURE_SENSORS.value # Offset to translate Dialin FPTemperaturesNames to DD firmware DD Pressure Sensors enum BETA_IOFP_VALVE_OFFSET = DDValveNames.NUM_OF_DD_VALVES.value # Offset to translate Dialin FPValveNames to DD firmware DD_Valves