Micropython Kütüphane Yazamama

[Firzen]

Herkese Merhabalar;

Uzun zamandık kontrol edilebilirlik ve gözlenebilirlik üzerine bir sürü yayın okudum. Kendi kütüphanelerimi oluşturdum ve Yükseklisans doktora çalışmalarımda kullandım. Şimdi bu kütüphaneleri bir kaç ürün içine yerleştirmek ve ticari ürün haline getirmek istiyorum. Klasik kalman filtresi ve PID artık çok doyurucu oldu. Cem Uzan'ında dediği gibi halk artık sıkıldı "After Party" istiyor. Robust'lığını garantilediğim Fuzzy Controller ve IMM methodlarımın çalışması için sensör ve haberleşme protokollerinide sabitlemem lazım. Latency olayı ve kesme vs. Bu işlemler için normalde C dilinde yazılması gerektiği konusunda bir sürü tartışma vs okusam da ben python ile devam edeceğim.

Gelelim en basit sorularıma I2C kütüphanesi yazamamak. Raspberry Pi Pico kartı aldım. Ardından da bu kart ile PIN-IO, ADC, PWM, UART denemesi yaptım bunlar tamam. Ama I2C ve SPI ile sorunum hala devam etmekte. I2C hazır library'yi çalıştırıyorum çalışmıyor. Bu alanda çalışma yapmış denemiş olan varsa bir bilgi paylaşımı yapabilir mi acaba? Örneğin I2C kütüphanesi nasıl yazılıyor?
Ben sadece çipin adresini okuyabildim. Sonra hep hata mesajı aldım.

Teşekkürler.

P.S. youtubedaki videoları zaten denedim hiçbiri olmadı ya ben yanlış yapıyorum ya onlar. O yüzden gerçekten denemiş birisinden bilgi almak istiyorum.

[muuzoo]

Sadece basit bir düşünce olarak, I2C için hazır kütüphanenin de çalışmadığını belirtmişsiniz, acaba devrede PULL-UP dirençleri mevcut mu I2C için? Gözden kaçabiliyor bazen.

[MC_Skywalker]

MicroPython haricinde CircutPyton diye bir yorumlayıcı var. Acaba verilen örenekelrde böyle bir fark olabilirmi. Örneklerin çoğı Adafruit kütüpanlerini kullanıyor. Acaba bu her iki yorumlayıcı için bir fark sıkıntı çıkartıyor olabilir mi?

[Firzen]

Yazdığım en basit hali ile I2C Code:

Sistemin Cevabı:

Burada WHO_AM_I yazınca da (yani o byte) aynı cevabı veriyor. Böyle bir durumda Pull-up Pull-down yapılıp yapılmadığını anlamış olmuyor muyum? Mesela Pull-up olmasa öip cevap vermez değil mi?

MicroPython ile Circuitpython maalesef aynı değil AMA unified library ile birbirine benzetim sağlanıyor.

Kullandığım çip Raspberry Pi Pico, Kullandığım ve çalışmayan kütüphanelerde Raspberry Pi Pico-Dev.

[mehmet]

Allta BMP180 ile basınç, sıcaklık ve
denizden yüksekliği okuyan kodlar.
RPi pico w üzerinde devamlı çalışıyor.
Bunun değiştirilmiş hali Thinkspeak
üzerine yarım saatte bilgi yolluyor.
I²C kısmı, bmpsensor modülü içerisinde.
Pek bir şey yapmaya gerek yok.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# https://kitflix.com/raspberrypi-bmp180-python/

import bmpsensor
import time

def bmp180_Oku():
    t, p, a = bmpsensor.readBmp180()
    return t, p, a


print(f"BMP180 okunacak. Çıkmak için CTRL+C")
print(f"")

try:
    while True:
        sicaklik, basinc, yukseklik = bmp180_Oku()
        basinc /= 100
        print(f"Ortam Sıcaklığı    {sicaklik}°C.")  # degC
        print(f"Ortam Basıncı      {basinc}hPa.") # Pressure in Pa 
        print(f"Denizden Yükseklik {yukseklik}m.") # Altitude in meters
        print(f"")

except KeyboardInterrupt:
    print(f"Çıkış yapıldı.\n\r")

[Firzen]

