Hilfe gesucht, Erklärung einer Funktion in Python

Humpelschlumpf · Mar 14th 2023

Hallo zusammen,

das ist mein erster Beitrag.

Mein Name ist Andre, bin 46 Jahre (alter Sack) und beschäftige mich hobbymäßig mit Heimautomatisierung auf dem Raspi.

Was hab ich vor:

Ich will mittels Raspberry ein paar Umweltsensoren auslesen und auf dem Display grafisch darstellen.

Wenn alles läuft binde ich noch meine Fritzbox mit ein und steuere darüber die Thermostate.

Soweit so gut.

Ich bin allerdings auf CODESYS für Raspberry (Soft-SPS) unterwegs und habe dementsprechend von Python nicht sonderliche Ahnung.

Derzeit bin ich dabei, mir ein Gerätetreiber für ein ICP10125 zu schreiben.

Das ist ein barometrischer Luftdrucksensor, der über den I2C-Bus läuft.

Für den ICP10125 gibt es ein Python-Gerätetreiber.

In diesem ist eine Funktion enthalten, über diese alle Lese und Schreibbefehle zu den Registern des Chips gehen.

Nur leider verstehe ich diese Funktion nicht.

Es geht hier um die Funktion rdwr()

Kann mir jemand erklären wie die funktioniert?

Ich hatte zunächst versucht, die CHIP_ID zu lesen.

Da ich aber nicht weis, auf welches Register, ich welche Byte-Reihenfolge senden muss und

welches Register ich anschließend lesen muss, sagt der ICP demzufolge nö mit dir rede ich nicht.

PS: ansprechen kann ich den Chip, da ich gewisse Daten auf die Adresse 16#63 (0x63) senden kann.

Über Hilfe wäre ich äußerst Dankbar.

Gruß

Andre

Python

import time
import struct
from smbus2 import SMBus, i2c_msg

__version__ = '0.0.1'

NORMAL = 0x6825
LOW_POWER = 0x609C
LOW_NOISE = 0x70DF
ULTRA_LOW_NOISE = 0x7866
NORMAL_T_FIRST = 0x6825
NORMAL_P_FIRST = 0x48A3
LOW_POWER_T_FIRST = 0x609C
LOW_POWER_P_FIRST = 0x401A
LOW_NOISE_T_FIRST = 0x70DF
LOW_NOISE_P_FIRST = 0x5059
ULN_T_FIRST = 0x7866
ULN_P_FIRST = 0x58E0

SOFT_RESET = 0x805D
READ_ID = 0xEFC8
MOVE_ADDRESS_PTR = 0xC595
READ_OTP = 0xC7F7

DEFAULT_I2C_ADDRESS = 0x63
CHIP_ID = 0x08

OK = 0
CRC_FAIL = 1

MEASUREMENT_DELAYS = {
    NORMAL: 7,           # 5.6 to 6.3ms
    LOW_POWER: 2,        # 1.6 to 1.8ms
    LOW_NOISE: 24,       # 20.8 to 23.8ms
    ULTRA_LOW_NOISE: 95  # 83.2 to 94.5ms
}


