Skip to content
Snippets Groups Projects
Select Git revision
  • 12a2097b410139eae70dbd031a916b9d3a4487ff
  • main default protected
  • experiment-step-sharing
  • experiment-color-coded-graph
  • experiment-separated-graph
5 results

cwl_parsing.py

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    apspu_lib.py 22.22 KiB
    """
    Copyright 2021 Stichting Nederlandse Wetenschappelijk Onderzoek Instituten,
    ASTRON Netherlands Institute for Radio Astronomy
    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at
     http://www.apache.org/licenses/LICENSE-2.0
    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
    
    
      Created: 2022-17-12
    This file contains the APSPU class with all monitoring function. It can be used stand-alone to test it.
    
    """
    
    import sys
    sys.path.insert(0, '.')
    import os
    import math
    from APSPU_I2C import *
    
    
    if 0: #os.name == "posix":
        from I2C_serial_pi3 import *
    elif 1:
        from I2C_serial_iss import *
    else:
        from I2C_serial import *
    
    I2CBUSNR = 1    # Bus used on the Pi
    DEBUG = False   # Set True to print debug information on the screen
    I_OK_MARGIN = 0.5 # I_error in Amps
    
    class ApspuClass:
        #
        # Toplevel Class that contains all parts on a APSU
        #
        def __init__(self):
            self.status = False
            self.pols = []
            for pol in list(CTR_POLS.keys()):
                self.pols.append(PolClass(pol))
            self.fans = FanmonitorClass()
            self.eeprom = EepromClass()
    
        def read_all(self):
            #
            # Function to read all monitoring points of the APSPU
            #
            print("--------- \nRead APSPU Status\n---------")
            for pol in self.pols:
                pol.read_all()
            self.fans.read_all()
            return True
    
        def print_status(self):
            #
            # Function to dump monitoring information on the screen
            #
            for pol in self.pols:
                pol.print_status()
            self.fans.print_status()
            return True
    
        def set_pols(self):
            #
            # Function to set output voltage level on the APSPU
            # Values are read from the APSPU_I2C
            #
            print("--------- \nProgram Pols\n---------")
            for pol in self.pols:
                print(f"Set Pol {pol.name} to {VOUT_POLS[pol.name]:4.2f} V")
                pol.set_vout_pol(VOUT_POLS[pol.name])
                vout = pol.read_vout_set()
                if not (0.9*VOUT_POLS[pol.name] < vout < 1.1*VOUT_POLS[pol.name]):
                    print(f"POL {pol.name:10} Error setting Vout, "
                          f"set to {VOUT_POLS[pol.name]} read back {vout}")
                    exit()
                pol.set_vout_ov_limit_pol(1.2*VOUT_POLS[pol.name])
                ov_out = pol.read_ov_limit()
                if not (1.1*VOUT_POLS[pol.name] < ov_out < 1.3*VOUT_POLS[pol.name]):
                    print(f"POL {pol.name:10} Error setting output overvoltage"
                          f"set {1.2*VOUT_POLS[pol.name]} read back {ov_out}")
                    exit()
                pol.set_on_off_config()
                pol.on_off(True)
                pol.write_to_nvm()
            print("Done")
        
        def check_apspu(self):
            #
            # Function to check values of read_all() Used during production
            #
            # Return True is OK, False if not OK
            #
            print("--------- \nCheck APSPU \n---------")
            check_ok = True
            for pol in self.pols:
                check_ok = check_ok & pol.check_pol()
            check_ok = check_ok & self.fans.check_fans()
            check_ok = check_ok & self.eeprom.wr_rd_eeprom(value="PROD_CHECK", address=0x30)
            if check_ok:
                print("APSPU OK")
            else:
                print(">>> APSPU NOT OK <<<")
            return check_ok
     
        def apspu_on_off(self, on):
            #
            # Function to switch off the POLs on APSPU
            # on = True to switch APSU on
            # on = False to switch APSU off
            # Return: always True
            #
            if on:
                print(f">> Switch APSPU ON")
            else:
                print(">> Switch APSPU OFF")
            for pol in self.pols:
                pol.on_off(on)
            return True
    
    
    class EepromClass:
        #
        # Class to handle EEPROM communication 
        #
        def __init__(self):
            #
            # All monitor. read and write functions for the EEPROM
            #
            self.dev_i2c_eeprom = I2C(EEPROM)
            self.dev_i2c_eeprom.bus_nr = I2CBUSNR
    
        def write_eeprom(self, data="APSPU", address=0):
            #
            # Write the EEPROM with the serial number etc.
            #
            # Data = data to write in string formal
            # Address = address to write the data to
            # Return True if successfully
            #
            ret_ack, ret_value = self.dev_i2c_eeprom.read_bytes(0)
            if ret_ack < 1:
                print("EEPROM not found during write")
                return False
            else:
                wr_data = bytearray(data.encode("utf-8", errors="ignore"))
                for loc, data_byte in enumerate(wr_data):
                    self.dev_i2c_eeprom.write_bytes(address+loc, data_byte)
                    sleep(0.1)
                return True
    
        def read_eeprom(self, address=0, nof_bytes=5):
            #
            # Read the EEPROM with the serial number etc.
            #
            # Address = address to read from
            # nof_bytes = number of bytes to read
            # return string with the data from the flash
            #
            ret_ack, ret_value = self.dev_i2c_eeprom.read_last_reg(1)
            if ret_ack < 1:
                print("no EEPROM found during read")
                return False
            else:
                ret_ack, ret_value = self.dev_i2c_eeprom.read_bytes(address, nof_bytes)
                str_return = bytes.fromhex(ret_value[:nof_bytes*2]).decode('UTF-8')
                return str_return
    
        def wr_rd_eeprom(self, value="APSPU-1", address=0):
            #
            # Write and Read the EEPROM to check functionality
            #
            # value = string with data to write
            # address = address to write the data to
            # return True if read back is same as write value
            #
            if self.write_eeprom(value, address=0):
                ret_value = self.read_eeprom(address=0, nof_bytes=len(value))
                stri = "Wrote to EEPROM register 0x{2:x} : {0}, Read from EEPROM: {1}".format(value, ret_value, address)
                print(stri)
                if ret_value == value:
                    return True
                else:
                    return False
    
    
    class PolClass:
        #
        # Class to read all monitoring points Point of Load DC/DC converter
        #
        def __init__(self, name):
            #
            # All monitoring points Point of Load DC/DC converter
            #
            self.name = name
            self.vout = 0
            self.vin = 0
            self.iout = 0
            self.temp = 0
            self.pol_dev = I2C(CTR_POLS[self.name])
            self.pol_dev.bus_nr = I2CBUSNR
            ret_ack, ret_value = self.pol_dev.read_bytes(1)
            if ret_ack < 1:
                stri = " Device {0} at address 0x{1:X} not found".format(self.name, CTR_POLS[self.name])
                print(stri)
                self.status = False
            else:
                self.status = True
    
        def read_vout_set(self):
            #
            # Function to read the output voltage of the Point of Load DC/DC converter
            #
            # Return: output value is of else 999
            #
            if self.status:
                ret_ack, raw_value = self.pol_dev.read_bytes(LP_VOUT_COMMAND, 2)
                if len(raw_value) < 4:
                    raw_value = '0' + raw_value
                ret_value = int(raw_value[2:], 16) * 2**8
                ret_value += int(raw_value[:2], 16)
                output_value = ret_value * 2**-11
                if DEBUG:
                    print(f"Output set to: = {output_value:5.2f} V using hex value {raw_value}")
                return output_value
            else:
                return 999
    
        def set_on_off_config(self):
            #
            # Function to set the output of the DC/DC converter
            #
            # Return always I2C ack
            #
            ret_ack = False
            if self.status:
                if DEBUG:
                    ret_ack, raw_value = self.pol_dev.read_bytes(LP_ON_OFF_CONFIG, 1)
                    print(f"Current setting ON/OFF register 0x{int(raw_value, 16):02x}")
                on_off_bit = not 1 << 0
                polarity_pin = 1 << 1
                use_external = 0 << 2
                use_soft = 1 << 3
                default_off = 1 << 4
                wr_data = on_off_bit + polarity_pin + use_external + use_soft + default_off
                ret_ack = self.pol_dev.write_bytes(LP_ON_OFF_CONFIG, wr_data)
            return ret_ack
    
        def on_off(self, on=True):
            #
            # Function to switch on or off a POL
            #
            # if on = True switch on the POL
            # if on = Flase swtich off the POL
            # Return I2C ack
            #
            wr_data = (on << 7) + 0
            ret_ack = self.pol_dev.write_bytes(LP_OPERATION, wr_data)
            return ret_ack
    
        def set_vout_pol(self, value):
            #
            # Function to read the output voltage of the Point of Load DC/DC converter
            #
            # value is the output voltage level in V
            # return I2C ack
            #
            ret_ack = False
            if self.status:
                ret_ack = self.pol_dev.write_bytes(LP_WRITE_PROTECT, 0)
                if not ret_ack:
                    print("Error set Write Protect")
                set_value = int(value * (2**11))
                hex_set_value = hex(set_value)
                wr_value = (hex_set_value[4:6] + hex_set_value[2:4])
                if DEBUG:
                    print(f"Calculated wr_value is {wr_value}")
                wr_data = []
                wr_data.append(int(hex_set_value[4:6], 16))
                wr_data.append(int(hex_set_value[2:4], 16))
                ret_ack = self.pol_dev.write_bytes(LP_VOUT_COMMAND, wr_data)
                if not ret_ack:
                    print("Error write data to POL")
            return ret_ack
    
        def set_vout_ov_limit_pol(self, value):
            #
            # Function to read the output voltage of the Point of Load DC/DC converter
            #
            # value is the overvoltage level of the output
            # return I2C ack
            #
            ret_ack = False
            if self.status:
                set_value = int(value * (2**11))
                hex_set_value = hex(set_value)
                wr_value = (hex_set_value[4:6] + hex_set_value[2:4])
                if DEBUG:
                    print(f"Calculated wr_value is {wr_value}")
                wr_data = []
                wr_data.append(int(hex_set_value[4:6], 16))
                wr_data.append(int(hex_set_value[2:4], 16))
                ret_ack = self.pol_dev.write_bytes(LP_VOUT_OV_LIMIT, wr_data)
            return ret_ack
        
        def write_to_nvm(self):
            #
            # Function to write the POL's registers to NVM memory
            #
            # return is always True
            #
            print(f"Store to NVM for POL {self.name}")
            if True:
                ret_ack = self.pol_dev.write_pointer(0x15)
                sleep(1)
            else:
                self.pol_dev.close()
                command = f"i2cset -y 1 0x{CTR_POLS[self.name]:02X} 0x15 cp" 
                os.system(command)
                os.system(command)
                self.pol_dev.open()
            return True
    
        def read_vin(self):
            #
            # Function to read the output voltage of the Point of Load DC/DC converter
            #
            if self.status:
                ret_ack, raw_value = self.pol_dev.read_bytes(LP_VIN, 2)
                if not ret_ack:
                    self.iout=999
                    return False
                if len(raw_value) < 4:
                    raw_value = '0' + raw_value
                ret_value = []
                ret_value.append(int(raw_value[:2], 16))
                ret_value.append(int(raw_value[2:], 16))
                output_value = calc_lin_2bytes(ret_value)
                self.vin = output_value
                return True
    
        def read_vout(self):
            #
            # Function to read the output voltage of the Point of Load DC/DC converter
            #
            if self.status:
                ret_ack, vout_mod = self.pol_dev.read_bytes(LP_VOUT_MODE, 1)
                if not ret_ack:
                    self.vout=999
                    return False
                ret_ack, raw_value = self.pol_dev.read_bytes(LP_VOUT, 2) 
                if not ret_ack:
                    self.vout=999
                    return False
                vout_mod = int(vout_mod, 16)
                ret_value = []
                ret_value.append(int(raw_value[:2], 16))
                try:
                    ret_value.append(int(raw_value[2:], 16))
                except:
                    ret_value.append(0)
                self.vout = calc_lin_3bytes(ret_value, [vout_mod])
            else:
                self.vout = 999
            return self.vout
    
        def read_ov_limit(self):
            #
            # Function to read the output voltage of the Point of Load DC/DC converter
            # and print on the screen
            #
            # Return OV limit
            #
            output_value = 0
            if self.status:
                ret_ack, raw_value = self.pol_dev.read_bytes(LP_VOUT_OV_LIMIT, 2)
                if len(raw_value) < 4:
                    raw_value = '0' + raw_value
                ret_value = int(raw_value[2:], 16) * 2**8
                ret_value += int(raw_value[:2], 16)
                output_value = ret_value * 2**-11
                if DEBUG:
                    print(f"Output OV limit is set to: = {output_value:5.2f} V using hex value {raw_value}")
            return output_value
    
        def read_uv_limit(self):
            #
            # Function to read the output voltage of the Point of Load DC/DC converter
            #
            # Return UV limit if OK else False
            #
            if self.status:
                ret_ack, raw_value = self.pol_dev.read_bytes(LP_VOUT_UV_LIMIT, 2)
                if len(raw_value) < 4:
                    raw_value = '0' + raw_value
                ret_value = []
                ret_value = int(raw_value[2:], 16) * 2**8
                ret_value += int(raw_value[:2], 16)
                output_value = ret_value * 2**-11
                if DEBUG:
                    print(f"Output UV limit is set to: = {output_value:5.2f} V using hex value {raw_value}")
                return output_value
            else:
                return False
    
        def read_iout(self):
            #
            # Function to read the output current of the Point of Load DC/DC converter
            #
            if self.status:
                ret_ack, raw_value = self.pol_dev.read_bytes(0x39, 2)
                ret_ack, raw_value = self.pol_dev.read_bytes(LP_IOUT, 2)
                ret_value = []
                ret_value.append(int(raw_value[:2], 16))
                ret_value.append(int(raw_value[2:], 16))
                self.iout = calc_lin_2bytes(ret_value)
            else:
                self.iout = 999
            return self.iout
    
        def read_temp(self):
            #
            # Function to read the temperature of the Point of Load DC/DC converter
            #
            if self.status:
                ret_ack, raw_value = self.pol_dev.read_bytes(LP_temp, 2)
                ret_value = []
                ret_value.append(int(raw_value[:2], 16))
                ret_value.append(int(raw_value[2:], 16))
                self.temp = calc_lin_2bytes(ret_value)
            else:
                self.temp = 999
            return self.temp
    
        def read_all(self):
            #
            # Function to read all monitoring points of the Point of Load DC/DC converter
            #
            self.read_vout()
            self.read_iout()
            self.read_temp()
            self.read_vin()
    
        def check_pol(self):
            #
            # Function to read all monitoring points of the Point of Load DC/DC converter
            #
            # Return True if OK, False if not OK
            #
            self.read_all()
            self.on_off(on=True)
            check_ok = False
            expected_vout = VOUT_POLS[self.name]
            if 0.9*expected_vout < self.vout < 1.1*expected_vout:
                check_ok = True
            else:
                check_ok = False
                print(f"POL {self.name:10} Vout not OK, expected {expected_vout} V, measured {self.vout} V")
                return check_ok
            vin_low = 45
            vin_high = 50
            if vin_low < self.vin < vin_high:
                check_ok = True
            else:
                check_ok = False
                print(f"POL {self.name:10} Vin not OK, expected {vin_low} V - {vin_high} V, measured {self.vout} V ")
                return check_ok
            temp_low = 20
            temp_high = 50
            if temp_low < self.temp < temp_high:
                check_ok = True
            else:
                check_ok = False
                print(f"POL {self.name:10} TEMP not OK, expected {temp_low} C - {temp_high} C, measured {self.temp} C ")
                return check_ok
            i_low = (1-I_OK_MARGIN)*IOUT_POLS[self.name]
            i_high = (1+I_OK_MARGIN)*IOUT_POLS[self.name]
            if i_low < self.iout < i_high:
                check_ok = True
            else:
                check_ok = False
                print(f"POL {self.name:10} Iout not OK,"
                      f" expected {i_low:4.2f} A - {i_high:4.2f} A, measured {self.iout:4.2f} A ")
            return check_ok
    
        def print_status(self):
            #
            # Function to dump all monitoring points of the Point of Load DC/DC converter on the screen
            #
            if self.status:
                stri = f"POL: {self.name:10} :"
                stri += "Vin :{0: >5.2f} V ".format(self.vin)
                stri += "Vout :{0: >5.2f} V ".format(self.vout)
                stri += "Iout :{0: >5.2f} A ".format(self.iout)
                stri += "Temp :{0: >5.2f} \N{DEGREE SIGN}C".format(self.temp)
                print(stri)
                self.read_vout_set()
    
    
    class FanmonitorClass:
        #
        # Class to read all monitoring points fan units in the APS
        #
        def __init__(self):
            #
            # All monitoring points for the fans
            #
            self.rpm = []
            self.fanmonitor_dev = I2C(MAX6620)
            self.fanmonitor_dev.bus_nr = I2CBUSNR
            ret_ack, ret_value = self.fanmonitor_dev.read_bytes(1)
            if ret_ack < 1:
                stri = " Device {0} at address 0x{1:X} not found".format("MAX6620", MAX6620)
                print(stri)
                self.status = False
            else:
                if DEBUG:
                    stri = "Device {0} at address 0x{1:X} is found ".format("MAX6620", MAX6620)
                    print(stri)
                try:
                    self.set_active()
                except:
                    self.status = False
                else:
                    self.status = True
    
        def set_active(self):
            #
            # Function to activate monitoring
            #
            ret_ack, reg_before = self.fanmonitor_dev.read_bytes(REG_GLOBAL, 1)
            self.fanmonitor_dev.write_bytes(REG_GLOBAL, RUN_MONITOR)
            ret_ack, reg_after = self.fanmonitor_dev.read_bytes(REG_GLOBAL, 1)
            if DEBUG:
                stri = "Reg at address 0x{0} before : {1} and after write action {2}"\
                    .format(REG_GLOBAL, reg_before, reg_after)
                print(stri)
            fan_config_reg = int((math.log(TACH_PERIODS) / math.log(2))) << 5
            for fan_cnt in range(NOF_APS_FANS):
                self.fanmonitor_dev.write_bytes(0x02 + fan_cnt, 0x88)
                self.fanmonitor_dev.write_bytes(0x06 + fan_cnt, fan_config_reg)
    
        def read_fan(self, fan_nr):
            #
            # Function to a single fan
            # fan_nr is the fan to read ranging from 0 till 2
            # return the speed of the fan
            #
            if fan_nr > NOF_APS_FANS:
                return 0
            ret_ack, tach_msb = self.fanmonitor_dev.read_bytes(REG_TACH_MSP_REGS[fan_nr], 1)
            tach_msb = int(tach_msb, 16) & 0xFF
            tach_lsb = 255
            tach = 99999
            if tach_msb > 254:
                rpm = 0
            else:
                ret_ack, tach_lsb = self.fanmonitor_dev.read_bytes(REG_TACH_LSP_REGS[fan_nr], 1)
                tach_lsb = int(tach_lsb, 16) & 0xE0
                tach = tach_msb*16 + tach_lsb/8
                rpm = float((TACH_COUNT_FREQ*TACH_PERIODS*60))/(FAN_TACHS*tach)
            if DEBUG:
                stri = "MSP: {0}, LSB: {1}, TACH : {2}".format(tach_msb, tach_lsb, tach)
                print(stri)
            return rpm
        
        def read_all(self):
            #
            # Function to read all fan's
            #
            self.rpm = []
            for fan_counter in range(NOF_APS_FANS):
                self.rpm.append(self.read_fan(fan_counter))
                
        def check_fans(self):
            #
            # Function to check fan speeds
            #
            self.read_all()
            check_ok = True
            for cnt, speed in enumerate(self.rpm):
                if speed < 10:
                    print(f"Low speed on fan {cnt}")
                    check_ok = False
            return check_ok
    
        def print_status(self):
            #
            # Function to dump all monitoring points of the Point of Load DC/DC converter on the screen
            #
            if self.status:
                stri = "Fan speeds of "
                for fan_cnt in range(len(self.rpm)):
                    stri += "FAN_" + str(fan_cnt+1)
                    stri += " :{0: <5.2f} RPM ".format(self.rpm[fan_cnt])
                print(stri)
    
    
    def temperature(pol=0):
        apspu = ApspuClass()
        measured_temperature = apspu.pols[pol].read_temp()
        print(f"Temparature of POL {apspu.pols[pol].name} is {measured_temperature:4.2f} deg C")
        return measured_temperature
    
    def voltage(pol=0):
        apspu = ApspuClass()
        measured_voltage = apspu.pols[pol].read_vout()
        print(f"Output voltage of POL {apspu.pols[pol].name} is {measured_voltage:4.2f} V")
        return measured_voltage
    
    def current(pol=0):
        apspu = ApspuClass()
        measured_current = apspu.pols[pol].read_iout()
        print(f"Output current of POL {apspu.pols[pol].name} is {measured_current:4.2f} A")
        return measured_current
    
    def apspu_off():
        apspu = ApspuClass()
        apspu.apspu_on_off(False)
        print(f"All outputs off")
    
    def apspu_on():
        apspu = ApspuClass()
        apspu.apspu_on_off(True)
        print(f"All outputs on")
    
    def fan_speed(fan_nr):
        apspu = ApspuClass()
        fan_speed = apspu.fans.read_fan(fan_nr)
        print(f"Fan nr {fan_nr} has speed {fan_speed}")
        return fan_speed
    
    def program_dc_dc_converters():
        apspu = ApspuClass()
        apspu.set_pols()
        print("Set DC/DC converters on APSPU")
    
    def main():
        #
        # Function to test the class, read all info and dump on the screen
        #
        user = "Henri"
        apspu = ApspuClass()
        if user == "Gijs":
            apspu.apspu_on_off(False)
            sleep(5)
            apspu.set_pols()
            apspu.apspu_on_off(True)
            sleep(10)
            apspu.read_all()
            apspu.print_status()
            apspu.check_apspu()
            apspu.eeprom.wr_rd_eeprom(value="APSPU-2", address=0)
        elif user == "Henri":
            apspu_off()
            program_dc_dc_converters()
            apspu_on()
            for pol_cnt in range(3):
                temperature(pol_cnt)
                voltage(pol_cnt)
                current(pol_cnt)
            for fan_cnt in range(3):
                fan_speed(fan_cnt)
            apspu_off()
    
        else:
            print("Ok")
    
    if __name__ == "__main__":
        main()