''' ========================================================= ______ ______ /\ _ \ /\ _ \ __ \ \ \/\_\ ____ ____ ___\ \ \_\ \/\_\ \ \ \/_/_ / __ \ / __ \ / __'\ \ __/\/_/_ \ \ \_\ \/\ \_\ \/\ \__//\ \_\ \ \ \/ /\ \ \ \____/\ \____/\ \____\ \____/\ \_\ \ \_\ \/___/ \/___/ \/____/\/___/ \/_/ \/_/ __ /\ \ __ ---------------- \ \ \ /\_\ ____ __ __ __ _ pi.cocorobo.cn \ \ \ \/_/_ / _ \/\ \/\ \/\ \/ \ ---------------- \ \ \____ /\ \/\ \/\ \ \ \_\ \/> > 3) & 1 def _trigger_measurement(self): self._write_command([0xAC, 0x33, 0x00]) start_time = time.time() time.sleep(85 * 10 ** -3) while True: data = self._read_data(7) if not (data[0] >> 7) & 1: return data if time.time() - start_time > 0.08: raise TimeoutError("AHT20 measurement timeout") time.sleep(0.01) def _crc8(self, data): crc = 0xFF for byte in data: crc ^= byte for _ in range(8): if crc & 0x80: crc = (crc << 1) ^ 0x31 else: crc <<= 1 crc &= 0xFF return crc def read_raw_data(self): data = self._trigger_measurement() crc = self._crc8(data[:6]) if crc != data[6]: raise ValueError("CRC check failed") return data def get_humidity(self): data = self.read_raw_data() raw_humidity = ((data[1] << 16) | (data[2] << 8) | data[3]) >> 4 return (raw_humidity * 100.0) / (2 ** 20) def get_temperature(self): data = self.read_raw_data() raw_temp = ((data[3] & 0x0F) << 16) | (data[4] << 8) | data[5] return (raw_temp * 200.0) / (2 ** 20) - 50.0 ################################## # 2、MPU6050陀螺仪六轴传感器 # ################################## SLAVE_ADDR = 0x68 PWR_MGMT_1 = 0x6B PWR_MGMT_2 = 0x6C WHO_AM_I = 0x75 GYRO_X = 0x43 GYRO_Y = 0x45 GYRO_Z = 0x47 ACCL_X = 0x3B ACCL_Y = 0x3D ACCL_Z = 0x3F class MPU6050: def __init__(self, BusNum=2): self.BusNum = BusNum self.cmd_soft_reset() def cmd_soft_reset(self): _BUS.write_byte_data(SLAVE_ADDR, PWR_MGMT_1, 0) time.sleep(0.04) def read_byte(self, addr): return _BUS.read_byte_data(SLAVE_ADDR, addr) def read_word(self, addr): h = _BUS.read_byte_data(SLAVE_ADDR, addr) l = _BUS.read_byte_data(SLAVE_ADDR, addr + 1) return (h << 8) + l def read_word_i2c(self, addr): val = self.read_word(addr) if val >= 0x8000: return -((65535 - val) + 1) return val def dist(self, x, y): return math.sqrt((x * x) + (y * y)) def get_x_rotat(self, x, y, z): rad = math.atan2(y, self.dist(x, z)) return math.degrees(rad) def get_y_rotat(self, x, y, z): rad = math.atan2(x, self.dist(y, z)) return -math.degrees(rad) def read_gyro(self): GYR_X = self.read_word_i2c(GYRO_X) GYR_Y = self.read_word_i2c(GYRO_Y) GYR_Z = self.read_word_i2c(GYRO_Z) return (GYR_X / 131), (GYR_Y / 131), (GYR_Z / 131) def read_acc(self): ACC_X = self.read_word_i2c(ACCL_X) ACC_Y = self.read_word_i2c(ACCL_Y) ACC_Z = self.read_word_i2c(ACCL_Z) CALC_ACC_X = ACC_X / 16384.0 CALC_ACC_Y = ACC_Y / 16384.0 CALC_ACC_Z = ACC_Z / 16384.0 return (CALC_ACC_X, CALC_ACC_Y, CALC_ACC_Z, self.get_x_rotat(CALC_ACC_X, CALC_ACC_Y, CALC_ACC_Z), self.get_y_rotat(CALC_ACC_X, CALC_ACC_Y, CALC_ACC_Z)) def get_gyro_x(self): return self.read_gyro()[0] def get_gyro_y(self): return self.read_gyro()[1] def get_gyro_z(self): return self.read_gyro()[2] def get_acc_x(self): return self.read_acc()[0] def get_acc_y(self): return self.read_acc()[1] def get_acc_z(self): return self.read_acc()[2] def get_angle_x(self): return self.read_acc()[3] def get_angle_y(self): return self.read_acc()[4] ################################## # 3、板载4颗按钮A、B、C、D # ################################## class BUTTON: def __init__(self, gpioId): self.gpio = 224 + gpioId SUNXI_GPIO.setcfg(self.gpio, SUNXI_GPIO.IN) def is_pressed(self): self.getValue = SUNXI_GPIO.input(self.gpio) return self.getValue != 1 ################################## # 4、光线强度传感器 # ################################## class LIGHTINTENSITY: def __init__(self, addr=b"0x05070080") -> None: self.addr = addr self.path = "/sys/class/sunxi_dump/dump" self.file = open(self.path, "wb+") self.last = self.value() def value(self): self.file.write(b"0x05070080") self.file.seek(0) return int(self.file.read()[:-1], 16) def __del__(self): try: if self.file: self.file.close() del self.file except Exception: pass ################################## # 5、板载LED灯 # ################################## class LED: def __init__(self, gpioId=69): self.gpioId = gpioId SUNXI_GPIO.setcfg(self.gpioId, SUNXI_GPIO.OUT) SUNXI_GPIO.setcfg(259, SUNXI_GPIO.OUT) def out(self, value): if value == 0 or value == 1: self.value = 1 - value else: self.value = 0 SUNXI_GPIO.output(self.gpioId, self.value) SUNXI_GPIO.output(259, self.value) ################################## # 6、QMI8658 # ################################## class QMI8658(object): def __init__(self, smbus=2, address=0X6B): self._address = address self._bus = _BUS bRet = self.WhoAmI() if bRet: self.Read_Revision() else: return None self.Config_apply() def _read_byte(self, cmd): rec = self._bus.read_i2c_block_data(int(self._address), int(cmd), 1) return rec[0] def _read_block(self, reg, length=1): rec = self._bus.read_i2c_block_data(int(self._address), int(reg), length) return rec def _read_u16(self, cmd): LSB = self._bus.read_i2c_block_data(int(self._address), int(cmd), 1) MSB = self._bus.read_i2c_block_data(int(self._address), int(cmd) + 1, 1) return (MSB[0] << 8) + LSB[0] def _write_byte(self, cmd, val): self._bus.write_i2c_block_data(int(self._address), int(cmd), bytes([int(val)])) def WhoAmI(self): return (0x05) == self._read_byte(0x00) def Read_Revision(self): return self._read_byte(0x01) def Config_apply(self): self._write_byte(0x02, 0x40) self._write_byte(0x08, 0x03) self._write_byte(0x03, 0x04) self._write_byte(0x04, 0x64) self._write_byte(0x06, 0x11) def Read_Raw_XYZ(self): xyz = [0, 0, 0, 0, 0, 0] vals = {} raw_timestamp = self._read_block(0x30, 3) raw_acc_xyz = self._read_block(0x35, 6) raw_gyro_xyz = self._read_block(0x3b, 6) raw_xyz = self._read_block(0x35, 12) timestamp = (raw_timestamp[2] << 16) | (raw_timestamp[1] << 8) | (raw_timestamp[0]) for i in range(6): xyz[i] = (raw_xyz[(i * 2) + 1] << 8) | (raw_xyz[i * 2]) if xyz[i] >= 32767: xyz[i] = xyz[i] - 65535 vals["AcX"] = xyz[0] vals["AcY"] = xyz[1] vals["AcZ"] = xyz[2] vals["GyX"] = xyz[3] vals["GyY"] = xyz[4] vals["GyZ"] = xyz[5] return vals def Read_XYZ(self): xyz = [0, 0, 0, 0, 0, 0] raw_xyz = self.Read_Raw_XYZ() acc_lsb_div = (1 << 12) gyro_lsb_div = 64 for i in range(3): xyz[i] = raw_xyz[i] / acc_lsb_div xyz[i + 3] = raw_xyz[i + 3] * 1.0 / gyro_lsb_div return xyz def get_accel(self, samples=10, calibration=None): result = {} for _ in range(samples): v = self.Read_Raw_XYZ() for m in v.keys(): result[m] = result.get(m, 0) + v[m] / samples if calibration: for m in calibration.keys(): if m == "AcZ": pass else: result[m] -= calibration[m] return result def calibrate(self, threshold=50): print('Calibrating QMI8658... ', end='') while True: v1 = self.get_accel(100) v2 = self.get_accel(100) if all(abs(v1[m] - v2[m]) < threshold for m in v1.keys()): print('Done.') return v1 def getPitchYawRollGxGyGz(self, calibration): caliData = self.get_accel(10, calibration) pitch = caliData["AcX"] / 182.04 yaw = caliData["AcY"] / 182.04 roll = caliData["AcZ"] / 182.04 groX = caliData["GyX"] groY = caliData["GyY"] groZ = caliData["GyZ"] if roll < 0: if pitch >= 0: pitch = 180 - pitch elif pitch < 0: pitch = (180 + pitch) * (-1) if yaw >= 0: yaw = 180 - yaw elif pitch < 0: yaw = (180 + yaw) * (-1) return [pitch, yaw, roll, groX, groY, groZ] ################################## # 7、STM8S # ################################## class stm8s(object): address = 0x50 def __init__(self): self.bus = _BUS self.reset() time.sleep(0.05) def clear(self): self.write(1, 1) self.reset() time.sleep(0.05) def write(self, addr, val): return _bus_write_byte(self.address, addr, val) def fastWrite(self, addr, valBlock): try: self.bus.write_i2c_block_data(self.address, addr, valBlock) return True except Exception: return False def read(self, addr): return _bus_read_byte(self.address, addr) def reset(self): self.write(0, 1) time.sleep(0.05) def dump(self): for i in range(0, 32): print(i, self.read(i)) class singleRgb(stm8s): def __init__(self): super().__init__() self.valR = 0 self.valG = 0 self.valB = 0 self.brightness = 255 self.show() def setColor(self, r, g, b): self.valR = int(r * self.brightness / 255) if 0 <= r <= 255 else 0 self.valG = int(g * self.brightness / 255) if 0 <= g <= 255 else 0 self.valB = int(b * self.brightness / 255) if 0 <= b <= 255 else 0 def setBrightness(self, brightness): self.brightness = brightness if 0 <= brightness <= 255 else 0 def show(self): self.write(31, self.valB) self.write(30, self.valG) self.write(29, self.valR) self.write(28, 6) class dcMotor(stm8s): def __init__(self, id): super().__init__() self.id = id def m1a(self, val=0): self.write(4, 5); self.write(7, val) def m1b(self, val=0): self.write(8, 5); self.write(11, val) def m2a(self, val=0): self.write(12, 5); self.write(15, val) def m2b(self, val=0): self.write(16, 5); self.write(19, val) def dcMotorCtrl(self, dir, speed): dir = 1 - dir if 0 < speed < 255: if self.id == 1: if dir == 0: self.m1a(speed); self.m1b(255) else: self.m1a(255); self.m1b(speed) elif self.id == 2: if dir == 0: self.m2a(speed); self.m2b(255) else: self.m2a(255); self.m2b(speed) elif speed == 0: if self.id == 1: self.m1a(255); self.m1b(255) elif self.id == 2: self.m2a(255); self.m2b(255) elif speed == 255: if self.id == 1: if dir == 0: self.m1a(0); self.m1b(255) else: self.m1a(255); self.m1b(0) elif self.id == 2: if dir == 0: self.m2a(0); self.m2b(255) else: self.m2a(255); self.m2b(0) class multiFuncGpio(stm8s): def __init__(self, id=0, mode=0): super().__init__() self.id = id self.rgbId = 0 self.brightness = 255 self.mode = mode self.write(20 + self.id * 4, self.mode) self.rgbSerial = [0, 0, 0] * 16 time.sleep(0.05) def servoCtrl(self, angle): if 0 <= angle <= 180: self.write(23 + self.id * 4, angle) def digitalRead(self): try: self.gpioVal = self.read(23 + self.id * 4) if self.gpioVal is not None: return self.gpioVal except Exception: print("ERROR 233") def digitalWrite(self, val): if 0 <= val <= 1: self.write(23 + self.id * 4, val) def analogRead(self): try: self.adcValH = self.read(22 + self.id * 4) self.adcValL = self.read(23 + self.id * 4) if self.adcValH is not None and self.adcValL is not None: self.adcVal = self.adcValL + (self.adcValH << 8) return self.adcVal except Exception: print("ERROR 2333") def analogWrite(self, val): if 0 <= val <= 255: self.write(23 + self.id * 4, val) def beep(self, frequency): if 20 <= frequency <= 12000: self.write(22 + self.id * 4, frequency >> 8) self.write(23 + self.id * 4, frequency & 0b11111111) def pixelInit_(self): for i in range(16): self.setPixelColor(i, 0, 0, 0) self.pixelShow() self.write(23 + self.id * 4, 1) def setBrightness(self, brightness): self.brightness = brightness if 0 <= brightness <= 255 else 0 def setPixelColor(self, rgbId, r, g, b): self.setRgbSerialColor(rgbId, r, g, b) def pixelShow(self): for i in range(16): self.write(32 + self.id * 48 + i * 3, self.rgbSerial[i * 3]) self.write(33 + self.id * 48 + i * 3, self.rgbSerial[i * 3 + 1]) self.write(34 + self.id * 48 + i * 3, self.rgbSerial[i * 3 + 2]) self.write(23 + self.id * 4, 1) def setRgbSerialColor(self, rgbId, r, g, b): self.rgbId = rgbId if 0 <= rgbId <= 15 else 0 self.valR = int(r * self.brightness / 255) if 0 <= r <= 255 else 0 self.valG = int(g * self.brightness / 255) if 0 <= g <= 255 else 0 self.valB = int(b * self.brightness / 255) if 0 <= b <= 255 else 0 self.rgbSerial[self.rgbId * 3] = self.valR self.rgbSerial[self.rgbId * 3 + 1] = self.valG self.rgbSerial[self.rgbId * 3 + 2] = self.valB def rgbSerialShow(self): self.pixelShow() ################################## # 8、UvcVideo # ################################## if camera is not None: class UvcVideo(camera.MaixVideo): def __init__(self, source="/dev/videoX"): self.source = source super(UvcVideo, self).__init__() usb_path = '/sys/devices/platform/soc/usbc0/otg_role' if os.popen('cat %s' % usb_path).read().strip() != 'usb_host': os.system('echo "usb_host" > %s' % usb_path) time.sleep(2) def config(self, size=None, video=2, horizontal=0, vertical=0): if size is None: size = (320, 240) super(UvcVideo, self).config(size) print('[camera] config input size(%d, %d, %d)' % (self.width(), self.height(), video)) if self.cam: self.cam = None try: import _coco_camera self.cam = _coco_camera.Camera(self.width(), self.height(), video, horizontal, vertical) except Exception as e: print(e) self.cam = None def read(self): if self.cam is None: self.config() if self.cam: ret, frame = self.cam.read() if ret: return frame else: try: self.config() except Exception as e: print(e) return None def __del__(self): if self.cam: self.cam = None else: class UvcVideo(object): def __init__(self, *args, **kwargs): raise RuntimeError("maix.camera not available") ################################## # 9、DS3231 # ################################## class DS3231(object): def __init__(self): self.bus = _BUS self.address = 0x68 def bcd_to_int(self, bcd, n=2): return int(('%x' % bcd)[-n:]) def int_to_bcd(self, x, n=2): return int(str(x)[-n:], 0x10) def read(self, addr): return _bus_read_byte(self.address, addr) def write(self, addr, val): return _bus_write_byte(self.address, addr, val) def setYear(self, year): self.write(0x06, self.int_to_bcd(year % 100)) def setMonth(self, month): if not 1 <= month <= 12: raise ValueError('Month is out of range [1,12].') self.write(0x05, self.int_to_bcd(month)) def setDay(self, day): if not 1 <= day <= 31: raise ValueError('Day is out of range [1,31].') self.write(0x04, self.int_to_bcd(day)) def setDayOfWeek(self, dayOfWeek): if not 1 <= dayOfWeek <= 7: raise ValueError('Day of week is out of range [1,7].') self.write(0x03, self.int_to_bcd(dayOfWeek)) def setHour(self, hour): if not 0 <= hour < 24: raise ValueError('Hours is out of range [0,23].') self.write(0x02, self.int_to_bcd(hour) & 0x3F) def setMinutes(self, minutes): if not 0 <= minutes < 59: raise ValueError('Minutes is out of range [0,59].') self.write(0x01, self.int_to_bcd(minutes)) def setSeconds(self, seconds): if not 0 <= seconds < 60: raise ValueError('Seconds is out of range [0,59].') self.write(0x00, self.int_to_bcd(seconds)) def setDateTime(self, year, month, day, dayOfWeek, hour, minutes, seconds): self.setYear(year) self.setMonth(month) self.setDay(day) self.setDayOfWeek(dayOfWeek) self.setHour(hour) self.setMinutes(minutes) self.setSeconds(seconds) def getYear(self): return self.bcd_to_int(self.read(0x06)) def getMonth(self): temp = self.read(0x05) return self.bcd_to_int((temp << 4 >> 4) & 0x7F) def getDay(self): return self.bcd_to_int(self.read(0x04)) def getDayOfWeek(self): return self.bcd_to_int(self.read(0x03)) def getHour(self): temp = self.read(0x02) if temp == 0x64: temp = 0x40 return self.bcd_to_int((temp << 4 >> 4) & 0x3F) def getMinutes(self): return self.bcd_to_int(self.read(0x01)) def getSeconds(self): return self.bcd_to_int(self.read(0x00)) def getDateTime(self): return [self.getYear(), self.getMonth(), self.getDay(), self.getDayOfWeek(), self.getHour(), self.getMinutes(), self.getSeconds()] def convertToByteType(self, number): return bytes([number]) def decodeToDec(self, byte): return ((byte[0] >> 4) * 10) + (byte[0] & 0x0F) def encodeToByte(self, dec): tens = floor(dec / 10) ones = dec - tens * 10 return (tens << 4) + ones def decodeAlarmType(self, alarmTime): if len(alarmTime) > 4: m1Bit = (alarmTime[3] & 0x80) >> 7 else: m1Bit = False m2Bit = (alarmTime[2] & 0x80) >> 7 m3Bit = (alarmTime[1] & 0x80) >> 7 m4Bit = (alarmTime[0] & 0x80) >> 7 dayBit = (alarmTime[0] & 0x40) >> 6 if m1Bit and m2Bit and m3Bit and m4Bit: return "everySecond" elif not m1Bit and m2Bit and m3Bit and m4Bit: return "everyMinute" elif not m1Bit and not m2Bit and m3Bit and m4Bit: return "everyHour" elif not m1Bit and not m2Bit and not m3Bit and m4Bit: return "everyDay" elif not dayBit and not m1Bit and not m2Bit and not m3Bit and not m4Bit: return "everyMonth" elif dayBit and not m1Bit and not m2Bit and not m3Bit and not m4Bit: return "everyWeek" else: return "noValidAlarmType" def decodeAlarmTime(self, alarmTime): alarmTime[0] = self.decodeToDec(self.convertToByteType(alarmTime[0] & 0x3F)) alarmTime[1] = self.decodeToDec(self.convertToByteType(alarmTime[1] & 0x3F)) alarmTime[2] = self.decodeToDec(self.convertToByteType(alarmTime[2] & 0x7F)) if len(alarmTime) > 4: alarmTime[3] = self.decodeToDec(self.convertToByteType(alarmTime[3] & 0x7F)) return alarmTime def encodeAlarmType(self, alarmType): table = { "everySecond": 15, "everyMinute": 14, "everyHour": 12, "everyDay": 8, "everyMonth": 0, "everyWeek": 16, } if alarmType not in table: raise ValueError("Not a supported alarmType.") return table[alarmType] def encodeDateTime(self, day, hour, minutes, seconds, alarmType): alarmBits = self.encodeAlarmType(alarmType) alarmTime = [0, 0, 0, 0] alarmTime[0] = (self.encodeToByte(day) & 0x3F) | \ ((alarmBits & 0x10) << 2) | ((alarmBits & 0x08) << 4) alarmTime[1] = (self.encodeToByte(hour) & 0x3F) | ((alarmBits & 0x04) << 5) alarmTime[2] = (self.encodeToByte(minutes) & 0x7F) | ((alarmBits & 0x02) << 6) alarmTime[3] = (self.encodeToByte(seconds) & 0x7F) | ((alarmBits & 0x01) << 7) return alarmTime ################################## # 10、SPI_WS2812 # ################################## class SPI_WS2812(object): def __init__(self, num): self.num = num if _maix_spi is not None: self.spi = _maix_spi.SpiDev() self.spi.open(0, 0) self.spi.mode = 3 self.spi.bits_per_word = 8 self.spi.max_speed_hz = 20000000 else: self.spi = None self.data = [[0, 0, 0]] * self.num self.brightness = 255 def setRgbColor(self, id, r, g, b): if id > (self.num - 1): raise ValueError("Out of range!") self.data[id] = [int(g * self.brightness / 255), int(r * self.brightness / 255), int(b * self.brightness / 255)] def setBrightness(self, brightness): if brightness > 255 or brightness < 0: raise ValueError("Out of range!") self.brightness = brightness def rgbShow(self): if self.spi is None: return tx = [] for rgb in self.data: for byte in rgb: for ibit in range(3, -1, -1): tx.append(((byte >> (2 * ibit + 1)) & 1) * 0x60 + ((byte >> (2 * ibit + 0)) & 1) * 0x06 + 0x88) print([hex(v) for v in tx]) tx[0] = 0b1000000 tx.append(0b1000000) self.spi.xfer(tx, int(4 / 1.05e-6)) def __del__(self): try: self.data = [[0, 0, 0]] * self.num self.rgbShow() except Exception: pass ################################## # 11、HCSR04 # ################################## class HCSR04: def __init__(self, Trig, Echo): pinNum = {"TXD": "198", "RXD": "199", "CLK": "224", "MOSI": "225", "MISO": "226", "CS": "227"} os.system("insmod /home/drivers/hcsr04.ko") pinCmd = pinNum.get(Trig) + "," + pinNum.get(Echo) + ",23200,0" os.system('echo "' + pinCmd + '" > /sys/class/hcsr04/value') def getDistance(self): startTime = round(time.time(), 3) while True: dis = os.popen("cat /sys/class/hcsr04/value").read() if "-1" in dis: pass else: dis = round(int(dis) / 58, 2) if dis < 250: return dis if (round(time.time(), 3) - startTime) >= 1.200: return -1 def __del__(self): os.system("rmmod /home/drivers/hcsr04.ko") ################################## # 12、MCP23017 # ################################## class MCP23017(object): address = 0x20 iodira = 0x00 iodirb = 0x01 gppua = 0x0c gppub = 0x0d gpioa = 0x12 gpiob = 0x13 DIR_A_V = 0xff DIR_B_V = 0xff GPPU_A_V = 0x00 GPPU_B_V = 0x00 GPIO_A_V = 0x00 GPIO_B_V = 0x00 def __init__(self): self.bus = _BUS self.reset() def reset(self): self.DIR_A = self.DIR_A_V self.DIR_B = self.DIR_B_V self.GPPU_A = self.GPPU_A_V self.GPPU_B = self.GPPU_B_V self.GPIO_A = self.GPIO_A_V self.GPIO_B = self.GPIO_B_V self.byteWrite(self.iodira, self.DIR_A_V) self.byteWrite(self.iodirb, self.DIR_B_V) self.byteWrite(self.gppua, self.GPPU_A_V) self.byteWrite(self.gppub, self.GPPU_B_V) self.byteWrite(self.gpioa, self.GPIO_A_V) self.byteWrite(self.gpiob, self.GPIO_B_V) print("mcp23017 reset completed!") def byteWrite(self, addr, val): return _bus_write_byte(self.address, addr, val) def byteRead(self, addr): return _bus_read_byte(self.address, addr) def bitWrite(self, reg, num, val): return ((reg >> (num + 1) << (num + 1)) | (val << num) | (reg << (8 - num) & 0xFF) >> (8 - num)) def bitRead(self, reg, num): return ((reg << (7 - num) & 0xFF) >> 7) def setGpioDir(self, channel, pinId, direction): if channel == "A": self.DIR_A = self.bitWrite(self.DIR_A, pinId, direction) self.byteWrite(self.iodira, self.DIR_A) elif channel == "B": self.DIR_B = self.bitWrite(self.DIR_B, pinId, direction) self.byteWrite(self.iodirb, self.DIR_B) else: raise ValueError("Error Channel!") def setGppuEnable(self, channel, pinId, pullup): if channel == "A": self.GPPU_A = self.bitWrite(self.GPPU_A, pinId, pullup) self.byteWrite(self.gppua, self.GPPU_A) elif channel == "B": self.GPPU_B = self.bitWrite(self.GPPU_B, pinId, pullup) self.byteWrite(self.gppub, self.GPPU_B) else: raise ValueError("Error Channel!") def setGpioLevel(self, channel, pinId, level): if channel == "A": self.GPIO_A = self.bitWrite(self.GPIO_A, pinId, level) self.byteWrite(self.gpioa, self.GPIO_A) elif channel == "B": self.GPIO_B = self.bitWrite(self.GPIO_B, pinId, level) self.byteWrite(self.gpiob, self.GPIO_B) else: raise ValueError("Error Channel!") def getGpioLevel(self, channel, pinId): if channel == "A": v = self.byteRead(self.gpioa) elif channel == "B": v = self.byteRead(self.gpiob) else: raise ValueError("Error Channel!") return None if v is None else self.bitRead(v, pinId) def __del__(self): try: self.reset() time.sleep(0.001) except Exception: pass print("del mcp23017") ################################## # 13、ADS1115 # ################################## class ADS1115(object): DEVICE_REG_CONVERSION = 0x00 DEVICE_REG_CONFIG = 0x01 DEVICE_REG_LO_THRESH = 0x02 DEVICE_REG_HI_THRESH = 0x03 CONFIG_OS = 0x8000 CONFIG_OS_START = 0x8000 CONFIG_OS_PERFORMING_CONVERSION = 0x0000 CONFIG_OS_NOT_PERFORMING_OPERATION = 0x8000 CONFIG_MUX_AIN0P_AIN1N = 0x0000 CONFIG_MUX_AIN1P_AIN3N = 0x1000 CONFIG_MUX_AIN2P_AIN3N = 0x2000 CONFIG_MUX_AIN3P_AIN3N = 0x3000 CONFIG_MUX_AIN0P_GNDN = 0x4000 CONFIG_MUX_AIN1P_GNDN = 0x5000 CONFIG_MUX_AIN2P_GNDN = 0x6000 CONFIG_MUX_AIN3P_GNDN = 0x7000 CONFIG_FSR_6V144 = 0x0000 CONFIG_FSR_4V096 = 0x0200 CONFIG_FSR_2V048 = 0x0400 CONFIG_FSR_1V024 = 0x0600 CONFIG_FSR_0V512 = 0x0800 CONFIG_FSR_0V256 = 0x0A00 CONFIG_MODE_CONTINUOUS = 0x0000 CONFIG_MODE_SINGLE_SHOT = 0x0100 CONFIG_DATA_RATE_8SPS = 0x0000 CONFIG_DATA_RATE_16SPS = 0x0020 CONFIG_DATA_RATE_32SPS = 0x0040 CONFIG_DATA_RATE_64SPS = 0x0060 CONFIG_DATA_RATE_128SPS = 0x0080 CONFIG_DATA_RATE_2508SPS = 0x00A0 CONFIG_DATA_RATE_475SPS = 0x00C0 CONFIG_DATA_RATE_860SPS = 0x00E0 CONFIG_COMP_MODE_TRADITIONAL = 0x0000 CONFIG_COMP_MODE_WINDOW = 0x0010 CONFIG_COMP_POL_ACTIVE_LOW = 0x0000 CONFIG_COMP_POL_ACTIVE_HIGH = 0x0008 CONFIG_COMP_LAT = 0x0004 CONFIG_COMP_LAT_NON_LATCHING = 0x0000 CONFIG_COMP_LAT_LATCHING = 0x0004 CONFIG_COMP_QUE = 0x0003 CONFIG_COMP_QUE_1_CONV = 0x0000 CONFIG_COMP_QUE_2_CONV = 0x0001 CONFIG_COMP_QUE_4_CONV = 0x0002 CONFIG_COMP_QUE_DISABLE = 0x0003 addr = 0x49 def __init__(self): self.bus = _BUS self.address = 0x49 self.config = 0x0000 def swap(self, val): return ((val & 0xff00) >> 8) | ((val & 0x00ff) << 8) def readAdc(self, channel): if channel > 3 or channel < 0: return False self.config = (self.CONFIG_OS_START + ((channel + 4) << 12) + self.CONFIG_FSR_6V144 + self.CONFIG_MODE_SINGLE_SHOT + self.CONFIG_DATA_RATE_128SPS + self.CONFIG_COMP_MODE_TRADITIONAL + self.CONFIG_COMP_POL_ACTIVE_LOW + self.CONFIG_COMP_LAT_NON_LATCHING + self.CONFIG_COMP_QUE_DISABLE) # 写配置（带重试，原来这里就是 Errno 70 抛出的地方） ok = False for _ in range(3): try: _BUS.write_word_data(self.address, self.DEVICE_REG_CONFIG, self.swap(self.config)) ok = True break except OSError: time.sleep(0.02) except Exception: time.sleep(0.02) if not ok: return None deadline = time.time() + 0.5 while time.time() < deadline: try: status = self.swap(_BUS.read_word_data( self.address, self.DEVICE_REG_CONFIG)) except OSError: time.sleep(0.005) continue if (status & self.CONFIG_OS) != self.CONFIG_OS_PERFORMING_CONVERSION: break try: result = self.swap(_BUS.read_word_data( self.address, self.DEVICE_REG_CONVERSION)) except OSError: return None volt = round((result * 6.144 / 32768), 3) return [result, volt] ################################## # 14、PCA9685 # ################################## class PCA9685(object): def __init__(self, freq=400, min_us=460, max_us=2400, address=0x40, degrees=180): self.bus = _BUS self.address = address self.period = 1000000 / freq self.min_duty = self._us2duty(min_us) self.max_duty = self._us2duty(max_us) self.freq(freq) self.reset() print("Pca9685 init") def write(self, addr, val): return _bus_write_byte(self.address, addr, val) def read(self, addr): return _bus_read_byte(self.address, addr) def reset(self): self.write(0x00, 0x00) def freq(self, freq=None): if freq is None: v = self.read(0xfe) if v is None: return 400 return int(25000000.0 / 4096 / (v - 0.5)) prescale = int(25000000.0 / 4096 / freq + 0.5) self.write(0x00, 0x10) self.write(0xfe, prescale) self.reset() time.sleep(0.01) self.write(0x00, 0xa1) def pwm(self, index, on=None, off=None): if not 0 <= index <= 15: raise ValueError("Pin ID out of range!") if on is None or off is None: return _bus_read_block(self.address, 0x06 + index * 4, 4) data = [0] * 4 data[0] = int(hex(on & 0xff), 16) data[1] = int(hex((on >> 8) & 0xff), 16) data[2] = int(hex(off & 0xff), 16) data[3] = int(hex((off >> 8) & 0xff), 16) for i in range(0, 4): self.write(0x06 + i + index * 4, data[i]) def duty(self, index, value=None): if value is None: return self.pwm(index) elif not 0 <= value <= 4095: raise ValueError("Out of range!") elif value == 0: self.pwm(index, 0, 4096) elif value == 4095: self.pwm(index, 4096, 0) else: self.pwm(index, 0, value) def _us2duty(self, value): return 4095 * value / self.period # 【关键修复】原来的 __del__ 会 for i in range(16): duty(i,0)， # 退出时会把 P0 舵机所在通道也清零，导致舵机被强行拉回 0°。 # 这里改成 pass，退出时不动任何通道。 def __del__(self): pass class extDcMotor(PCA9685): def __init__(self, motorId): PCA9685.__init__(self) self.motorPin = [11, 12, 13, 10, 9, 8, 5, 6, 7, 4, 3, 2] self.motorId = motorId def speedControl(self, speed): self.speed = abs(speed) if not -255 <= speed <= 255: raise ValueError("Out of range!") if self.motorId == "C": if speed < 0: self.duty(self.motorPin[0], 4095) self.duty(self.motorPin[1], 0) self.duty(self.motorPin[2], int(self.speed * 16)) else: self.duty(self.motorPin[0], 0) self.duty(self.motorPin[1], 4095) self.duty(self.motorPin[2], int(self.speed * 16)) elif self.motorId == "D": if speed < 0: self.duty(self.motorPin[3], 4095) self.duty(self.motorPin[4], 0) self.duty(self.motorPin[5], int(self.speed * 16)) else: self.duty(self.motorPin[3], 0) self.duty(self.motorPin[4], 4095) self.duty(self.motorPin[5], int(self.speed * 16)) elif self.motorId == "E": if speed < 0: self.duty(self.motorPin[6], 4095) self.duty(self.motorPin[7], 0) self.duty(self.motorPin[8], int(self.speed * 16)) else: self.duty(self.motorPin[6], 0) self.duty(self.motorPin[7], 4095) self.duty(self.motorPin[8], int(self.speed * 16)) elif self.motorId == "F": if speed < 0: self.duty(self.motorPin[9], 4095) self.duty(self.motorPin[10], 0) self.duty(self.motorPin[11], int(self.speed * 16)) else: self.duty(self.motorPin[9], 0) self.duty(self.motorPin[10], 4095) self.duty(self.motorPin[11], int(self.speed * 16)) def __del__(self): try: if self.motorId == "C": for i in range(3): self.duty(self.motorPin[i], 0) elif self.motorId == "D": for i in range(3, 6): self.duty(self.motorPin[i], 0) elif self.motorId == "E": for i in range(6, 9): self.duty(self.motorPin[i], 0) elif self.motorId == "F": for i in range(9, 12): self.duty(self.motorPin[i], 0) except Exception: pass time.sleep(0.001) class extServo(PCA9685): def __init__(self, servoId): PCA9685.__init__(self) self.servoId = servoId self.servoPin = [14, 15, 1, 0] self.degrees = 180 def position(self, degrees=None): if degrees is None: return if self.max_duty > 3800: self.max_duty = 3800 span = self.max_duty - self.min_duty duty = self.min_duty + span * degrees / self.degrees duty = int(min(self.max_duty, max(self.min_duty, int(duty)))) self.duty(self.servoPin[self.servoId], duty) def release(self): self.duty(self.servoPin[self.servoId], 0) def __del__(self): try: self.duty(self.servoPin[self.servoId], 0) except Exception: pass class extOutputPin(PCA9685): def __init__(self): PCA9685.__init__(self) self.pin = [14, 15, 1, 0] self.pinId = 0 def digitalSet(self, pinId, val): self.pinId = pinId if val == 0: self.duty(self.pin[self.pinId], 0) else: self.duty(self.pin[self.pinId], 4095) def pwmWrite(self, pinId, val): self.pinId = pinId self.duty(self.pin[self.pinId], val * 16) def __del__(self): try: self.duty(self.pin[self.pinId], 0) except Exception: pass # ============================================================ # 【关键修复】extGpio 不再继承 extOutputPin(PCA9685) # —— 只保留 MCP23017 + ADS1115 # # 原因：extOutputPin 里的通道映射 [14,15,1,0] 与 extServo 完全一致， # 一旦使用 a0-a3 / p0-p3，就会和 extServo 抢同一个 PCA9685 通道， # 导致 P0 舵机被搞死。 # # analogWrite 被移除（原来是通过 PCA9685 输出的 PWM）。 # 若确实需要 PWM 输出，请单独创建一个 extOutputPin 实例， # 并避开 extServo 使用的通道。 # ============================================================ class extGpio(MCP23017, ADS1115): def __init__(self): MCP23017.__init__(self) ADS1115.__init__(self) self.address = 0x20 # 默认选择 MCP23017 # ---------------- pinMode ---------------- def pinMode(self, pinId, mode): self.address = 0x20 # pinId: 0~3 = p0~p3 | 4~7 = a0~a3 # mode : 0 = 输入 | 1 = 输出 if 0 <= pinId <= 3: if mode == 0: if pinId == 0: self.setGpioDir("B", 2, 0); self.setGpioLevel("B", 2, 1) self.setGpioDir("B", 3, 1) elif pinId == 1: self.setGpioDir("B", 1, 0); self.setGpioLevel("B", 1, 1) self.setGpioDir("B", 0, 1) elif pinId == 2: self.setGpioDir("B", 6, 0); self.setGpioLevel("B", 6, 1) self.setGpioDir("A", 1, 1) elif pinId == 3: self.setGpioDir("B", 5, 0); self.setGpioLevel("B", 5, 1) self.setGpioDir("B", 4, 1) elif mode == 1: if pinId == 0: self.setGpioDir("B", 2, 0); self.setGpioLevel("B", 2, 1) self.setGpioDir("B", 3, 0) elif pinId == 1: self.setGpioDir("B", 1, 0); self.setGpioLevel("B", 1, 1) self.setGpioDir("B", 0, 0) elif pinId == 2: self.setGpioDir("B", 6, 0); self.setGpioLevel("B", 6, 1) self.setGpioDir("A", 1, 0) elif pinId == 3: self.setGpioDir("B", 5, 0); self.setGpioLevel("B", 5, 1) self.setGpioDir("B", 4, 0) elif 4 <= pinId <= 7: if mode == 0: if pinId == 4: # a0 self.setGpioDir("A", 4, 0); self.setGpioLevel("A", 4, 1) self.setGpioDir("A", 5, 1) elif pinId == 5: # a1 self.setGpioDir("A", 3, 0); self.setGpioLevel("A", 3, 1) self.setGpioDir("A", 7, 1) elif pinId == 6: # a2 self.setGpioDir("A", 2, 0); self.setGpioLevel("A", 2, 1) self.setGpioDir("A", 6, 1) elif pinId == 7: # a3 self.setGpioDir("B", 7, 0); self.setGpioLevel("B", 7, 1) self.setGpioDir("A", 0, 1) elif mode == 1: if pinId == 4: self.setGpioDir("A", 4, 0); self.setGpioLevel("A", 4, 1) self.setGpioDir("A", 5, 0) elif pinId == 5: self.setGpioDir("A", 3, 0); self.setGpioLevel("A", 3, 1) self.setGpioDir("A", 7, 0) elif pinId == 6: self.setGpioDir("A", 2, 0); self.setGpioLevel("A", 2, 1) self.setGpioDir("A", 6, 0) elif pinId == 7: self.setGpioDir("B", 7, 0); self.setGpioLevel("B", 7, 1) self.setGpioDir("A", 0, 0) # ---------------- digitalWrite ---------------- def digitalWrite(self, pinId, level): self.address = 0x20 if 0 <= pinId <= 3: if pinId == 0: self.setGpioLevel("B", 3, level) elif pinId == 1: self.setGpioLevel("B", 0, level) elif pinId == 2: self.setGpioLevel("A", 1, level) elif pinId == 3: self.setGpioLevel("B", 4, level) elif 4 <= pinId <= 7: if pinId == 4: self.setGpioLevel("A", 5, level) elif pinId == 5: self.setGpioLevel("A", 7, level) elif pinId == 6: self.setGpioLevel("A", 6, level) elif pinId == 7: self.setGpioLevel("A", 0, level) # ---------------- digitalRead ---------------- def digitalRead(self, pinId): self.address = 0x20 if 0 <= pinId <= 3: if pinId == 0: return self.getGpioLevel("B", 3) elif pinId == 1: return self.getGpioLevel("B", 0) elif pinId == 2: return self.getGpioLevel("A", 1) elif pinId == 3: return self.getGpioLevel("B", 4) elif 4 <= pinId <= 7: if pinId == 4: return self.getGpioLevel("A", 5) elif pinId == 5: return self.getGpioLevel("A", 7) elif pinId == 6: return self.getGpioLevel("A", 6) elif pinId == 7: return self.getGpioLevel("A", 0) return None # ---------------- analogRead (a0-a3) ---------------- def analogRead(self, pinId): adcPin = [3, 1, 0, 2] if not 4 <= pinId <= 7: return None self.address = 0x20 if pinId == 4: # a0 self.setGpioDir("A", 4, 1) self.setGpioLevel("A", 4, 0) elif pinId == 5: # a1 self.setGpioDir("A", 3, 1) self.setGpioLevel("A", 3, 0) elif pinId == 6: # a2 self.setGpioDir("A", 2, 1) self.setGpioLevel("A", 2, 0) elif pinId == 7: # a3 self.setGpioDir("B", 7, 0) self.setGpioLevel("B", 7, 0) # 给 MCP23017 通道切换留点时间，否则下一条 I2C 会打到 0x49 时总线刚切完 time.sleep(0.005) self.address = 0x49 try: return self.readAdc(adcPin[pinId - 4]) finally: self.address = 0x20 # ---------------- __del__ ---------------- def __del__(self): try: self.address = 0x20 MCP23017.__del__(self) except Exception: pass ################################## # 15、LazyImport # ################################## class LazyImport(object): def __init__(self, name): self.name = name self.module = None def __getattr__(self, item): if self.module is None: self.module = __import__(self.name) return getattr(self.module, item)