class ICP10125:
    sensor_constants = [0, 0, 0, 0]

    def __init__(self, bus_nr, address=None):
        if address is None:
            address = DEFAULT_I2C_ADDRESS
        self.address = address

        self.bus = SMBus(bus_nr)

        chip_id = self.chip_id()
        if chip_id != CHIP_ID:
            raise RuntimeError("ICP10125: Invalid Chip ID {:02x}, expected: {:02x}".format(chip_id, CHIP_ID))

        self.read_otp()

    def rdwr(self, command, length=0, delay=0):
        if type(command) is int:
            msg_w = i2c_msg.write(self.address, struct.pack(">H", command))
        else:
            msg_w = i2c_msg.write(self.address, command)

        self.bus.i2c_rdwr(msg_w)

        time.sleep(delay / 1000.0)

        if length > 0:
            msg_r = i2c_msg.read(self.address, length)
            self.bus.i2c_rdwr(msg_r)
            if length == 1:
                return list(msg_r)[0]
            else:
                result = list(msg_r)
                data = []
                for chunk in range(0, len(result), 3):
                    if self.crc8(result[chunk:chunk + 2]) != result[chunk + 2]:
                        raise ValueError("ICP10125: Invalid CRC8 in response.")
                    data.append((result[chunk] << 8) | result[chunk + 1])
                if len(data) == 1:
                    return data[0]
                else:
                    return data

        return []

    def chip_id(self):
        result = self.rdwr(READ_ID, 3)
        return result & 0x3f

    def read_otp(self):
        move_address_ptr = [
            MOVE_ADDRESS_PTR >> 8, MOVE_ADDRESS_PTR & 0xff,
            0x00, 0x66, 0x9c  # Address CRC8
        ]
        self.rdwr(move_address_ptr)

        for x in range(4):
            self.sensor_constants[x] = self.rdwr(READ_OTP, 3)

    def reset(self):
        self.rdwr(SOFT_RESET, delay=0.1)

    def measure(self, measure_command=NORMAL):
        delay = MEASUREMENT_DELAYS[measure_command]

        result = self.rdwr(measure_command, 9, delay)

        temperature = result[0]
        pressure = (result[1] << 8) | (result[2] >> 8)

        self.process_data(pressure, temperature)

        return self.pressure, self.temperature

    def calculate_altitude(self, pressure, qnh=1013.25):
        return 44330.0 * (1.0 - pow(pressure / qnh, 1.0 / 5.255))

    def process_data(self, p_LSB, T_LSB):
        LUT_lower = 3.5 * (1 << 20)
        LUT_upper = 11.5 * (1 << 20)
        quadr_factor = 1.0 / 16777216.0
        offst_factor = 2048.0

        t = T_LSB - 32768
        s1 = LUT_lower + (self.sensor_constants[0] * t * t) * quadr_factor
        s2 = offst_factor * self.sensor_constants[3] + (self.sensor_constants[1] * t * t) * quadr_factor
        s3 = LUT_upper + (self.sensor_constants[2] * t * t) * quadr_factor

        A, B, C = self.calculate_conversion_constants([s1, s2, s3])

        self.pressure = A + B / (C + p_LSB)
        self.temperature = -45.0 + 175.0 / 65536.0 * T_LSB

    def calculate_conversion_constants(self, p_LUT):
        p_Pa = [45000.0, 80000.0, 105000.0]

        C = (p_LUT[0] * p_LUT[1] * (p_Pa[0] - p_Pa[1]) +   # noqa: W504
        p_LUT[1] * p_LUT[2] * (p_Pa[1] - p_Pa[2]) +        # noqa: W504
        p_LUT[2] * p_LUT[0] * (p_Pa[2] - p_Pa[0])) / (p_LUT[2] * (p_Pa[0] - p_Pa[1]) +   # noqa: W504
        p_LUT[0] * (p_Pa[1] - p_Pa[2]) +                   # noqa: W504
        p_LUT[1] * (p_Pa[2] - p_Pa[0]))
        A = (p_Pa[0] * p_LUT[0] - p_Pa[1] * p_LUT[1] - (p_Pa[1] - p_Pa[0]) * C) / (p_LUT[0] - p_LUT[1])
        B = (p_Pa[0] - A) * (p_LUT[0] + C)

        return A, B, C

    def crc8(self, data, polynomial=0x31):
        result = 0xff
        for byte in data:
            result ^= byte
            for bit in range(8):
                if result & 0x80:
                    result <<= 1
                    result ^= polynomial
                else:
                    result <<= 1
        return result & 0xff

Display More

Tell · Mar 14th 2023

> Da ich aber nicht weis, auf welches Register, ich welche Byte-Reihenfolge senden muss

> und welches Register ich anschließend lesen muss

Das musst du gar nicht wissen, die Methode chip_id macht das fuer dich.

__blackjack__ · Mar 14th 2023

Humpelschlumpf An der Klasse ist schon mal falsch das `sensor_constants` eine Klassenvariable und damit global ist und für alle Exemplare gilt.

Dann werden `temperature` und `pressure` ausserhalb der `__init__()` eingeführt und durch `measure()` sowohl gesetzt als auch zurückgegeben. Das ist wirr und verwirrend.

`calculate_altitude()` steckt als reine Funktion in der Klasse. Wird aber auch gar nicht verwendet. Auch `crc8` hat als Funktion in der Klasse nichts zu suchen.

Die `rdwr()`-Methode ist sowohl was die Argumente als auch die Rückgabewerte angeht ein bisschen sehr flexibel. Für meinen Geschmack zu flexibel. Mindestens der Rückgabewert sollte konsistent eine Liste mit Zahlen sein, statt manchmal auch eine einzelne Zahl.

Was genau verstehst Du denn nicht? Mal davon abgesehen, dass die IMHO zu viel macht, ist das was sie macht, ja nicht so besonders kompliziert.

Ungetesteter, überarbeiteter Quelltext:

Python

import struct
import time

from smbus2 import SMBus, i2c_msg

__version__ = "0.0.1"

NORMAL = 0x6825
LOW_POWER = 0x609C
LOW_NOISE = 0x70DF
ULTRA_LOW_NOISE = 0x7866
NORMAL_T_FIRST = 0x6825
NORMAL_P_FIRST = 0x48A3
LOW_POWER_T_FIRST = 0x609C
LOW_POWER_P_FIRST = 0x401A
LOW_NOISE_T_FIRST = 0x70DF
LOW_NOISE_P_FIRST = 0x5059
ULN_T_FIRST = 0x7866
ULN_P_FIRST = 0x58E0