Mehmet Hocam öncelikle teşekkür ederim ama aradığım bilgi bu değil.
Mesela bu kod Raspberry Pi için, Raspberry Pi-Pico için değil. Ayrıca gönderdiğiniz kod zaten Python'da yazılmış ve Micropython ile çalışmıyor. Sorum ise şu 10 numaralı satırda "t, p, a = bmpsensor.readBmp180()" yazılmış. Orada bulunan sistemi kendim yazmak istiyorum. Hazır olanı ise daha sonra modifiye edebilmek için.

[fide]

Bmpsensor adında ayrı bir .py dosyası mevcut pi pico yada esp32 içinde. BMP.read() komutu o dosya içindeki fonksiyonu çağırıyor. @mehmet ile aynı şekilde ssd1306, aht10, bmp180, sht30 çalıştırdım. Bulabilirsem dosyaları upload edeyim akşam.

[mehmet]

Ben µP olduğunu sonradan idrak ettim.

https://github.com/micropython-IMU/micropython-bmp180

https://github.com/micropython/micropython

[Firzen]

Bmpsensor adında ayrı bir .py dosyası mevcut pi pico yada esp32 içinde. BMP.read() komutu o dosya içindeki fonksiyonu çağırıyor. @mehmet ile aynı şekilde ssd1306, aht10, bmp180, sht30 çalıştırdım. Bulabilirsem dosyaları upload edeyim akşam.

Teşekkürler tekrardan faydalı olur gerçekten bende çalışma mantığını biliyorum obje oluşturup kütüphane çalıştırıyor. Hatta datasheet manuelleri okuyorum dediği gibi yapıyorum ama çalışmadı o yüzden sanırım işin püf noktası var zannettiğimden sordum.
Olmadı bir de C dilinde başlamayı dahi düşündüm.

21 Nisan 2022, 00:06:09

Ben µP olduğunu sonradan idrak ettim.

https://github.com/micropython-IMU/micropython-bmp180

https://github.com/micropython/micropython

Teşekkürler hocam çoktan forkladım bunları BMP sensorle ile bir başlayıp denerim yazmayı tekrardan umarım bu sefer olur.

[fide]

Bu ESP32 içinde ssd1306 için driver dosyası. Bütün ekran komutları burada

#MicroPython SSD1306 OLED driver, I2C and SPI interfaces created by Adafruit

import time
import framebuf

# register definitions
SET_CONTRAST        = const(0x81)
SET_ENTIRE_ON      = const(0xa4)
SET_NORM_INV        = const(0xa6)
SET_DISP            = const(0xae)
SET_MEM_ADDR        = const(0x20)
SET_COL_ADDR        = const(0x21)
SET_PAGE_ADDR      = const(0x22)
SET_DISP_START_LINE = const(0x40)
SET_SEG_REMAP      = const(0xa0)
SET_MUX_RATIO      = const(0xa8)
SET_COM_OUT_DIR    = const(0xc0)
SET_DISP_OFFSET    = const(0xd3)
SET_COM_PIN_CFG    = const(0xda)
SET_DISP_CLK_DIV    = const(0xd5)
SET_PRECHARGE      = const(0xd9)
SET_VCOM_DESEL      = const(0xdb)
SET_CHARGE_PUMP    = const(0x8d)


