########################################################################### # # Copyright (c) 2019-2020 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 load_cells.py # # @date 14-Apr-2020 # @author S. Nash # # @brief # # ############################################################################ import struct from ..utils.conversions import integer_to_bytearray, float_to_bytearray from .constants import RESET,NO_RESET from ..protocols.CAN import (DenaliMessage, DenaliChannels) from ..utils.base import _AbstractSubSystem, _publish class DGLoadCells(_AbstractSubSystem): """ DGLoadCells Dialysate Generator (DG) Dialin API sub-class for load cell related commands. """ # Pressure/Occlusion message IDs MSG_ID_DG_LOAD_CELL_DATA = 0x000C MSG_ID_DG_LOAD_CELL_OVERRIDE = 0xA005 MSG_ID_DG_LOAD_CELL_DATA_BROADCAST_INTERVAL_OVERRIDE = 0xA00D # Load cell broadcast message field positions START_POS_LC_A1 = DenaliMessage.PAYLOAD_START_INDEX END_POS_LC_A1 = START_POS_LC_A1 + 4 START_POS_LC_A2 = END_POS_LC_A1 END_POS_LC_A2 = START_POS_LC_A2 + 4 START_POS_LC_B1 = END_POS_LC_A2 END_POS_LC_B1 = START_POS_LC_B1 + 4 START_POS_LC_B2 = END_POS_LC_B1 END_POS_LC_B2 = START_POS_LC_B2 + 4 # Load Cell IDs LOAD_CELL_A1 = 0 LOAD_CELL_A2 = 1 LOAD_CELL_B1 = 2 LOAD_CELL_B2 = 3 def __init__(self, can_interface=None): """ @param can_interface: Denali Can Messenger object """ super().__init__() self.can_interface = can_interface if self.can_interface is not None: channel_id = DenaliChannels.dg_sync_broadcast_ch_id msg_id = self.MSG_ID_DG_LOAD_CELL_DATA self.can_interface.register_receiving_publication_function(channel_id, msg_id, self._handler_load_cells_sync) self.load_cell_A1 = 0.0 self.load_cell_A2 = 0.0 self.load_cell_B1 = 0.0 self.load_cell_B2 = 0.0 def get_load_cells(self): """ Gets the current load cell weights @return: List containing load cell values: [A1, A2, B1, B2] """ return [self.load_cell_A1, self.load_cell_A2, self.load_cell_B1, self.load_cell_B2] @_publish(["load_cell_A1", "load_cell_A2", "load_cell_B1", "load_cell_B2"]) def _handler_load_cells_sync(self, message): """ Handles published load cell data messages. Load cell data are captured for reference. @param message: published load cell data message @return: None """ a1 = struct.unpack('f', bytearray( message['message'][self.START_POS_LC_A1:self.END_POS_LC_A1])) a2 = struct.unpack('f', bytearray( message['message'][self.START_POS_LC_A2:self.END_POS_LC_A2])) b1 = struct.unpack('f', bytearray( message['message'][self.START_POS_LC_B1:self.END_POS_LC_B1])) b2 = struct.unpack('f', bytearray( message['message'][self.START_POS_LC_B2:self.END_POS_LC_B2])) self.load_cell_A1 = a1[0] self.load_cell_A2 = a2[0] self.load_cell_B1 = b1[0] self.load_cell_B2 = b2[0] def cmd_load_cell_override(self, grams, sensor, reset=NO_RESET): """ Constructs and sends the load cell override command Constraints: Must be logged into DG. Given sensor must be one of the sensors listed below. @param grams: float - weight (in grams) to override sensor with @param sensor: unsigned int - sensor ID @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise Load Cell sensor IDs: \n 0 = A1 \n 1 = A2 \n 2 = B1 \n 3 = B2 \n """ rst = integer_to_bytearray(reset) grm = float_to_bytearray(grams) idx = integer_to_bytearray(sensor) payload = rst + grm + idx message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_dg_ch_id, message_id=self.MSG_ID_DG_LOAD_CELL_OVERRIDE, payload=payload) print("override load cell weight value for sensor " + str(sensor)) # Send message received_message = self.can_interface.send(message) # If there is content... if received_message is not None: # response payload is OK or not OK return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] else: print("Timeout!!!!") return False def cmd_load_cell_data_broadcast_interval_override(self, ms, reset=NO_RESET): """ Constructs and sends the load cell data broadcast interval override command Constraints: Must be logged into DG. Given interval must be non-zero and a multiple of the DG priority task interval (10 ms). @param ms: integer - interval (in ms) to override with @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) mis = integer_to_bytearray(ms) payload = rst + mis message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_dg_ch_id, message_id=self.MSG_ID_DG_LOAD_CELL_DATA_BROADCAST_INTERVAL_OVERRIDE, payload=payload) print("override DG load cell broadcast interval") # Send message received_message = self.can_interface.send(message) # If there is content... if received_message is not None: if reset == RESET: str_res = "reset back to normal: " else: str_res = str(ms) + " ms: " print("Load cell data broadcast interval overridden to " + str_res + str(received_message['message'][DenaliMessage.PAYLOAD_START_INDEX])) # response payload is OK or not OK return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] else: print("Timeout!!!!") return False