SOFT_RESET = 0x805D
READ_ID = 0xEFC8
MOVE_ADDRESS_PTR = 0xC595
READ_OTP = 0xC7F7

DEFAULT_I2C_ADDRESS = 0x63
CHIP_ID = 0x08

MEASUREMENT_DELAYS = {
    NORMAL: 7,  # 5.6 to 6.3ms
    LOW_POWER: 2,  # 1.6 to 1.8ms
    LOW_NOISE: 24,  # 20.8 to 23.8ms
    ULTRA_LOW_NOISE: 95,  # 83.2 to 94.5ms
}


def crc8(data, polynomial=0x31):
    result = 0xFF
    for byte in data:
        result ^= byte
        for _ in range(8):
            result <<= 1
            if result & 0x100:
                result ^= polynomial
    return result & 0xFF


class ICP10125:
    def __init__(self, bus_number, address=DEFAULT_I2C_ADDRESS):
        self.address = address
        self.bus = SMBus(bus_number)

        chip_id = self.chip_id()
        if chip_id != CHIP_ID:
            raise RuntimeError(
                f"ICP10125: Invalid Chip ID {chip_id:02x}, "
                f"expected: {CHIP_ID:02x}"
            )

        self.sensor_constants = self.read_otp()

    def rdwr(self, command, length=0, delay=0):
        if isinstance(command, int):
            message = i2c_msg.write(self.address, struct.pack(">H", command))
        else:
            message = i2c_msg.write(self.address, command)

        self.bus.i2c_rdwr(message)

        time.sleep(delay / 1000)

        if length > 0:
            message = i2c_msg.read(self.address, length)
            self.bus.i2c_rdwr(message)
            if length == 1:
                return message.buf[0]
            else:
                data = []
                result = list(message)
                for index in range(0, len(result), 3):
                    if crc8(result[index : index + 2]) != result[index + 2]:
                        raise ValueError("ICP10125: Invalid CRC8 in response.")
                    data.append((result[index] << 8) | result[index + 1])

                if len(data) == 1:
                    return data[0]
                else:
                    return data

        return []

    def chip_id(self):
        return self.rdwr(READ_ID, 3) & 0x3F

    def read_otp(self):
        self.rdwr(
            [
                MOVE_ADDRESS_PTR >> 8,
                MOVE_ADDRESS_PTR & 0xFF,
                0x00,
                0x66,
                0x9C,  # Address CRC8
            ]
        )
        return [self.rdwr(READ_OTP, 3) for _ in range(4)]

    def reset(self):
        self.rdwr(SOFT_RESET, delay=0.1)

    def calculate_conversion_constants(self, p_lut):
        p_pa = [45_000, 80_000, 105_000]
        c = (
            p_lut[0] * p_lut[1] * (p_pa[0] - p_pa[1])
            + p_lut[1] * p_lut[2] * (p_pa[1] - p_pa[2])
            + p_lut[2] * p_lut[0] * (p_pa[2] - p_pa[0])
        ) / (
            p_lut[2] * (p_pa[0] - p_pa[1])
            + p_lut[0] * (p_pa[1] - p_pa[2])
            + p_lut[1] * (p_pa[2] - p_pa[0])
        )
        a = (
            p_pa[0] * p_lut[0] - p_pa[1] * p_lut[1] - (p_pa[1] - p_pa[0]) * c
        ) / (p_lut[0] - p_lut[1])
        return a, (p_pa[0] - a) * (p_lut[0] + c), c

    def process_data(self, pressure, temperature):
        lut_lower = 3.5 * (1 << 20)
        lut_upper = 11.5 * (1 << 20)
        quadr_factor = 1 / 16777216
        offst_factor = 2048

        t = temperature - 32768
        a, b, c = self.calculate_conversion_constants(
            [
                lut_lower + (self.sensor_constants[0] * t * t) * quadr_factor,
                (
                    offst_factor * self.sensor_constants[3]
                    + (self.sensor_constants[1] * t * t) * quadr_factor
                ),
                lut_upper + (self.sensor_constants[2] * t * t) * quadr_factor,
            ]
        )

        return (a + b / (c + pressure), -45 + 175 / 65536 * temperature)

    def measure(self, measure_command=NORMAL):
        result = self.rdwr(
            measure_command, 9, MEASUREMENT_DELAYS[measure_command]
        )
        return self.process_data(
            (result[1] << 8) | (result[2] >> 8), result[0]
        )

Display More

Humpelschlumpf · Mar 15th 2023