class SSD1306:
    def __init__(self, width, height, external_vcc):
        self.width = width
        self.height = height
        self.external_vcc = external_vcc
        self.pages = self.height // 8
        # Note the subclass must initialize self.framebuf to a framebuffer.
        # This is necessary because the underlying data buffer is different
        # between I2C and SPI implementations (I2C needs an extra byte).
        self.poweron()
        self.init_display()

    def init_display(self):
        for cmd in (
            SET_DISP | 0x00, # off
            # address setting
            SET_MEM_ADDR, 0x00, # horizontal
            # resolution and layout
            SET_DISP_START_LINE | 0x00,
            SET_SEG_REMAP | 0x01, # column addr 127 mapped to SEG0
            SET_MUX_RATIO, self.height - 1,
            SET_COM_OUT_DIR | 0x08, # scan from COM[N] to COM0
            SET_DISP_OFFSET, 0x00,
            SET_COM_PIN_CFG, 0x02 if self.height == 32 else 0x12,
            # timing and driving scheme
            SET_DISP_CLK_DIV, 0x80,
            SET_PRECHARGE, 0x22 if self.external_vcc else 0xf1,
            SET_VCOM_DESEL, 0x30, # 0.83*Vcc
            # display
            SET_CONTRAST, 0xff, # maximum
            SET_ENTIRE_ON, # output follows RAM contents
            SET_NORM_INV, # not inverted
            # charge pump
            SET_CHARGE_PUMP, 0x10 if self.external_vcc else 0x14,
            SET_DISP | 0x01): # on
            self.write_cmd(cmd)
        self.fill(0)
        self.show()

    def poweroff(self):
        self.write_cmd(SET_DISP | 0x00)

    def contrast(self, contrast):
        self.write_cmd(SET_CONTRAST)
        self.write_cmd(contrast)

    def invert(self, invert):
        self.write_cmd(SET_NORM_INV | (invert & 1))

    def show(self):
        x0 = 0
        x1 = self.width - 1
        if self.width == 64:
            # displays with width of 64 pixels are shifted by 32
            x0 += 32
            x1 += 32
        self.write_cmd(SET_COL_ADDR)
        self.write_cmd(x0)
        self.write_cmd(x1)
        self.write_cmd(SET_PAGE_ADDR)
        self.write_cmd(0)
        self.write_cmd(self.pages - 1)
        self.write_framebuf()

    def fill(self, col):
        self.framebuf.fill(col)

    def pixel(self, x, y, col):
        self.framebuf.pixel(x, y, col)

    def scroll(self, dx, dy):
        self.framebuf.scroll(dx, dy)

    def text(self, string, x, y, col=1):
        self.framebuf.text(string, x, y, col)