Hallöchen,

wie ich schon sagte, ich hab von Python wenig Ahnung.

Dies ist hier der Orginaltreiber von TDK invensense.

Ich versuche zu verstehen, an welche Adresse ich das jeweilige Kommando sende und von welcher Adresse ich die Antwort bekomme.

So wie ich das bisher verstanden habe, müsste ich also z.B. für die CHIP_ID den Wert 0xEFC8 an die CHIP_ADRESSE 0x63 Senden und

bekomme dann von der Adresse 0x63 die CHIP_ID 0x08 zurück.

Korrekt?

Genauso müsste das dann auch mit den Kalibrierwerten laufen.

Schreibe das ARRAY move_address_ptr[] an die 0x63 und lese dann 4x die 0x63 ???

Hab ich das richtig verstanden?

Die nächste Schwierigkeit wird dann das schreiben von 16Bit-werten sein.

Normalerweise kann ich nur jeweils 1 Byte übergeben (write8(16#63, 16#EF);)

Bei write8(16#63, 16#EFC8); gibt es einen Compilerfehler.

Hier müsste ich mich noch mal schlau machen.

Grüße

Andre

RTFM · Mar 15th 2023

Du brauchst eigentlich kein Reverse Engeneering betreiben.

Das was Du suchst steht im Datenblatt (Kapitel 5)

Servus !

Humpelschlumpf · Mar 15th 2023

RTFM,

stimmt du hast Recht.

Alle Befehle gehen als 16Bit an die Adresse 0x63.

Gelesen wird auch 16Bit von der selben Adresse.

Ich hab mich von den Komplizierten Code irritieren lassen.

Grüße

Andre

Humpelschlumpf · Mar 17th 2023

So mal ein Zwischenstand meiner Bemühungen.

Ich hab jetzt die Komunikation in Codesys nachgebaut.

Natürlich geht es nicht, wäre ja auch zu einfach.

Schon beim senden des 1. Befehls, geht der ICP in Störung. Überprüfen konnte ich das mit dem Befehl:

i2cset -f 1 0x63 0xEF 0xC8 i

i2cget -y 1 0x63

Als Ergebnis bekomme ich dann ein ERROR.

Sobald ich jedoch kurz das Pythonscript laufen lasse, redet der ICP wieder und ich bekomme dann ein 0x01.

Das Problem liegt also bei Codesys. Normalerweise kann ich nur Byteweise senden, der ICP jedoch will WORD.

Es scheint aber die Möglichkeit zu geben, Daten in WORD zu senden. Dazu mache ich mich gerade schlau.

Mein weiteres Vorgehen ist, den Datenverehr auf dem I2C-Bus zu analysieren.

Dazu habe ich mir ein Logic Analyzer bestellt und werd mal den Datenverkehr untersuchen.

Ich schätze, der Vergleich des Datenverkehrs beim Pythonscript und Codesys wird mir Aufschluss geben wo es wirklich klemmt.

Gruß

Andre

Humpelschlumpf · Mar 21st 2023

So nun bin ich schlauer.

Ich hab den LogicAnalyzer angeschlossen und mal die Kommunikation mit getrackt.

Das Pythonscript hab ich auf read CHIP_ID und read OTP beschränkt.

Chip ID >>> 0x63 0xef 0xc8

Antwort <<< 0x63 0x01 0x48 0xf1 Das Ergibt 0x0148 + 0xf1(CRC8) >>>> Binär 101001000 wobei die 1000 für die Chip_ID steht.

MOVE_ADDRESS_PTR >>> 0x63 0xc5 0x00 0x66 0x9c

read OTP>>> 0x63 0xc7 0xf7

Antwort <<< 0x63 0x01 0x84 0xcb ergibt 388 + 0xcb(CRC8)

read OTP>>> 0x63 0xc7 0xf7

Antwort <<< 0x63 0x02 0x7f 0x8e ergibt 639 + 0x8e(CRC8)

read OTP>>> 0x63 0xc7 0xf7

Antwort <<< 0x63 0x03 0x9e 0x8A ergibt 926 + 0x8a(CRC8)

read OTP>>> 0x63 0xc7 0xf7

Antwort <<< 0x63 0x0f 0x04 0xdc ergibt 3844 + 0xdc(CRC8)

Jetzt ist mir alles klar. Der ICP arbeitet zwar mit 16Bit, aber die Daten werden dennoch Byteweise gesendet und empfangen.

Das heißt, ich muss die Kommandos einfach nur in ein ARRAY packen und senden.

Beim lesen genauso. Daten von 0x63 lesen und in ein ARRAY schreiben.

Nachtrag:

Ich habs jetzt genauso eingegeben uns siehe da, ich bekomme eine Antwort.

Users Viewing This Thread