class SSD1306_I2C(SSD1306):
    def __init__(self, width, height, i2c, addr=0x3c, external_vcc=False):
        self.i2c = i2c
        self.addr = addr
        self.temp = bytearray(2)
        # Add an extra byte to the data buffer to hold an I2C data/command byte
        # to use hardware-compatible I2C transactions.  A memoryview of the
        # buffer is used to mask this byte from the framebuffer operations
        # (without a major memory hit as memoryview doesn't copy to a separate
        # buffer).
        self.buffer = bytearray(((height // 8) * width) + 1)
        self.buffer[0] = 0x40  # Set first byte of data buffer to Co=0, D/C=1
        self.framebuf = framebuf.FrameBuffer1(memoryview(self.buffer)[1:], width, height)
        super().__init__(width, height, external_vcc)

    def write_cmd(self, cmd):
        self.temp[0] = 0x80 # Co=1, D/C#=0
        self.temp[1] = cmd
        self.i2c.writeto(self.addr, self.temp)

    def write_framebuf(self):
        # Blast out the frame buffer using a single I2C transaction to support
        # hardware I2C interfaces.
        self.i2c.writeto(self.addr, self.buffer)

    def poweron(self):
        pass


class SSD1306_SPI(SSD1306):
    def __init__(self, width, height, spi, dc, res, cs, external_vcc=False):
        self.rate = 10 * 1024 * 1024
        dc.init(dc.OUT, value=0)
        res.init(res.OUT, value=0)
        cs.init(cs.OUT, value=1)
        self.spi = spi
        self.dc = dc
        self.res = res
        self.cs = cs
        self.buffer = bytearray((height // 8) * width)
        self.framebuf = framebuf.FrameBuffer1(self.buffer, width, height)
        super().__init__(width, height, external_vcc)

    def write_cmd(self, cmd):
        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
        self.cs.high()
        self.dc.low()
        self.cs.low()
        self.spi.write(bytearray([cmd]))
        self.cs.high()

    def write_framebuf(self):
        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
        self.cs.high()
        self.dc.high()
        self.cs.low()
        self.spi.write(self.buffer)
        self.cs.high()

    def poweron(self):
        self.res.high()
        time.sleep_ms(1)
        self.res.low()
        time.sleep_ms(10)
        self.res.high()

BH1750 yada GY30 olarak bilinen Işık sensörü uPython driver

import time

OP_SINGLE_HRES1 = 0x20
OP_SINGLE_HRES2 = 0x21
OP_SINGLE_LRES = 0x23

DELAY_HMODE = 180  # 180ms in H-mode
DELAY_LMODE = 24  # 24ms in L-mode


def sample(i2c, mode=OP_SINGLE_HRES1, i2c_addr=0x23):
    """
        Performs a single sampling. returns the result in lux
    """

    i2c.writeto(i2c_addr, b"\x00")  # make sure device is in a clean state
    i2c.writeto(i2c_addr, b"\x01")  # power up
    i2c.writeto(i2c_addr, bytes([mode]))  # set measurement mode

    time.sleep_ms(DELAY_LMODE if mode == OP_SINGLE_LRES else DELAY_HMODE)

    raw = i2c.readfrom(i2c_addr, 2)
    i2c.writeto(i2c_addr, b"\x00")  # power down again

    # we must divide the end result by 1.2 to get the lux
    return ((raw[0] << 24) | (raw[1] << 16)) // 78642

Main.py içindeki tanımlama

# ESP32 Pin assignment 
i2c = SoftI2C(scl=Pin(22), sda=Pin(21))
# oled ekran
oled_width = 128
oled_height = 64

oled = ssd1306.SSD1306_I2C(oled_width, oled_height, i2c)
# aht10 sensor tanımı
sensor = ahtx0.AHT10(i2c)

SHT30 Driver

from machine import SoftI2C, Pin
import time

__version__ = '0.2.1'
__author__ = 'Roberto Sánchez'
__license__ = "Apache License 2.0. https://www.apache.org/licenses/LICENSE-2.0"

# I2C address B 0x45 ADDR (pin 2) connected to VDD
DEFAULT_I2C_ADDRESS = 0x44

class SHT30():
    """
    SHT30 sensor driver in pure python based on I2C bus
    
    References: 
    * https://www.sensirion.com/fileadmin/user_upload/customers/sensirion/Dokumente/2_Humidity_Sensors/Sensirion_Humidity_Sensors_SHT3x_Datasheet_digital.pdf
    * https://www.wemos.cc/sites/default/files/2016-11/SHT30-DIS_datasheet.pdf
    * https://github.com/wemos/WEMOS_SHT3x_Arduino_Library
    * https://www.sensirion.com/fileadmin/user_upload/customers/sensirion/Dokumente/11_Sample_Codes_Software/Humidity_Sensors/Sensirion_Humidity_Sensors_SHT3x_Sample_Code_V2.pdf
    """
    POLYNOMIAL = 0x131  # P(x) = x^8 + x^5 + x^4 + 1 = 100110001

    ALERT_PENDING_MASK = 0x8000 # 15
    HEATER_MASK = 0x2000        # 13
    RH_ALERT_MASK = 0x0800		# 11
    T_ALERT_MASK = 0x0400		# 10
    RESET_MASK = 0x0010	        # 4
    CMD_STATUS_MASK = 0x0002	# 1
    WRITE_STATUS_MASK = 0x0001	# 0

    # MSB = 0x2C LSB = 0x06 Repeatability = High, Clock stretching = enabled
    MEASURE_CMD = b'\x2C\x10'
    STATUS_CMD = b'\xF3\x2D'
    RESET_CMD = b'\x30\xA2'
    CLEAR_STATUS_CMD = b'\x30\x41'
    ENABLE_HEATER_CMD = b'\x30\x6D'
    DISABLE_HEATER_CMD = b'\x30\x66'

    def __init__(self, scl_pin=22, sda_pin=21, delta_temp = 0, delta_hum = 0, i2c_address=DEFAULT_I2C_ADDRESS):
        self.i2c = SoftI2C(scl=Pin(scl_pin), sda=Pin(sda_pin))
        self.i2c_addr = i2c_address
        self.set_delta(delta_temp, delta_hum)
        time.sleep_ms(50)
    
    def init(self, scl_pin=22, sda_pin=21):
        """
        Init the I2C bus using the new pin values
        """
        self.i2c.init(scl=Pin(scl_pin), sda=Pin(sda_pin))
    
    def is_present(self):
        """
        Return true if the sensor is correctly conneced, False otherwise
        """
        return self.i2c_addr in self.i2c.scan()
    
    def set_delta(self, delta_temp = 0, delta_hum = 0):
        """
        Apply a delta value on the future measurements of temperature and/or humidity
        The units are Celsius for temperature and percent for humidity (can be negative values)
        """
        self.delta_temp = delta_temp
        self.delta_hum = delta_hum
    
    def _check_crc(self, data):
        # calculates 8-Bit checksum with given polynomial
        crc = 0xFF
        
        for b in data[:-1]:
            crc ^= b;
            for _ in range(8, 0, -1):
                if crc & 0x80:
                    crc = (crc << 1) ^ SHT30.POLYNOMIAL;
                else:
                    crc <<= 1
        crc_to_check = data[-1]
        return crc_to_check == crc
    
    def send_cmd(self, cmd_request, response_size=6, read_delay_ms=100):
        """
        Send a command to the sensor and read (optionally) the response
        The responsed data is validated by CRC
        """
        try:
            self.i2c.start(); 
            self.i2c.writeto(self.i2c_addr, cmd_request); 
            if not response_size:
                self.i2c.stop(); 	
                return
            time.sleep_ms(read_delay_ms)
            data = self.i2c.readfrom(self.i2c_addr, response_size) 
            self.i2c.stop(); 
            for i in range(response_size//3):
                if not self._check_crc(data[i*3:(i+1)*3]): # pos 2 and 5 are CRC
                    raise SHT30Error(SHT30Error.CRC_ERROR)
            if data == bytearray(response_size):
                raise SHT30Error(SHT30Error.DATA_ERROR)
            return data
        except OSError as ex:
            if 'I2C' in ex.args[0]:
                raise SHT30Error(SHT30Error.BUS_ERROR)
            raise ex

    def clear_status(self):
        """
        Clear the status register
        """
        return self.send_cmd(SHT30.CLEAR_STATUS_CMD, None); 

    def reset(self):
        """
        Send a soft-reset to the sensor
        """
        return self.send_cmd(SHT30.RESET_CMD, None); 

    def status(self, raw=False):
        """
        Get the sensor status register. 
        It returns a int value or the bytearray(3) if raw==True
        """
        data = self.send_cmd(SHT30.STATUS_CMD, 3, read_delay_ms=20); 

        if raw:
            return data

        status_register = data[0] << 8 | data[1]
        return status_register
    
    def measure(self, raw=False):
        """
        If raw==True returns a bytearrya(6) with sensor direct measurement otherwise
        It gets the temperature (T) and humidity (RH) measurement and return them.
        
        The units are Celsius and percent
        """
        data = self.send_cmd(SHT30.MEASURE_CMD, 6); 

        if raw:
            return data

        t_celsius = (((data[0] << 8 |  data[1]) * 175) / 0xFFFF) - 45 + self.delta_temp;
        rh = (((data[3] << 8 | data[4]) * 100.0) / 0xFFFF) + self.delta_hum;
        return t_celsius, rh

    def measure_int(self, raw=False):
        """
        Get the temperature (T) and humidity (RH) measurement using integers.
        If raw==True returns a bytearrya(6) with sensor direct measurement otherwise
        It returns a tuple with 4 values: T integer, T decimal, H integer, H decimal
        For instance to return T=24.0512 and RH= 34.662 This method will return
        (24, 5, 34, 66) Only 2 decimal digits are returned, .05 becomes 5
        Delta values are not applied in this method
        The units are Celsius and percent.
        """
        data = self.send_cmd(SHT30.MEASURE_CMD, 6); 
        if raw: 
            return data
        aux = (data[0] << 8 | data[1]) * 175
        t_int = (aux // 0xffff) - 45;
        t_dec = (aux % 0xffff * 100) // 0xffff
        aux = (data[3] << 8 | data[4]) * 100
        h_int = aux // 0xffff
        h_dec = (aux % 0xffff * 100) // 0xffff
        return t_int, t_dec, h_int, h_dec


class SHT30Error(Exception):
    """
    Custom exception for errors on sensor management
    """
    BUS_ERROR = 0x01 
    DATA_ERROR = 0x02
    CRC_ERROR = 0x03

    def __init__(self, error_code=None):
        self.error_code = error_code
        super().__init__(self.get_message())
    
    def get_message(self):
        if self.error_code == SHT30Error.BUS_ERROR:
            return "Bus error"
        elif self.error_code == SHT30Error.DATA_ERROR:
            return "Data error"
        elif self.error_code == SHT30Error.CRC_ERROR:
            return "CRC error"
        else:
            return "Unknown error"