Merge pull request #1785 from openmv/pylint

misc: Add Python linter workflow.
This commit is contained in:
Ibrahim Abdelkader 2023-02-19 22:17:34 +02:00 committed by GitHub
commit a9e5c41a71
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
26 changed files with 1416 additions and 943 deletions

51
.github/workflows/python-linter.yml vendored Normal file
View File

@ -0,0 +1,51 @@
# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
name: '🔎 Python Linter'
on:
push:
branches:
- 'master'
paths:
- 'scripts/libraries/*.py'
pull_request:
types:
- opened
- reopened
- synchronize
branches:
- 'master'
paths:
- 'scripts/libraries/*.py'
jobs:
build:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.10"]
steps:
- name: '⏳ Checkout repository'
uses: actions/checkout@v3
- name: '🐍 Set up Python ${{ matrix.python-version }}'
uses: actions/setup-python@v3
with:
python-version: ${{ matrix.python-version }}
- name: '🛠 Install dependencies'
run: |
python -m pip install --upgrade pip
python -m pip install flake8 pytest
if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
- name: '😾 Lint with flake8'
run: |
# stop the build if there are Python syntax errors or undefined names
flake8 --count --select=E9,F63,F7,F82 --ignore=F821 --show-source --statistics scripts/libraries/
flake8 --count --max-complexity=15 --max-line-length=120 --ignore=F821,E722,E741,C901,E713,W605,E203,W503,F841,F403,F405 --statistics scripts/libraries/

View File

@ -0,0 +1,63 @@
# OpenMV M7 I2C interface with Garmin Lidar Lite V3 - By: Grant Phillips - Sun Apr 8 2018
# Returns a basic distance reading from the lidar in cm for the target point and prints to console
# Uses default lidar settings. For more advanced settings, see the I2C commands in the manual:
# https://static.garmin.com/pumac/LIDAR_Lite_v3_Operation_Manual_and_Technical_Specifications.pdf
# I2C Control of LIDAR Lite V3
# 1. Write 0x04 to register 0x00
# 2. Read register 0x01. Repeat until bit 0 (LSB) goes low.
# 3. Read two bytes from 0x8f (high byte 0x0f then low byte 0x10) to obtain 16 bit measurement in cm
# HARDWARE CONNECTIONS:
# Connect the lidar SCL line (green) to I2C 2 SCL on openMV (Pin 4)
# Connect the lidar SDA line (blue) to I2C 2 SDA on openMV (pin 5)
# 680uF filter capacitor in parallel with the lidar
# 10k pullup resistors on the SCL and SDA lines to +5Vdc
import pyb
from pyb import I2C
lidarReady = bytearray([0xFF]) # holds the returned data for ready check
lidarReadyCheck = bytes([1]) # to compare bit 0 of lidarReady
startBuf = bytearray([0x00, 0x04]) # step 1 address and data
readyBuf = bytearray([0x01]) # step 2 address for readiness check
distBuf = bytearray([0x8F]) # step 3 address for distance reading
distance = -1 # variable for distance reading
# I2C setup
Lidar = I2C(2, I2C.MASTER) # initialise I2C 2 bus in master mode
while True:
distance = -1 # reset to -1 so we know when we get a real reading
try: # handles errors thrown up if we have an I2C error
# Step 1 Write 0x04 to register 0x00
Lidar.send(startBuf, 0x62) # this is making it read (laser visible)
# Step 2 Read register 0x01 and wait for bit 0 to go low
while lidarReady[0] & readyBuf[0]:
Lidar.send(readyBuf, 0x62)
lidarReady = Lidar.recv(1, 0x62)
pyb.delay(50) # This seems to help reduce errors on the I2C bus
lidarReady = bytearray([0xFF]) # reset the ready check data for next reading
# Step 3 Read the distance measurement from 0x8f (0x0f and 0x10)
Lidar.send(distBuf, 0x62)
dist = Lidar.recv(2, 0x62)
distance = dist[0]
distance <<= 8 # move 2 bytes into a 16 bit int
distance |= dist[1]
pyb.delay(100) # allow time between readings, can go faster but more errors
except OSError: # reninitialise i2c bus if error
Lidar.init(I2C.MASTER)
print("error, reinitialising")
if distance > -1:
print("Distance:", distance, "cm")

View File

@ -1,3 +1,6 @@
from apds9960.device import APDS9960, uAPDS9960
__all__ = [ 'APDS9960', 'uAPDS9960', ]
__all__ = [
"APDS9960",
"uAPDS9960",
]

View File

@ -7,7 +7,7 @@ APDS9960_GESTURE_SENSITIVITY_1 = 50
APDS9960_GESTURE_SENSITIVITY_2 = 20
# APDS9960 device IDs
APDS9960_DEV_ID = [0xab, 0x9c, 0xa8, -0x55]
APDS9960_DEV_ID = [0xAB, 0x9C, 0xA8, -0x55]
# APDS9960 times
APDS9960_TIME_FIFO_PAUSE = 30
@ -21,11 +21,11 @@ APDS9960_REG_AILTH = 0x85
APDS9960_REG_AIHTL = 0x86
APDS9960_REG_AIHTH = 0x87
APDS9960_REG_PILT = 0x89
APDS9960_REG_PIHT = 0x8b
APDS9960_REG_PERS = 0x8c
APDS9960_REG_CONFIG1 = 0x8d
APDS9960_REG_PPULSE = 0x8e
APDS9960_REG_CONTROL = 0x8f
APDS9960_REG_PIHT = 0x8B
APDS9960_REG_PERS = 0x8C
APDS9960_REG_CONFIG1 = 0x8D
APDS9960_REG_PPULSE = 0x8E
APDS9960_REG_CONTROL = 0x8F
APDS9960_REG_CONFIG2 = 0x90
APDS9960_REG_ID = 0x92
APDS9960_REG_STATUS = 0x93
@ -35,33 +35,33 @@ APDS9960_REG_RDATAL = 0x96
APDS9960_REG_RDATAH = 0x97
APDS9960_REG_GDATAL = 0x98
APDS9960_REG_GDATAH = 0x99
APDS9960_REG_BDATAL = 0x9a
APDS9960_REG_BDATAH = 0x9b
APDS9960_REG_PDATA = 0x9c
APDS9960_REG_POFFSET_UR = 0x9d
APDS9960_REG_POFFSET_DL = 0x9e
APDS9960_REG_CONFIG3 = 0x9f
APDS9960_REG_GPENTH = 0xa0
APDS9960_REG_GEXTH = 0xa1
APDS9960_REG_GCONF1 = 0xa2
APDS9960_REG_GCONF2 = 0xa3
APDS9960_REG_GOFFSET_U = 0xa4
APDS9960_REG_GOFFSET_D = 0xa5
APDS9960_REG_GOFFSET_L = 0xa7
APDS9960_REG_GOFFSET_R = 0xa9
APDS9960_REG_GPULSE = 0xa6
APDS9960_REG_GCONF3 = 0xaA
APDS9960_REG_GCONF4 = 0xaB
APDS9960_REG_GFLVL = 0xae
APDS9960_REG_GSTATUS = 0xaf
APDS9960_REG_IFORCE = 0xe4
APDS9960_REG_PICLEAR = 0xe5
APDS9960_REG_CICLEAR = 0xe6
APDS9960_REG_AICLEAR = 0xe7
APDS9960_REG_GFIFO_U = 0xfc
APDS9960_REG_GFIFO_D = 0xfd
APDS9960_REG_GFIFO_L = 0xfe
APDS9960_REG_GFIFO_R = 0xff
APDS9960_REG_BDATAL = 0x9A
APDS9960_REG_BDATAH = 0x9B
APDS9960_REG_PDATA = 0x9C
APDS9960_REG_POFFSET_UR = 0x9D
APDS9960_REG_POFFSET_DL = 0x9E
APDS9960_REG_CONFIG3 = 0x9F
APDS9960_REG_GPENTH = 0xA0
APDS9960_REG_GEXTH = 0xA1
APDS9960_REG_GCONF1 = 0xA2
APDS9960_REG_GCONF2 = 0xA3
APDS9960_REG_GOFFSET_U = 0xA4
APDS9960_REG_GOFFSET_D = 0xA5
APDS9960_REG_GOFFSET_L = 0xA7
APDS9960_REG_GOFFSET_R = 0xA9
APDS9960_REG_GPULSE = 0xA6
APDS9960_REG_GCONF3 = 0xAA
APDS9960_REG_GCONF4 = 0xAB
APDS9960_REG_GFLVL = 0xAE
APDS9960_REG_GSTATUS = 0xAF
APDS9960_REG_IFORCE = 0xE4
APDS9960_REG_PICLEAR = 0xE5
APDS9960_REG_CICLEAR = 0xE6
APDS9960_REG_AICLEAR = 0xE7
APDS9960_REG_GFIFO_U = 0xFC
APDS9960_REG_GFIFO_D = 0xFD
APDS9960_REG_GFIFO_L = 0xFE
APDS9960_REG_GFIFO_R = 0xFF
# APDS9960 bit fields
APDS9960_BIT_PON = 0b00000001
@ -136,7 +136,7 @@ APDS9960_DEFAULT_PGAIN = APDS9960_PGAIN_4X
APDS9960_DEFAULT_AGAIN = APDS9960_AGAIN_4X
APDS9960_DEFAULT_PILT = 0 # Low proximity threshold
APDS9960_DEFAULT_PIHT = 50 # High proximity threshold
APDS9960_DEFAULT_AILT = 0xffff # Force interrupt for calibration
APDS9960_DEFAULT_AILT = 0xFFFF # Force interrupt for calibration
APDS9960_DEFAULT_AIHT = 0
APDS9960_DEFAULT_PERS = 0x11 # 2 consecutive prox or ALS for int.
APDS9960_DEFAULT_CONFIG2 = 0x01 # No saturation interrupts or LED boost
@ -148,7 +148,7 @@ APDS9960_DEFAULT_GGAIN = APDS9960_GGAIN_4X
APDS9960_DEFAULT_GLDRIVE = APDS9960_LED_DRIVE_100MA
APDS9960_DEFAULT_GWTIME = APDS9960_GWTIME_2_8MS
APDS9960_DEFAULT_GOFFSET = 0 # No offset scaling for gesture mode
APDS9960_DEFAULT_GPULSE = 0xc9 # 32us, 10 pulses
APDS9960_DEFAULT_GPULSE = 0xC9 # 32us, 10 pulses
APDS9960_DEFAULT_GCONF3 = 0 # All photodiodes active during gesture
APDS9960_DEFAULT_GIEN = 0 # Disable gesture interrupts

View File

@ -3,6 +3,7 @@ from apds9960.exceptions import *
from time import sleep_ms
class APDS9960:
class GestureData:
def __init__(self):
@ -78,7 +79,6 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_GCONF3, APDS9960_DEFAULT_GCONF3)
self.setGestureIntEnable(APDS9960_DEFAULT_GIEN)
def getMode(self):
return self._read_byte_data(APDS9960_REG_ENABLE)
@ -92,19 +92,18 @@ class APDS9960:
# change bit(s) in ENABLE register */
if mode == APDS9960_MODE_ALL:
if enable:
reg_val = 0x7f
reg_val = 0x7F
else:
reg_val = 0x00
else:
if enable:
reg_val |= (1 << mode);
reg_val |= 1 << mode
else:
reg_val &= ~(1 << mode);
reg_val &= ~(1 << mode)
# write value to ENABLE register
self._write_byte_data(APDS9960_REG_ENABLE, reg_val)
# start the light (R/G/B/Ambient) sensor
def enableLightSensor(self, interrupts=True):
self.setAmbientLightGain(APDS9960_DEFAULT_AGAIN)
@ -117,7 +116,6 @@ class APDS9960:
self.setAmbientLightIntEnable(False)
self.setMode(APDS9960_MODE_AMBIENT_LIGHT, False)
# start the proximity sensor
def enableProximitySensor(self, interrupts=True):
self.setProximityGain(APDS9960_DEFAULT_PGAIN)
@ -131,11 +129,10 @@ class APDS9960:
self.setProximityIntEnable(False)
self.setMode(APDS9960_MODE_PROXIMITY, False)
# start the gesture recognition engine
def enableGestureSensor(self, interrupts=True):
self.resetGestureParameters()
self._write_byte_data(APDS9960_REG_WTIME, 0xff)
self._write_byte_data(APDS9960_REG_WTIME, 0xFF)
self._write_byte_data(APDS9960_REG_PPULSE, APDS9960_DEFAULT_GESTURE_PPULSE)
self.setLEDBoost(APDS9960_LED_BOOST_300)
self.setGestureIntEnable(interrupts)
@ -152,16 +149,14 @@ class APDS9960:
self.setGestureMode(False)
self.setMode(APDS9960_MODE_GESTURE, False)
# check if there is a gesture available
def isGestureAvailable(self):
val = self._read_byte_data(APDS9960_REG_GSTATUS)
# shift and mask out GVALID bit
val &= APDS9960_BIT_GVALID;
return (val == APDS9960_BIT_GVALID)
val &= APDS9960_BIT_GVALID
return val == APDS9960_BIT_GVALID
# processes a gesture event and returns best guessed gesture
def readGesture(self):
@ -174,7 +169,7 @@ class APDS9960:
return APDS9960_DIR_NONE
# keep looping as long as gesture data is valid
while(self.isGestureAvailable()):
while self.isGestureAvailable():
# read the current FIFO level
fifo_level = self._read_byte_data(APDS9960_REG_GFLVL)
@ -215,18 +210,14 @@ class APDS9960:
self.resetGestureParameters()
return motion
# turn the APDS-9960 on
def enablePower(self):
self.setMode(APDS9960_MODE_POWER, True)
def disablePower(self):
# turn the APDS-9960 off
self.setMode(APDS9960_MODE_POWER, False)
# *******************************************************************************
# ambient light and color sensor controls
# *******************************************************************************
@ -271,7 +262,6 @@ class APDS9960:
return l + (h << 8)
# *******************************************************************************
# Proximity sensor controls
# *******************************************************************************
@ -280,7 +270,6 @@ class APDS9960:
def readProximity(self):
return self._read_byte_data(APDS9960_REG_PDATA)
# *******************************************************************************
# High-level gesture controls
# *******************************************************************************
@ -302,7 +291,6 @@ class APDS9960:
self.gesture_state_ = 0
self.gesture_motion_ = APDS9960_DIR_NONE
def processGestureData(self):
"""Processes the raw gesture data to determine swipe direction
@ -326,11 +314,12 @@ class APDS9960:
if self.gesture_data_.total_gestures <= 32 and self.gesture_data_.total_gestures > 0:
# find the first value in U/D/L/R above the threshold
for i in range(0, self.gesture_data_.total_gestures):
if self.gesture_data_.u_data[i] > APDS9960_GESTURE_THRESHOLD_OUT and \
self.gesture_data_.d_data[i] > APDS9960_GESTURE_THRESHOLD_OUT and \
self.gesture_data_.l_data[i] > APDS9960_GESTURE_THRESHOLD_OUT and \
self.gesture_data_.r_data[i] > APDS9960_GESTURE_THRESHOLD_OUT:
if (
self.gesture_data_.u_data[i] > APDS9960_GESTURE_THRESHOLD_OUT
and self.gesture_data_.d_data[i] > APDS9960_GESTURE_THRESHOLD_OUT
and self.gesture_data_.l_data[i] > APDS9960_GESTURE_THRESHOLD_OUT
and self.gesture_data_.r_data[i] > APDS9960_GESTURE_THRESHOLD_OUT
):
u_first = self.gesture_data_.u_data[i]
d_first = self.gesture_data_.d_data[i]
l_first = self.gesture_data_.l_data[i]
@ -343,11 +332,12 @@ class APDS9960:
# find the last value in U/D/L/R above the threshold
for i in reversed(range(0, self.gesture_data_.total_gestures)):
if self.gesture_data_.u_data[i] > APDS9960_GESTURE_THRESHOLD_OUT and \
self.gesture_data_.d_data[i] > APDS9960_GESTURE_THRESHOLD_OUT and \
self.gesture_data_.l_data[i] > APDS9960_GESTURE_THRESHOLD_OUT and \
self.gesture_data_.r_data[i] > APDS9960_GESTURE_THRESHOLD_OUT:
if (
self.gesture_data_.u_data[i] > APDS9960_GESTURE_THRESHOLD_OUT
and self.gesture_data_.d_data[i] > APDS9960_GESTURE_THRESHOLD_OUT
and self.gesture_data_.l_data[i] > APDS9960_GESTURE_THRESHOLD_OUT
and self.gesture_data_.r_data[i] > APDS9960_GESTURE_THRESHOLD_OUT
):
u_last = self.gesture_data_.u_data[i]
d_last = self.gesture_data_.d_data[i]
l_last = self.gesture_data_.l_data[i]
@ -386,9 +376,10 @@ class APDS9960:
# determine Near/Far gesture
if self.gesture_ud_count_ == 0 and self.gesture_lr_count_ == 0:
if abs(ud_delta) < APDS9960_GESTURE_SENSITIVITY_2 and \
abs(lr_delta) < APDS9960_GESTURE_SENSITIVITY_2:
if (
abs(ud_delta) < APDS9960_GESTURE_SENSITIVITY_2
and abs(lr_delta) < APDS9960_GESTURE_SENSITIVITY_2
):
if ud_delta == 0 and lr_delta == 0:
self.gesture_near_count_ += 1
elif ud_delta != 0 or lr_delta != 0:
@ -401,9 +392,10 @@ class APDS9960:
self.gesture_state_ = APDS9960_STATE_FAR
return True
else:
if abs(ud_delta) < APDS9960_GESTURE_SENSITIVITY_2 and \
abs(lr_delta) < APDS9960_GESTURE_SENSITIVITY_2:
if (
abs(ud_delta) < APDS9960_GESTURE_SENSITIVITY_2
and abs(lr_delta) < APDS9960_GESTURE_SENSITIVITY_2
):
if ud_delta == 0 and lr_delta == 0:
self.gesture_near_count_ += 1
@ -416,8 +408,7 @@ class APDS9960:
return False
def decodeGesture(self):
"""Determines swipe direction or near/far state.
"""
"""Determines swipe direction or near/far state."""
# return if near or far event is detected
if self.gesture_state_ == APDS9960_STATE_NEAR:
@ -462,33 +453,26 @@ class APDS9960:
return True
# *******************************************************************************
# Getters and setters for register values
# *******************************************************************************
def getProxIntLowThresh(self):
"""Returns the lower threshold for proximity detection
"""
"""Returns the lower threshold for proximity detection"""
return self._read_byte_data(APDS9960_REG_PILT)
def setProxIntLowThresh(self, threshold):
"""Sets the lower threshold for proximity detection.
"""
"""Sets the lower threshold for proximity detection."""
self._write_byte_data(APDS9960_REG_PILT, threshold)
def getProxIntHighThresh(self):
"""Returns the high threshold for proximity detection.
"""
"""Returns the high threshold for proximity detection."""
return self._read_byte_data(APDS9960_REG_PIHT)
def setProxIntHighThresh(self, threshold):
"""Sets the high threshold for proximity detection.
"""
"""Sets the high threshold for proximity detection."""
self._write_byte_data(APDS9960_REG_PIHT, threshold)
def getLEDDrive(self):
"""Returns LED drive strength for proximity and ALS.
@ -528,7 +512,6 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_CONTROL, val)
def getProximityGain(self):
"""Returns receiver gain for proximity detection.
@ -568,7 +551,6 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_CONTROL, val)
def getAmbientLightGain(self):
"""Returns receiver gain for the ambient light sensor (ALS).
@ -584,7 +566,7 @@ class APDS9960:
val = self._read_byte_data(APDS9960_REG_CONTROL)
# shift and mask out ADRIVE bits
return (val & 0b00000011)
return val & 0b00000011
def setAmbientLightGain(self, drive):
"""Sets the receiver gain for the ambient light sensor (ALS).
@ -607,7 +589,6 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_CONTROL, val)
def getLEDBoost(self):
"""Get the current LED boost value.
@ -647,7 +628,6 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_CONFIG2, val)
def getProxGainCompEnable(self):
"""Gets proximity gain compensation enable.
@ -675,7 +655,6 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_CONFIG3, val)
def getProxPhotoMask(self):
"""Gets the current mask for enabled/disabled proximity photodiodes.
@ -718,7 +697,6 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_CONFIG3, val)
def getGestureEnterThresh(self):
"""Gets the entry proximity threshold for gesture sensing.
@ -735,7 +713,6 @@ class APDS9960:
"""
self._write_byte_data(APDS9960_REG_GPENTH, threshold)
def getGestureExitThresh(self):
"""Gets the exit proximity threshold for gesture sensing.
@ -752,7 +729,6 @@ class APDS9960:
"""
self._write_byte_data(APDS9960_REG_GEXTH, threshold)
def getGestureGain(self):
"""Gets the gain of the photodiode during gesture mode.
@ -792,7 +768,6 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_GCONF2, val)
def getGestureLEDDrive(self):
"""Gets the drive current of the LED during gesture mode.
@ -825,14 +800,13 @@ class APDS9960:
val = self._read_byte_data(APDS9960_REG_GCONF2)
# set bits in register to given value
drive &= 0b00000011;
drive = drive << 3;
val &= 0b11100111;
val |= drive;
drive &= 0b00000011
drive = drive << 3
val &= 0b11100111
val |= drive
self._write_byte_data(APDS9960_REG_GCONF2, val)
def getGestureWaitTime(self):
"""Gets the time in low power mode between gesture detections.
@ -879,14 +853,15 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_GCONF2, val)
def getLightIntLowThreshold(self):
"""Gets the low threshold for ambient light interrupts.
Returns:
int: threshold current low threshold stored on the APDS9960
"""
return self._read_byte_data(APDS9960_REG_AILTL) | (self._read_byte_data(APDS9960_REG_AILTH) << 8)
return self._read_byte_data(APDS9960_REG_AILTL) | (
self._read_byte_data(APDS9960_REG_AILTH) << 8
)
def setLightIntLowThreshold(self, threshold):
"""Sets the low threshold for ambient light interrupts.
@ -895,9 +870,8 @@ class APDS9960:
threshold (int): low threshold value for interrupt to trigger
"""
# break 16-bit threshold into 2 8-bit values
self._write_byte_data(APDS9960_REG_AILTL, threshold & 0x00ff)
self._write_byte_data(APDS9960_REG_AILTH, (threshold & 0xff00) >> 8)
self._write_byte_data(APDS9960_REG_AILTL, threshold & 0x00FF)
self._write_byte_data(APDS9960_REG_AILTH, (threshold & 0xFF00) >> 8)
def getLightIntHighThreshold(self):
"""Gets the high threshold for ambient light interrupts.
@ -905,7 +879,9 @@ class APDS9960:
Returns:
int: threshold current low threshold stored on the APDS9960
"""
return self._read_byte_data(APDS9960_REG_AIHTL) | (self._read_byte_data(APDS9960_REG_AIHTH) << 8)
return self._read_byte_data(APDS9960_REG_AIHTL) | (
self._read_byte_data(APDS9960_REG_AIHTH) << 8
)
def setLightIntHighThreshold(self, threshold):
"""Sets the high threshold for ambient light interrupts.
@ -914,9 +890,8 @@ class APDS9960:
threshold (int): high threshold value for interrupt to trigger
"""
# break 16-bit threshold into 2 8-bit values
self._write_byte_data(APDS9960_REG_AIHTL, threshold & 0x00ff)
self._write_byte_data(APDS9960_REG_AIHTH, (threshold & 0xff00) >> 8)
self._write_byte_data(APDS9960_REG_AIHTL, threshold & 0x00FF)
self._write_byte_data(APDS9960_REG_AIHTH, (threshold & 0xFF00) >> 8)
def getProximityIntLowThreshold(self):
"""Gets the low threshold for proximity interrupts.
@ -934,7 +909,6 @@ class APDS9960:
"""
self._write_byte_data(APDS9960_REG_PILT, threshold)
def getProximityIntHighThreshold(self):
"""Gets the high threshold for proximity interrupts.
@ -951,7 +925,6 @@ class APDS9960:
"""
self._write_byte_data(APDS9960_REG_PIHT, threshold)
def getAmbientLightIntEnable(self):
"""Gets if ambient light interrupts are enabled or not.
@ -970,13 +943,12 @@ class APDS9960:
val = self._read_byte_data(APDS9960_REG_ENABLE)
# set bits in register to given value
val &= 0b11101111;
val &= 0b11101111
if enable:
val |= 0b00010000
self._write_byte_data(APDS9960_REG_ENABLE, val)
def getProximityIntEnable(self):
"""Gets if proximity interrupts are enabled or not.
@ -995,13 +967,12 @@ class APDS9960:
val = self._read_byte_data(APDS9960_REG_ENABLE)
# set bits in register to given value
val &= 0b11011111;
val &= 0b11011111
if enable:
val |= 0b00100000
self._write_byte_data(APDS9960_REG_ENABLE, val)
def getGestureIntEnable(self):
"""Gets if gesture interrupts are enabled or not.
@ -1026,19 +997,14 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_GCONF4, val)
def clearAmbientLightInt(self):
"""Clears the ambient light interrupt.
"""
"""Clears the ambient light interrupt."""
self._read_byte_data(APDS9960_REG_AICLEAR)
def clearProximityInt(self):
"""Clears the proximity interrupt.
"""
"""Clears the proximity interrupt."""
self._read_byte_data(APDS9960_REG_PICLEAR)
def getGestureMode(self):
"""Tells if the gesture state machine is currently running.
@ -1063,7 +1029,6 @@ class APDS9960:
self._write_byte_data(APDS9960_REG_GCONF4, val)
# *******************************************************************************
# Raw I2C Reads and Writes
# *******************************************************************************
@ -1074,12 +1039,10 @@ class APDS9960:
def _write_byte_data(self, cmd, val):
return self.bus.write_byte_data(self.address, cmd, val)
def _read_i2c_block_data(self, cmd, num):
return self.bus.read_i2c_block_data(self.address, cmd, num)
class uAPDS9960(APDS9960):
"""
APDS9960 for MicroPython
@ -1087,6 +1050,7 @@ class uAPDS9960(APDS9960):
sensor = uAPDS9960(bus=I2C_instance,
address=APDS9960_I2C_ADDR, valid_id=APDS9960_DEV_ID)
"""
def _read_byte_data(self, cmd):
return self.bus.readfrom_mem(self.address, cmd, 1)[0]

View File

@ -1,6 +1,12 @@
class ADPS9960InvalidDevId(ValueError):
def __init__(self, id, valid_ids):
Exception.__init__(self, "Device id 0x{} is not a valied one (valid: {})!".format(format(id, '02x'), ', '.join(["0x{}".format(format(i, '02x')) for i in valid_ids])))
Exception.__init__(
self,
"Device id 0x{} is not a valied one (valid: {})!".format(
format(id, "02x"), ", ".join(["0x{}".format(format(i, "02x")) for i in valid_ids])
),
)
class ADPS9960InvalidMode(ValueError):
def __init__(self, mode):

View File

@ -11,9 +11,9 @@ MAGGYRO_MODE = 0x06
AMG_MODE = 0x07
IMUPLUS_MODE = 0x08
COMPASS_MODE = 0x09
M4G_MODE = 0x0a
NDOF_FMC_OFF_MODE = 0x0b
NDOF_MODE = 0x0c
M4G_MODE = 0x0A
NDOF_FMC_OFF_MODE = 0x0B
NDOF_MODE = 0x0C
AXIS_P0 = bytes([0x21, 0x04])
AXIS_P1 = bytes([0x24, 0x00])
@ -24,16 +24,17 @@ AXIS_P5 = bytes([0x21, 0x01])
AXIS_P6 = bytes([0x21, 0x07])
AXIS_P7 = bytes([0x24, 0x05])
_MODE_REGISTER = 0x3d
_POWER_REGISTER = 0x3e
_MODE_REGISTER = 0x3D
_POWER_REGISTER = 0x3E
_AXIS_MAP_CONFIG = 0x41
class BNO055:
def __init__(self, i2c, address=0x28, mode=NDOF_MODE, axis=AXIS_P4):
self.i2c = i2c
self.address = address
if self.read_id() != bytes([0xA0, 0xFB, 0x32, 0x0F]):
raise RuntimeError('Failed to find expected ID register values. Check wiring!')
raise RuntimeError("Failed to find expected ID register values. Check wiring!")
self.operation_mode(CONFIG_MODE)
self.system_trigger(0x20) # reset
pyb.delay(700)
@ -46,53 +47,68 @@ class BNO055:
pyb.delay(200)
def read_registers(self, register, size=1):
return(self.i2c.readfrom_mem(self.address, register, size))
return self.i2c.readfrom_mem(self.address, register, size)
def write_registers(self, register, data):
self.i2c.writeto_mem(self.address, register, data)
def operation_mode(self, mode=None):
if mode:
self.write_registers(_MODE_REGISTER, bytes([mode]))
else:
return(self.read_registers(_MODE_REGISTER, 1)[0])
return self.read_registers(_MODE_REGISTER, 1)[0]
def system_trigger(self, data):
self.write_registers(0x3f, bytes([data]))
self.write_registers(0x3F, bytes([data]))
def power_mode(self, mode=None):
if mode:
self.write_registers(_POWER_REGISTER, bytes([mode]))
else:
return(self.read_registers(_POWER_REGISTER, 1))
return self.read_registers(_POWER_REGISTER, 1)
def page(self, num=None):
if num:
self.write_registers(0x3f, bytes([num]))
self.write_registers(0x3F, bytes([num]))
else:
self.read_registers(0x3f)
self.read_registers(0x3F)
def temperature(self):
return(self.read_registers(0x34, 1)[0])
return self.read_registers(0x34, 1)[0]
def read_id(self):
return(self.read_registers(0x00, 4))
return self.read_registers(0x00, 4)
def axis(self, placement=None):
if placement:
self.write_registers(_AXIS_MAP_CONFIG, placement)
else:
return(self.read_registers(_AXIS_MAP_CONFIG, 2))
return self.read_registers(_AXIS_MAP_CONFIG, 2)
def quaternion(self):
data = struct.unpack("<hhhh", self.read_registers(0x20, 8))
return [d / (1 << 14) for d in data] # [w, x, y, z]
def euler(self):
data = struct.unpack("<hhh", self.read_registers(0x1A, 6))
return [d / 16 for d in data] # [yaw, roll, pitch]
def accelerometer(self):
data = struct.unpack("<hhh", self.read_registers(0x08, 6))
return [d / 100 for d in data] # [x, y, z]
def magnetometer(self):
data = struct.unpack("<hhh", self.read_registers(0x0E, 6))
return [d / 16 for d in data] # [x, y, z]
def gyroscope(self):
data = struct.unpack("<hhh", self.read_registers(0x14, 6))
return [d / 900 for d in data] # [x, y, z]
def linear_acceleration(self):
data = struct.unpack("<hhh", self.read_registers(0x28, 6))
return [d / 100 for d in data] # [x, y, z]
def gravity(self):
data = struct.unpack("<hhh", self.read_registers(0x2e, 6))
data = struct.unpack("<hhh", self.read_registers(0x2E, 6))
return [d / 100 for d in data] # [x, y, z]

View File

@ -10,7 +10,8 @@
import time
import struct
class HTS221():
class HTS221:
def __init__(self, i2c, data_rate=1, dev_addr=0x5F):
self.bus = i2c
self.odr = data_rate
@ -18,7 +19,7 @@ class HTS221():
# Set configuration register
# Humidity and temperature average configuration
self.bus.writeto_mem(self.slv_addr, 0x10, b'\x1B')
self.bus.writeto_mem(self.slv_addr, 0x10, b"\x1B")
# Set control register
# PD | BDU | ODR
@ -43,7 +44,7 @@ class HTS221():
self.T3 = self.read_reg(0x3E, 2)
def read_reg(self, reg_addr, size):
fmt = 'B' if size == 1 else 'H'
fmt = "B" if size == 1 else "H"
reg_addr = reg_addr if size == 1 else reg_addr | 0x80
return struct.unpack(fmt, self.bus.readfrom_mem(self.slv_addr, reg_addr, size))[0]
@ -55,4 +56,6 @@ class HTS221():
temp = self.read_reg(0x2A, 2)
if temp > 32767:
temp -= 65536
return ((self.T1 - self.T0) / 8.0) * (temp - self.T2) / (self.T3 - self.T2) + (self.T0 / 8.0)
return ((self.T1 - self.T0) / 8.0) * (temp - self.T2) / (self.T3 - self.T2) + (
self.T0 / 8.0
)

View File

@ -29,20 +29,44 @@ BAND_IN865 = 7
BAND_US915 = 8
BAND_US915_HYBRID = 9
CLASS_A = 'A'
CLASS_B = 'B'
CLASS_C = 'C'
CLASS_A = "A"
CLASS_B = "B"
CLASS_C = "C"
class LoraError(Exception): pass
class LoraErrorTimeout(LoraError): pass
class LoraErrorParam(LoraError): pass
class LoraErrorBusy(LoraError): pass
class LoraErrorOverflow(LoraError): pass
class LoraErrorNoNetwork(LoraError): pass
class LoraErrorRX(LoraError): pass
class LoraErrorUnknown(LoraError): pass
class Lora():
class LoraError(Exception):
pass
class LoraErrorTimeout(LoraError):
pass
class LoraErrorParam(LoraError):
pass
class LoraErrorBusy(LoraError):
pass
class LoraErrorOverflow(LoraError):
pass
class LoraErrorNoNetwork(LoraError):
pass
class LoraErrorRX(LoraError):
pass
class LoraErrorUnknown(LoraError):
pass
class Lora:
LoraErrors = {
"": LoraErrorTimeout, # empty buffer
"+ERR": LoraError,
@ -51,10 +75,12 @@ class Lora():
"+ERR_PARAM_OVERFLOW": LoraErrorOverflow,
"+ERR_NO_NETWORK": LoraErrorNoNetwork,
"+ERR_RX": LoraErrorRX,
"+ERR_UNKNOWN": LoraErrorUnknown
"+ERR_UNKNOWN": LoraErrorUnknown,
}
def __init__(self, uart=None, rst_pin=None, boot_pin=None, band=BAND_EU868, poll_ms=300000, debug=False):
def __init__(
self, uart=None, rst_pin=None, boot_pin=None, band=BAND_EU868, poll_ms=300000, debug=False
):
self.debug = debug
self.uart = uart
self.rst_pin = rst_pin
@ -79,28 +105,27 @@ class Lora():
def init_modem(self):
# Portenta vision shield configuration
if not self.rst_pin:
self.rst_pin = Pin('PC6', Pin.OUT_PP, Pin.PULL_UP, value=1)
self.rst_pin = Pin("PC6", Pin.OUT_PP, Pin.PULL_UP, value=1)
if not self.boot_pin:
self.boot_pin = Pin('PG7', Pin.OUT_PP, Pin.PULL_DOWN, value=0)
self.boot_pin = Pin("PG7", Pin.OUT_PP, Pin.PULL_DOWN, value=0)
if not self.uart:
self.uart = UART(8, 19200)
# self.uart = UART(1, 19200) # Use external module
self.uart.init(19200, bits=8, parity=None, stop=2, timeout=250, timeout_char=100)
def debug_print(self, data):
if self.debug:
print(data)
def is_arduino_firmware(self):
return 'ARD-078' in self.fw_version
return "ARD-078" in self.fw_version
def configure_class(self, _class):
self.send_command("+CLASS=", _class)
def configure_band(self, band):
self.send_command("+BAND=", band)
if (band == BAND_EU868 and self.is_arduino_firmware()):
if band == BAND_EU868 and self.is_arduino_firmware():
self.send_command("+DUTYCYCLE=", 1)
return True
@ -109,10 +134,10 @@ class Lora():
def set_autobaud(self, timeout=10000):
start = ticks_ms()
while ((ticks_ms() - start) < timeout):
if (self.send_command('', timeout=200, raise_error=False) == '+OK'):
while (ticks_ms() - start) < timeout:
if self.send_command("", timeout=200, raise_error=False) == "+OK":
sleep_ms(200)
while (self.uart.any()):
while self.uart.any():
self.uart.readchar()
return True
return False
@ -129,11 +154,11 @@ class Lora():
self.send_command("+FACNEW")
def restart(self):
if (self.set_autobaud() == False):
if self.set_autobaud() is False:
raise (LoraError("Failed to set autobaud"))
# Different delimiter as REBOOT response EVENT doesn't end with '\r'.
if (self.send_command("+REBOOT", delimiter="+EVENT=0,0", timeout=10000) != "+EVENT=0,0"):
if self.send_command("+REBOOT", delimiter="+EVENT=0,0", timeout=10000) != "+EVENT=0,0":
raise (LoraError("Failed to reboot module"))
sleep_ms(1000)
self.fw_version = self.get_fw_version()
@ -205,72 +230,74 @@ class Lora():
def join(self, timeout_ms):
if self.send_command("+JOIN", timeout=timeout_ms) != "+ACK":
return False
response = self.receive('\r', timeout=timeout_ms)
response = self.receive("\r", timeout=timeout_ms)
return response == "+EVENT=1,1"
def get_join_status(self):
return int(self.send_command("+NJS?")) == 1
def get_max_size(self):
if (self.is_arduino_firmware()):
if self.is_arduino_firmware():
return 64
return int(self.send_command("+MSIZE?", timeout=2000))
def poll(self):
if ((ticks_ms() - self.last_poll_ms) > self.poll_ms):
if (ticks_ms() - self.last_poll_ms) > self.poll_ms:
self.last_poll_ms = ticks_ms()
# Triggers a fake write
self.send_data('\0', True)
self.send_data("\0", True)
def send_data(self, buff, confirmed=True):
max_len = self.get_max_size()
if (len(buff) > max_len):
if len(buff) > max_len:
raise (LoraError("Packet exceeds max length"))
if self.send_command("+CTX " if confirmed else "+UTX ", len(buff), data=buff) != "+OK":
return False
if confirmed:
response = self.receive('\r', timeout=10000)
response = self.receive("\r", timeout=10000)
return response == "+ACK"
return True
def receive_data(self, timeout=1000):
response = self.receive('\r', timeout=timeout)
response = self.receive("\r", timeout=timeout)
if response.startswith("+RECV"):
params = response.split("=")[1].split(",")
port = params[0]
length = int(params[1])
dummy_data_length = 2 # Data starts with \n\n sequence
data = self.receive(max_bytes=length+dummy_data_length, timeout=timeout)[dummy_data_length:]
return {'port' : port, 'data' : data}
data = self.receive(max_bytes=length + dummy_data_length, timeout=timeout)[
dummy_data_length:
]
return {"port": port, "data": data}
def receive(self, delimiter=None, max_bytes=None, timeout=1000):
buf = []
start = ticks_ms()
while ((ticks_ms() - start) < timeout):
while (self.uart.any()):
while (ticks_ms() - start) < timeout:
while self.uart.any():
buf += chr(self.uart.readchar())
if max_bytes and len(buf) == max_bytes:
data = ''.join(buf)
data = "".join(buf)
self.debug_print(data)
return data
if (len(buf) and delimiter != None):
data = ''.join(buf)
trimmed = data[0:-1] if data[-1] == '\r' else data
if len(buf) and delimiter is not None:
data = "".join(buf)
trimmed = data[0:-1] if data[-1] == "\r" else data
if (isinstance(delimiter, str) and len(delimiter) == 1 and buf[-1] == delimiter):
if isinstance(delimiter, str) and len(delimiter) == 1 and buf[-1] == delimiter:
self.debug_print(trimmed)
return trimmed
if (isinstance(delimiter, str) and trimmed == delimiter):
if isinstance(delimiter, str) and trimmed == delimiter:
self.debug_print(trimmed)
return trimmed
if (isinstance(delimiter, list) and trimmed in delimiter):
if isinstance(delimiter, list) and trimmed in delimiter:
self.debug_print(trimmed)
return trimmed
data = ''.join(buf)
data = "".join(buf)
self.debug_print(data)
return data[0:-1] if len(data) != 0 and data[-1] == '\r' else data
return data[0:-1] if len(data) != 0 and data[-1] == "\r" else data
def available(self):
return self.uart.any()
@ -279,7 +306,7 @@ class Lora():
self.change_mode(MODE_OTAA)
self.send_command("+APPEUI=", appEui)
self.send_command("+APPKEY=", appKey)
if (devEui):
if devEui:
self.send_command("+DEVEUI=", devEui)
network_joined = self.join(timeout)
# This delay was in MKRWAN.h
@ -299,13 +326,13 @@ class Lora():
def handle_error(self, command, data):
if not data.startswith("+ERR") and data != "":
return
if (data in self.LoraErrors):
if data in self.LoraErrors:
raise (self.LoraErrors[data]('Command "%s" has failed!' % command))
raise (LoraError('Command: "%s" failed with unknown status: "%s"' % (command, data)))
def send_command(self, cmd, *args, delimiter='\r', data=None, timeout=1000, raise_error=True):
def send_command(self, cmd, *args, delimiter="\r", data=None, timeout=1000, raise_error=True):
# Write command and args
uart_cmd = 'AT'+cmd+''.join([str(x) for x in args])+'\r'
uart_cmd = "AT" + cmd + "".join([str(x) for x in args]) + "\r"
self.debug_print(uart_cmd)
self.uart.write(uart_cmd)
@ -321,6 +348,6 @@ class Lora():
if raise_error:
self.handle_error(cmd, response)
if cmd.endswith('?'):
return response.split('=')[1]
if cmd.endswith("?"):
return response.split("=")[1]
return response

View File

@ -5,7 +5,8 @@
# v1.0 2016.4
# v2.0 2019.7
class LPS22H():
class LPS22H:
LPS22_CTRL_REG1 = const(0x10)
LPS22_CTRL_REG2 = const(0x11)
LPS22_STATUS = const(0x27)
@ -30,8 +31,10 @@ class LPS22H():
else:
self.getreg(LPS22_CTRL_REG1)
self.oneshot = oneshot
if oneshot: self.rb[0] &= 0x0F
else: self.rb[0] |= 0x10
if oneshot:
self.rb[0] &= 0x0F
else:
self.rb[0] |= 0x10
self.setreg(LPS22_CTRL_REG1, self.rb[0])
def int16(self, d):
@ -71,7 +74,11 @@ class LPS22H():
return self.pressure_irq()
def altitude(self):
return (((1013.25 / self.pressure())**(1/5.257)) - 1.0) * (self.temperature() + 273.15) / 0.0065
return (
(((1013.25 / self.pressure()) ** (1 / 5.257)) - 1.0)
* (self.temperature() + 273.15)
/ 0.0065
)
def temperature_irq(self):
self.ONE_SHOT(2)

View File

@ -143,7 +143,7 @@ class LSM6DSOX:
self.reset()
# Load and configure MLC if UCF file is provided
if ucf != None:
if ucf is not None:
self.load_mlc(ucf)
# Set Gyroscope datarate and scale.

View File

@ -45,17 +45,18 @@ while (True):
"""
import array
class LSM9DS1:
WHO_AM_I = const(0xf)
WHO_AM_I = const(0xF)
CTRL_REG1_G = const(0x10)
INT_GEN_SRC_G = const(0x14)
OUT_TEMP = const(0x15)
OUT_G = const(0x18)
CTRL_REG4_G = const(0x1e)
CTRL_REG4_G = const(0x1E)
STATUS_REG = const(0x27)
OUT_XL = const(0x28)
FIFO_CTRL_REG = const(0x2e)
FIFO_SRC = const(0x2f)
FIFO_CTRL_REG = const(0x2E)
FIFO_SRC = const(0x2F)
OFFSET_REG_X_M = const(0x05)
CTRL_REG1_M = const(0x20)
OUT_M = const(0x28)
@ -67,12 +68,13 @@ class LSM9DS1:
self.address_gyro = address_gyro
self.address_magnet = address_magnet
# check id's of accelerometer/gyro and magnetometer
if (self.read_id_magnet() != b'=') or (self.read_id_gyro() != b'h'):
raise OSError("Invalid LSM9DS1 device, using address {}/{}".format(
address_gyro,address_magnet))
if (self.read_id_magnet() != b"=") or (self.read_id_gyro() != b"h"):
raise OSError(
"Invalid LSM9DS1 device, using address {}/{}".format(address_gyro, address_magnet)
)
# allocate scratch buffer for efficient conversions and memread op's
self.scratch = array.array('B',[0,0,0,0,0,0])
self.scratch_int = array.array('h',[0,0,0])
self.scratch = array.array("B", [0, 0, 0, 0, 0, 0])
self.scratch_int = array.array("h", [0, 0, 0])
self.init_gyro_accel()
self.init_magnetometer()
@ -91,7 +93,7 @@ class LSM9DS1:
mv = memoryview(self.scratch)
# angular control registers 1-3 / Orientation
mv[0] = ((sample_rate & 0x07) << 5) | ((self.SCALE_GYRO[scale_gyro][1] & 0x3) << 3)
mv[1:4] = b'\x00\x00\x00'
mv[1:4] = b"\x00\x00\x00"
i2c.writeto_mem(addr, CTRL_REG1_G, mv[:5])
# ctrl4 - enable x,y,z, outputs, no irq latching, no 4D
# ctrl5 - enable all axes, no decimation
@ -106,8 +108,8 @@ class LSM9DS1:
i2c.writeto_mem(addr, CTRL_REG4_G, mv[:6])
# fifo: use continous mode (overwrite old data if overflow)
i2c.writeto_mem(addr, FIFO_CTRL_REG, b'\x00')
i2c.writeto_mem(addr, FIFO_CTRL_REG, b'\xc0')
i2c.writeto_mem(addr, FIFO_CTRL_REG, b"\x00")
i2c.writeto_mem(addr, FIFO_CTRL_REG, b"\xc0")
self.scale_gyro = 32768 / self.SCALE_GYRO[scale_gyro][0]
self.scale_accel = 32768 / self.SCALE_ACCEL[scale_accel][0]
@ -137,11 +139,11 @@ class LSM9DS1:
"""
offset = [int(i * self.scale_factor_magnet) for i in offset]
mv = memoryview(self.scratch)
mv[0] = offset[0] & 0xff
mv[0] = offset[0] & 0xFF
mv[1] = offset[0] >> 8
mv[2] = offset[1] & 0xff
mv[2] = offset[1] & 0xFF
mv[3] = offset[1] >> 8
mv[4] = offset[2] & 0xff
mv[4] = offset[2] & 0xFF
mv[5] = offset[2] >> 8
self.i2c.writeto_mem(self.address_magnet, OFFSET_REG_X_M, mv[:6])
@ -177,8 +179,10 @@ class LSM9DS1:
def iter_accel_gyro(self):
"""A generator that returns tuples of (gyro,accelerometer) data from the fifo."""
while True:
fifo_state = int.from_bytes(self.i2c.readfrom_mem(self.address_gyro, FIFO_SRC, 1),'big')
if fifo_state & 0x3f:
fifo_state = int.from_bytes(
self.i2c.readfrom_mem(self.address_gyro, FIFO_SRC, 1), "big"
)
if fifo_state & 0x3F:
# print("Available samples=%d" % (fifo_state & 0x1f))
yield self.read_gyro(), self.read_accel()
else:

View File

@ -1,50 +1,283 @@
import struct
class ModbusRTU():
class ModbusRTU:
def __init__(self, uart, slave_id=0x01, register_num=30):
self.SLAVE_ID = slave_id
self.uart = uart
self.register_num = register_num
self.REGISTER = [0] * self.register_num
self.CRC16_TABLE = [
0x0000,0xC0C1,0xC181,0x0140,0xC301,0x03C0,0x0280,0xC241,0xC601,
0x06C0,0x0780,0xC741,0x0500,0xC5C1,0xC481,0x0440,0xCC01,0x0CC0,
0x0D80,0xCD41,0x0F00,0xCFC1,0xCE81,0x0E40,0x0A00,0xCAC1,0xCB81,
0x0B40,0xC901,0x09C0,0x0880,0xC841,0xD801,0x18C0,0x1980,0xD941,
0x1B00,0xDBC1,0xDA81,0x1A40,0x1E00,0xDEC1,0xDF81,0x1F40,0xDD01,
0x1DC0,0x1C80,0xDC41,0x1400,0xD4C1,0xD581,0x1540,0xD701,0x17C0,
0x1680,0xD641,0xD201,0x12C0,0x1380,0xD341,0x1100,0xD1C1,0xD081,
0x1040,0xF001,0x30C0,0x3180,0xF141,0x3300,0xF3C1,0xF281,0x3240,
0x3600,0xF6C1,0xF781,0x3740,0xF501,0x35C0,0x3480,0xF441,0x3C00,
0xFCC1,0xFD81,0x3D40,0xFF01,0x3FC0,0x3E80,0xFE41,0xFA01,0x3AC0,
0x3B80,0xFB41,0x3900,0xF9C1,0xF881,0x3840,0x2800,0xE8C1,0xE981,
0x2940,0xEB01,0x2BC0,0x2A80,0xEA41,0xEE01,0x2EC0,0x2F80,0xEF41,
0x2D00,0xEDC1,0xEC81,0x2C40,0xE401,0x24C0,0x2580,0xE541,0x2700,
0xE7C1,0xE681,0x2640,0x2200,0xE2C1,0xE381,0x2340,0xE101,0x21C0,
0x2080,0xE041,0xA001,0x60C0,0x6180,0xA141,0x6300,0xA3C1,0xA281,
0x6240,0x6600,0xA6C1,0xA781,0x6740,0xA501,0x65C0,0x6480,0xA441,
0x6C00,0xACC1,0xAD81,0x6D40,0xAF01,0x6FC0,0x6E80,0xAE41,0xAA01,
0x6AC0,0x6B80,0xAB41,0x6900,0xA9C1,0xA881,0x6840,0x7800,0xB8C1,
0xB981,0x7940,0xBB01,0x7BC0,0x7A80,0xBA41,0xBE01,0x7EC0,0x7F80,
0xBF41,0x7D00,0xBDC1,0xBC81,0x7C40,0xB401,0x74C0,0x7580,0xB541,
0x7700,0xB7C1,0xB681,0x7640,0x7200,0xB2C1,0xB381,0x7340,0xB101,
0x71C0,0x7080,0xB041,0x5000,0x90C1,0x9181,0x5140,0x9301,0x53C0,
0x5280,0x9241,0x9601,0x56C0,0x5780,0x9741,0x5500,0x95C1,0x9481,
0x5440,0x9C01,0x5CC0,0x5D80,0x9D41,0x5F00,0x9FC1,0x9E81,0x5E40,
0x5A00,0x9AC1,0x9B81,0x5B40,0x9901,0x59C0,0x5880,0x9841,0x8801,
0x48C0,0x4980,0x8941,0x4B00,0x8BC1,0x8A81,0x4A40,0x4E00,0x8EC1,
0x8F81,0x4F40,0x8D01,0x4DC0,0x4C80,0x8C41,0x4400,0x84C1,0x8581,
0x4540,0x8701,0x47C0,0x4680,0x8641,0x8201,0x42C0,0x4380,0x8341,
0x4100,0x81C1,0x8081,0x4040]
0x0000,
0xC0C1,
0xC181,
0x0140,
0xC301,
0x03C0,
0x0280,
0xC241,
0xC601,
0x06C0,
0x0780,
0xC741,
0x0500,
0xC5C1,
0xC481,
0x0440,
0xCC01,
0x0CC0,
0x0D80,
0xCD41,
0x0F00,
0xCFC1,
0xCE81,
0x0E40,
0x0A00,
0xCAC1,
0xCB81,
0x0B40,
0xC901,
0x09C0,
0x0880,
0xC841,
0xD801,
0x18C0,
0x1980,
0xD941,
0x1B00,
0xDBC1,
0xDA81,
0x1A40,
0x1E00,
0xDEC1,
0xDF81,
0x1F40,
0xDD01,
0x1DC0,
0x1C80,
0xDC41,
0x1400,
0xD4C1,
0xD581,
0x1540,
0xD701,
0x17C0,
0x1680,
0xD641,
0xD201,
0x12C0,
0x1380,
0xD341,
0x1100,
0xD1C1,
0xD081,
0x1040,
0xF001,
0x30C0,
0x3180,
0xF141,
0x3300,
0xF3C1,
0xF281,
0x3240,
0x3600,
0xF6C1,
0xF781,
0x3740,
0xF501,
0x35C0,
0x3480,
0xF441,
0x3C00,
0xFCC1,
0xFD81,
0x3D40,
0xFF01,
0x3FC0,
0x3E80,
0xFE41,
0xFA01,
0x3AC0,
0x3B80,
0xFB41,
0x3900,
0xF9C1,
0xF881,
0x3840,
0x2800,
0xE8C1,
0xE981,
0x2940,
0xEB01,
0x2BC0,
0x2A80,
0xEA41,
0xEE01,
0x2EC0,
0x2F80,
0xEF41,
0x2D00,
0xEDC1,
0xEC81,
0x2C40,
0xE401,
0x24C0,
0x2580,
0xE541,
0x2700,
0xE7C1,
0xE681,
0x2640,
0x2200,
0xE2C1,
0xE381,
0x2340,
0xE101,
0x21C0,
0x2080,
0xE041,
0xA001,
0x60C0,
0x6180,
0xA141,
0x6300,
0xA3C1,
0xA281,
0x6240,
0x6600,
0xA6C1,
0xA781,
0x6740,
0xA501,
0x65C0,
0x6480,
0xA441,
0x6C00,
0xACC1,
0xAD81,
0x6D40,
0xAF01,
0x6FC0,
0x6E80,
0xAE41,
0xAA01,
0x6AC0,
0x6B80,
0xAB41,
0x6900,
0xA9C1,
0xA881,
0x6840,
0x7800,
0xB8C1,
0xB981,
0x7940,
0xBB01,
0x7BC0,
0x7A80,
0xBA41,
0xBE01,
0x7EC0,
0x7F80,
0xBF41,
0x7D00,
0xBDC1,
0xBC81,
0x7C40,
0xB401,
0x74C0,
0x7580,
0xB541,
0x7700,
0xB7C1,
0xB681,
0x7640,
0x7200,
0xB2C1,
0xB381,
0x7340,
0xB101,
0x71C0,
0x7080,
0xB041,
0x5000,
0x90C1,
0x9181,
0x5140,
0x9301,
0x53C0,
0x5280,
0x9241,
0x9601,
0x56C0,
0x5780,
0x9741,
0x5500,
0x95C1,
0x9481,
0x5440,
0x9C01,
0x5CC0,
0x5D80,
0x9D41,
0x5F00,
0x9FC1,
0x9E81,
0x5E40,
0x5A00,
0x9AC1,
0x9B81,
0x5B40,
0x9901,
0x59C0,
0x5880,
0x9841,
0x8801,
0x48C0,
0x4980,
0x8941,
0x4B00,
0x8BC1,
0x8A81,
0x4A40,
0x4E00,
0x8EC1,
0x8F81,
0x4F40,
0x8D01,
0x4DC0,
0x4C80,
0x8C41,
0x4400,
0x84C1,
0x8581,
0x4540,
0x8701,
0x47C0,
0x4680,
0x8641,
0x8201,
0x42C0,
0x4380,
0x8341,
0x4100,
0x81C1,
0x8081,
0x4040,
]
def any(self):
return self.uart.any()
def clear(self):
self.REGISTER = [0] * self.register_num
def crc16(self, data):
crc = 0xFFFF
for char in data:
crc = (crc >> 8) ^ self.CRC16_TABLE[((crc) ^ char) & 0xFF]
return struct.pack('<H',crc)
return struct.pack("<H", crc)
def handle(self, debug=False):
REQUEST = self.uart.read()
if debug:
@ -53,7 +286,7 @@ class ModbusRTU():
error_check = REQUEST[-2:]
function_code = REQUEST[1]
data = REQUEST[2:]
RESPONSE = struct.pack('b', self.SLAVE_ID)
RESPONSE = struct.pack("b", self.SLAVE_ID)
if self.crc16(REQUEST[:-2]) != error_check:
if debug:
print("crc not match")
@ -64,49 +297,49 @@ class ModbusRTU():
print("got cmd id: ", additional_address)
return 0 # do nothing
if function_code == 0x03:
starting_address = struct.unpack('>h', data[:2])[0]
quantity_of_registers = struct.unpack('>h', data[2:4])[0]
starting_address = struct.unpack(">h", data[:2])[0]
quantity_of_registers = struct.unpack(">h", data[2:4])[0]
response_registers = []
try:
for i in range(starting_address, starting_address + quantity_of_registers):
response_registers.append(self.REGISTER[i])
except IndexError as err:
RESPONSE += struct.pack('b', function_code|0x80)
RESPONSE += struct.pack('b', 0x02) # Illegal Data Address
RESPONSE += struct.pack("b", function_code | 0x80)
RESPONSE += struct.pack("b", 0x02) # Illegal Data Address
if debug:
print("Illegal Data Address: ")
print(err)
else:
RESPONSE += struct.pack('b', function_code)
RESPONSE += struct.pack('b', 2*quantity_of_registers)
RESPONSE += struct.pack("b", function_code)
RESPONSE += struct.pack("b", 2 * quantity_of_registers)
for b in response_registers:
RESPONSE += struct.pack('>h', b)
RESPONSE += struct.pack(">h", b)
elif function_code == 0x06:
register_address = struct.unpack('>h', data[:2])[0]
register_value = struct.unpack('>h', data[2:4])[0]
register_address = struct.unpack(">h", data[:2])[0]
register_value = struct.unpack(">h", data[2:4])[0]
try:
self.REGISTER[register_address] = register_value
except IndexError as err:
RESPONSE += struct.pack('b', function_code|0x80)
RESPONSE += struct.pack('b', 0x02) # Illegal Data Address
RESPONSE += struct.pack("b", function_code | 0x80)
RESPONSE += struct.pack("b", 0x02) # Illegal Data Address
if debug:
print("Illegal Data Address: ")
print(err)
else:
RESPONSE += struct.pack('b', function_code)
RESPONSE += struct.pack('>h', register_address)
RESPONSE += struct.pack('>h', self.REGISTER[register_address])
RESPONSE += struct.pack("b", function_code)
RESPONSE += struct.pack(">h", register_address)
RESPONSE += struct.pack(">h", self.REGISTER[register_address])
elif function_code == 0x10:
starting_address = struct.unpack('>h', data[:2])[0]
quantity_of_registers = struct.unpack('>h', data[2:4])[0]
byte_of_registers = struct.unpack('b', data[4:5])[0]
starting_address = struct.unpack(">h", data[:2])[0]
quantity_of_registers = struct.unpack(">h", data[2:4])[0]
byte_of_registers = struct.unpack("b", data[4:5])[0]
try:
if byte_of_registers != 2 * quantity_of_registers:
raise struct.error
values = struct.unpack('>%dh'%quantity_of_registers, data[5:])
values = struct.unpack(">%dh" % quantity_of_registers, data[5:])
except struct.error as err:
RESPONSE += struct.pack('b', function_code|0x80)
RESPONSE += struct.pack('b', 0x03) # Illegal Data Value
RESPONSE += struct.pack("b", function_code | 0x80)
RESPONSE += struct.pack("b", 0x03) # Illegal Data Value
if debug:
print("Illegal Data Value, data length error")
else:
@ -114,18 +347,18 @@ class ModbusRTU():
for i in range(quantity_of_registers):
self.REGISTER[starting_address + i] = values[i]
except IndexError as err:
RESPONSE += struct.pack('b', function_code|0x80)
RESPONSE += struct.pack('b', 0x02) # Illegal Data Address
RESPONSE += struct.pack("b", function_code | 0x80)
RESPONSE += struct.pack("b", 0x02) # Illegal Data Address
if debug:
print("Illegal Data Address: ")
print(err)
else:
RESPONSE += struct.pack('b', function_code)
RESPONSE += struct.pack('>h', starting_address)
RESPONSE += struct.pack('>h', quantity_of_registers)
RESPONSE += struct.pack("b", function_code)
RESPONSE += struct.pack(">h", starting_address)
RESPONSE += struct.pack(">h", quantity_of_registers)
else:
RESPONSE += struct.pack('b', function_code|0x80)
RESPONSE += struct.pack('b', 0x01) # Illegal Function
RESPONSE += struct.pack("b", function_code | 0x80)
RESPONSE += struct.pack("b", 0x01) # Illegal Function
RESPONSE += self.crc16(RESPONSE)
if debug:
print("FUNCTION CODE: ", function_code)

View File

@ -24,15 +24,24 @@
import usocket as socket
import ustruct as struct
from ubinascii import hexlify
class MQTTException(Exception):
pass
class MQTTClient:
def __init__(self, client_id, server, port=0, user=None, password=None, keepalive=0,
ssl=False, ssl_params={}):
class MQTTClient:
def __init__(
self,
client_id,
server,
port=0,
user=None,
password=None,
keepalive=0,
ssl=False,
ssl_params={},
):
if port == 0:
port = 8883 if ssl else 1883
self.client_id = client_id
@ -60,7 +69,7 @@ class MQTTClient:
sh = 0
while 1:
b = self.sock.read(1)[0]
n |= (b & 0x7f) << sh
n |= (b & 0x7F) << sh
if not b & 0x80:
return n
sh += 7
@ -83,6 +92,7 @@ class MQTTClient:
self.sock.connect(addr)
if self.ssl:
import ussl
self.sock = ussl.wrap_socket(self.sock, **self.ssl_params)
premsg = bytearray(b"\x10\0\0\0\0\0")
msg = bytearray(b"\x04MQTT\x04\x02\0\0")
@ -102,8 +112,8 @@ class MQTTClient:
msg[6] |= self.lw_retain << 5
i = 1
while sz > 0x7f:
premsg[i] = (sz & 0x7f) | 0x80
while sz > 0x7F:
premsg[i] = (sz & 0x7F) | 0x80
sz >>= 7
i += 1
premsg[i] = sz
@ -139,8 +149,8 @@ class MQTTClient:
sz += 2
assert sz < 2097152
i = 1
while sz > 0x7f:
pkt[i] = (sz & 0x7f) | 0x80
while sz > 0x7F:
pkt[i] = (sz & 0x7F) | 0x80
sz >>= 7
i += 1
pkt[i] = sz
@ -191,7 +201,7 @@ class MQTTClient:
# messages processed internally.
def wait_msg(self):
res = self.sock.read(1)
if res == b"" or res == None:
if res == b"" or res is None:
return None
self.sock.setblocking(True)
if res == b"\xd0": # PINGRESP
@ -199,7 +209,7 @@ class MQTTClient:
assert sz == 0
return None
op = res[0]
if op & 0xf0 != 0x30:
if op & 0xF0 != 0x30:
return op
sz = self._recv_len()
topic_len = self.sock.read(2)

View File

@ -1,9 +1,14 @@
import pyb, micropython, array, uctypes
import micropython
import array
import uctypes
micropython.alloc_emergency_exception_buf(100)
class MutexException(OSError):
pass
class Mutex:
@micropython.asm_thumb
def _acquire(r0, r1): # Spinlock: wait on the semaphore. Return on success.
@ -33,7 +38,7 @@ class Mutex:
cpsie(0) # enable interrupts
def __init__(self):
self.lock = array.array('i', (0,)) # 1 if a process has the lock else 0
self.lock = array.array("i", (0,)) # 1 if a process has the lock else 0
# POSIX API pthread_mutex_lock() blocks the thread till resource is available.
def __enter__(self):
@ -46,10 +51,9 @@ class Mutex:
# POSIX pthread_mutex_unlock()
def release(self):
if self.lock[0] == 0:
raise MutexException('Semaphore already released')
raise MutexException("Semaphore already released")
self.lock[0] = 0
# POSIX pthread_mutex_trylock() API. When mutex is not available the function returns immediately
def test(self): # Nonblocking: try to acquire, return True if success.
return self._attempt(uctypes.addressof(self.lock)) == 0

View File

@ -1,4 +1,4 @@
'''
"""
Example:
from pid import PID
pid1 = PID(p=0.07, i=0, imax=90)
@ -6,21 +6,23 @@ while(True):
error = 50 #error should be caculated, target - mesure
output=pid1.get_pid(error,1)
#control value with output
'''
"""
from pyb import millis
from math import pi, isnan
class PID:
_kp = _ki = _kd = _integrator = _imax = 0
_last_error = _last_derivative = _last_t = 0
_RC = 1 / (2 * pi * 20)
def __init__(self, p=0, i=0, d=0, imax=0):
self._kp = float(p)
self._ki = float(i)
self._kd = float(d)
self._imax = abs(imax)
self._last_derivative = float('nan')
self._last_derivative = float("nan")
def get_pid(self, error, scaler):
tnow = millis()
@ -38,19 +40,22 @@ class PID:
self._last_derivative = 0
else:
derivative = (error - self._last_error) / delta_time
derivative = self._last_derivative + \
((delta_time / (self._RC + delta_time)) * \
(derivative - self._last_derivative))
derivative = self._last_derivative + (
(delta_time / (self._RC + delta_time)) * (derivative - self._last_derivative)
)
self._last_error = error
self._last_derivative = derivative
output += self._kd * derivative
output *= scaler
if abs(self._ki) > 0 and dt > 0:
self._integrator += (error * self._ki) * scaler * delta_time
if self._integrator < -self._imax: self._integrator = -self._imax
elif self._integrator > self._imax: self._integrator = self._imax
if self._integrator < -self._imax:
self._integrator = -self._imax
elif self._integrator > self._imax:
self._integrator = self._imax
output += self._integrator
return output
def reset_I(self):
self._integrator = 0
self._last_derivative = float('nan')
self._last_derivative = float("nan")

View File

@ -5,7 +5,13 @@
#
# This work is licensed under the MIT license, see the file LICENSE for details.
import gc, network, omv, pyb, select, socket, stm, struct
import gc
import omv
import pyb
import select
import socket
import stm
import struct
class rpc:

View File

@ -5,7 +5,10 @@
#
# This work is licensed under the MIT license, see the file LICENSE for details.
import network, pyb, re, socket, struct
import pyb
import re
import socket
import struct
class rtsp_server:

View File

@ -4,22 +4,23 @@ from pyb import SPI
# register definitions
SET_CONTRAST = const(0x81)
SET_ENTIRE_ON = const(0xa4)
SET_NORM_INV = const(0xa6)
SET_DISP = const(0xae)
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)
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):
@ -37,25 +38,35 @@ class SSD1306:
for cmd in (
SET_DISP | 0x00, # off
# address setting
SET_MEM_ADDR, 0x00, # horizontal
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_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,
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
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_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
SET_CHARGE_PUMP,
0x10 if self.external_vcc else 0x14,
SET_DISP | 0x01,
): # on
self.write_cmd(cmd)
self.fill(0)
self.show()
@ -97,8 +108,9 @@ class SSD1306:
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):
def __init__(self, width, height, i2c, addr=0x3C, external_vcc=False):
self.i2c = i2c
self.addr = addr
self.temp = bytearray(2)
@ -118,7 +130,6 @@ class SSD1306_I2C(SSD1306):
self.i2c.stop()
class SSD1306_SPI(SSD1306):
def __init__(self, width, height, spi, dc, res, cs, external_vcc=False):
self.rate = 10 * 1024 * 1024

View File

@ -1,17 +1,19 @@
import pyb
class Motor():
class Motor:
def __init__(self, channel):
if channel == 1:
self.pin1 = pyb.Pin('P3', pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin2 = pyb.Pin('P2', pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin1 = pyb.Pin("P3", pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin2 = pyb.Pin("P2", pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
tim = pyb.Timer(4, freq=1000)
self.power = tim.channel(1, pyb.Timer.PWM, pin=pyb.Pin("P7"), pulse_width_percent=0)
elif channel == 2:
self.pin1 = pyb.Pin('P1', pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin2 = pyb.Pin('P0', pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin1 = pyb.Pin("P1", pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin2 = pyb.Pin("P0", pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
tim = pyb.Timer(4, freq=1000)
self.power = tim.channel(2, pyb.Timer.PWM, pin=pyb.Pin("P8"), pulse_width_percent=0)
def set_speed(self, pwm):
if pwm < 0:
self.pin1.low()
@ -21,13 +23,14 @@ class Motor():
self.pin2.low()
self.power.pulse_width_percent(abs(pwm))
class Stepper():
class Stepper:
def __init__(self, stepnumber=200, rpms=2, power=50):
self.stepnumber = stepnumber
self.pin1 = pyb.Pin('P3', pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin2 = pyb.Pin('P2', pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin3 = pyb.Pin('P1', pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin4 = pyb.Pin('P0', pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin1 = pyb.Pin("P3", pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin2 = pyb.Pin("P2", pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin3 = pyb.Pin("P1", pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
self.pin4 = pyb.Pin("P0", pyb.Pin.OUT_PP, pyb.Pin.PULL_NONE)
tim = pyb.Timer(4, freq=1000)
self.power1 = tim.channel(1, pyb.Timer.PWM, pin=pyb.Pin("P7"), pulse_width_percent=0)
self.power2 = tim.channel(2, pyb.Timer.PWM, pin=pyb.Pin("P8"), pulse_width_percent=0)
@ -56,4 +59,3 @@ class Stepper():
self.pin3.value(phase[2])
self.pin4.value(phase[3])
pyb.udelay(self.delay_time)

View File

@ -42,22 +42,34 @@ verbose_l = 0
client_busy = False
# Interfaces: (IP-Address (string), IP-Address (integer), Netmask (integer))
_month_name = ("", "Jan", "Feb", "Mar", "Apr", "May", "Jun",
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec")
_month_name = (
"",
"Jan",
"Feb",
"Mar",
"Apr",
"May",
"Jun",
"Jul",
"Aug",
"Sep",
"Oct",
"Nov",
"Dec",
)
class FTP_client:
def __init__(self, ftpsocket, local_addr):
self.command_client, self.remote_addr = ftpsocket.accept()
self.remote_addr = self.remote_addr[0]
self.command_client.settimeout(_COMMAND_TIMEOUT)
log_msg(1, "FTP Command connection from:", self.remote_addr)
self.command_client.setsockopt(socket.SOL_SOCKET,
_SO_REGISTER_HANDLER,
self.exec_ftp_command)
self.command_client.setsockopt(
socket.SOL_SOCKET, _SO_REGISTER_HANDLER, self.exec_ftp_command
)
self.command_client.sendall("220 Hello, this is the {}.\r\n".format(sys.platform))
self.cwd = '/'
self.cwd = "/"
self.fromname = None
# self.logged_in = False
self.act_data_addr = self.remote_addr
@ -69,13 +81,12 @@ class FTP_client:
try:
for fname in uos.listdir(path):
data_client.sendall(self.make_description(path, fname, full))
except Exception as e: # path may be a file name or pattern
except Exception: # path may be a file name or pattern
path, pattern = self.split_path(path)
try:
for fname in uos.listdir(path):
if self.fncmp(fname, pattern):
data_client.sendall(
self.make_description(path, fname, full))
data_client.sendall(self.make_description(path, fname, full))
except:
pass
@ -83,20 +94,18 @@ class FTP_client:
global _month_name
if full:
stat = uos.stat(self.get_absolute_path(path, fname))
file_permissions = ("drwxr-xr-x"
if (stat[0] & 0o170000 == 0o040000)
else "-rw-r--r--")
file_permissions = "drwxr-xr-x" if (stat[0] & 0o170000 == 0o040000) else "-rw-r--r--"
file_size = stat[6]
tm = stat[7] & 0xffffffff
tm = stat[7] & 0xFFFFFFFF
tm = localtime(tm if tm < 0x80000000 else tm - 0x100000000)
if tm[0] != localtime()[0]:
description = "{} 1 owner group {:>10} {} {:2} {:>5} {}\r\n".\
format(file_permissions, file_size,
_month_name[tm[1]], tm[2], tm[0], fname)
description = "{} 1 owner group {:>10} {} {:2} {:>5} {}\r\n".format(
file_permissions, file_size, _month_name[tm[1]], tm[2], tm[0], fname
)
else:
description = "{} 1 owner group {:>10} {} {:2} {:02}:{:02} {}\r\n".\
format(file_permissions, file_size,
_month_name[tm[1]], tm[2], tm[3], tm[4], fname)
description = "{} 1 owner group {:>10} {} {:2} {:02}:{:02} {}\r\n".format(
file_permissions, file_size, _month_name[tm[1]], tm[2], tm[3], tm[4], fname
)
else:
description = fname + "\r\n"
return description
@ -125,22 +134,22 @@ class FTP_client:
# Just a few special cases "..", "." and ""
# If payload start's with /, set cwd to /
# and consider the remainder a relative path
if payload.startswith('/'):
if payload.startswith("/"):
cwd = "/"
for token in payload.split("/"):
if token == '..':
if token == "..":
cwd = self.split_path(cwd)[0]
elif token != '.' and token != '':
if cwd == '/':
elif token != "." and token != "":
if cwd == "/":
cwd += token
else:
cwd = cwd + '/' + token
cwd = cwd + "/" + token
return cwd
def split_path(self, path): # instead of path.rpartition('/')
tail = path.split('/')[-1]
tail = path.split("/")[-1]
head = path[: -(len(tail) + 1)]
return ('/' if head == '' else head, tail)
return ("/" if head == "" else head, tail)
# compare fname against pattern. Pattern may contain
# the wildcards ? and *.
@ -148,11 +157,11 @@ class FTP_client:
pi = 0
si = 0
while pi < len(pattern) and si < len(fname):
if (fname[si] == pattern[pi]) or (pattern[pi] == '?'):
if (fname[si] == pattern[pi]) or (pattern[pi] == "?"):
si += 1
pi += 1
else:
if pattern[pi] == '*': # recurse
if pattern[pi] == "*": # recurse
if pi == len(pattern.rstrip("*?")): # only wildcards left
return True
while si < len(fname):
@ -229,9 +238,9 @@ class FTP_client:
elif command == "SYST":
cl.sendall("215 UNIX Type: L8\r\n")
elif command in ("TYPE", "NOOP", "ABOR"): # just accept & ignore
cl.sendall('200 OK\r\n')
cl.sendall("200 OK\r\n")
elif command == "QUIT":
cl.sendall('221 Bye.\r\n')
cl.sendall("221 Bye.\r\n")
close_client(cl)
elif command == "PWD" or command == "XPWD":
cl.sendall('257 "{}"\r\n'.format(self.cwd))
@ -239,44 +248,44 @@ class FTP_client:
try:
if (uos.stat(path)[0] & 0o170000) == 0o040000:
self.cwd = path
cl.sendall('250 OK\r\n')
cl.sendall("250 OK\r\n")
else:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
elif command == "PASV":
cl.sendall('227 Entering Passive Mode ({},{},{}).\r\n'.format(
self.pasv_data_addr.replace('.', ','),
_DATA_PORT >> 8, _DATA_PORT % 256))
cl.sendall(
"227 Entering Passive Mode ({},{},{}).\r\n".format(
self.pasv_data_addr.replace(".", ","), _DATA_PORT >> 8, _DATA_PORT % 256
)
)
self.active = False
elif command == "PORT":
items = payload.split(",")
if len(items) >= 6:
self.act_data_addr = '.'.join(items[:4])
self.act_data_addr = ".".join(items[:4])
if self.act_data_addr == "127.0.1.1":
# replace by command session addr
self.act_data_addr = self.remote_addr
self.DATA_PORT = int(items[4]) * 256 + int(items[5])
cl.sendall('200 OK\r\n')
cl.sendall("200 OK\r\n")
self.active = True
else:
cl.sendall('504 Fail\r\n')
cl.sendall("504 Fail\r\n")
elif command == "LIST" or command == "NLST":
if payload.startswith("-"):
option = payload.split()[0].lower()
path = self.get_absolute_path(
self.cwd, payload[len(option):].lstrip())
path = self.get_absolute_path(self.cwd, payload[len(option) :].lstrip())
else:
option = ""
try:
data_client = self.open_dataclient()
cl.sendall("150 Directory listing:\r\n")
self.send_list_data(path, data_client,
command == "LIST" or 'l' in option)
self.send_list_data(path, data_client, command == "LIST" or "l" in option)
cl.sendall("226 Done.\r\n")
data_client.close()
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
if data_client is not None:
data_client.close()
elif command == "RETR":
@ -289,43 +298,47 @@ class FTP_client:
data_client = None
cl.sendall("226 Done.\r\n")
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
if data_client is not None:
data_client.close()
elif command == "STOR" or command == "APPE":
try:
data_client = self.open_dataclient()
cl.sendall("150 Opened data connection.\r\n")
self.save_file_data(path, data_client,
"wb" if command == "STOR" else "ab")
self.save_file_data(path, data_client, "wb" if command == "STOR" else "ab")
# if the next statement is reached,
# the data_client was closed.
data_client = None
cl.sendall("226 Done.\r\n")
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
if data_client is not None:
data_client.close()
elif command == "SIZE":
try:
cl.sendall('213 {}\r\n'.format(uos.stat(path)[6]))
cl.sendall("213 {}\r\n".format(uos.stat(path)[6]))
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
elif command == "MDTM":
try:
tm = localtime(uos.stat(path)[8])
cl.sendall('213 {:04d}{:02d}{:02d}{:02d}{:02d}{:02d}\r\n'.format(*tm[0:6]))
cl.sendall("213 {:04d}{:02d}{:02d}{:02d}{:02d}{:02d}\r\n".format(*tm[0:6]))
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
elif command == "STAT":
if payload == "":
cl.sendall("211-Connected to ({})\r\n"
cl.sendall(
"211-Connected to ({})\r\n"
" Data address ({})\r\n"
" TYPE: Binary STRU: File MODE: Stream\r\n"
" Session timeout {}\r\n"
"211 Client count is {}\r\n".format(
self.remote_addr, self.pasv_data_addr,
_COMMAND_TIMEOUT, len(client_list)))
self.remote_addr,
self.pasv_data_addr,
_COMMAND_TIMEOUT,
len(client_list),
)
)
else:
cl.sendall("213-Directory listing:\r\n")
self.send_list_data(path, cl, True)
@ -333,9 +346,9 @@ class FTP_client:
elif command == "DELE":
try:
uos.remove(path)
cl.sendall('250 OK\r\n')
cl.sendall("250 OK\r\n")
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
elif command == "RNFR":
try:
# just test if the name exists, exception if not
@ -343,35 +356,35 @@ class FTP_client:
self.fromname = path
cl.sendall("350 Rename from\r\n")
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
elif command == "RNTO":
try:
uos.rename(self.fromname, path)
cl.sendall('250 OK\r\n')
cl.sendall("250 OK\r\n")
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
self.fromname = None
elif command == "CDUP" or command == "XCUP":
self.cwd = self.get_absolute_path(self.cwd, "..")
cl.sendall('250 OK\r\n')
cl.sendall("250 OK\r\n")
elif command == "RMD" or command == "XRMD":
try:
uos.rmdir(path)
cl.sendall('250 OK\r\n')
cl.sendall("250 OK\r\n")
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
elif command == "MKD" or command == "XMKD":
try:
uos.mkdir(path)
cl.sendall('250 OK\r\n')
cl.sendall("250 OK\r\n")
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
elif command == "SITE":
try:
exec(payload.replace('\0','\n'))
cl.sendall('250 OK\r\n')
exec(payload.replace("\0", "\n"))
cl.sendall("250 OK\r\n")
except:
cl.sendall('550 Fail\r\n')
cl.sendall("550 Fail\r\n")
else:
cl.sendall("502 Unsupported command.\r\n")
# log_msg(2,
@ -422,8 +435,7 @@ def accept_ftp_connect(ftpsocket, local_addr):
def num_ip(ip):
items = ip.split(".")
return (int(items[0]) << 24 | int(items[1]) << 16 |
int(items[2]) << 8 | int(items[3]))
return int(items[0]) << 24 | int(items[1]) << 16 | int(items[2]) << 8 | int(items[3])
def stop():
@ -432,8 +444,7 @@ def stop():
global client_busy
for client in client_list:
client.command_client.setsockopt(socket.SOL_SOCKET,
_SO_REGISTER_HANDLER, None)
client.command_client.setsockopt(socket.SOL_SOCKET, _SO_REGISTER_HANDLER, None)
client.command_client.close()
del client_list
client_list = []
@ -470,19 +481,20 @@ def start(port=21, verbose=0, splash=True):
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(addr[0][4])
sock.listen(1)
sock.setsockopt(socket.SOL_SOCKET,
_SO_REGISTER_HANDLER,
lambda s : accept_ftp_connect(s, ifconfig[0]))
sock.setsockopt(
socket.SOL_SOCKET, _SO_REGISTER_HANDLER, lambda s: accept_ftp_connect(s, ifconfig[0])
)
ftpsockets.append(sock)
if splash:
print("FTP server started on {}:{}".format(ifconfig[0], port))
datasocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
datasocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
datasocket.bind(('0.0.0.0', _DATA_PORT))
datasocket.bind(("0.0.0.0", _DATA_PORT))
datasocket.listen(1)
datasocket.settimeout(10)
def restart(port=21, verbose=0, splash=True):
stop()
sleep_ms(200)

View File

@ -2,23 +2,25 @@
# copyright (c) 2018 Shawwwn <shawwwn1@gmail.com>
# License: MIT
# Internet Checksum Algorithm
# Author: Olav Morken
# https://github.com/olavmrk/python-ping/blob/master/ping.py
# @data: bytes
def checksum(data):
if len(data) & 0x1: # Odd number of bytes
data += b'\0'
data += b"\0"
cs = 0
for pos in range(0, len(data), 2):
b1 = data[pos]
b2 = data[pos + 1]
cs += (b1 << 8) + b2
while cs >= 0x10000:
cs = (cs & 0xffff) + (cs >> 16)
cs = ~cs & 0xffff
cs = (cs & 0xFFFF) + (cs >> 16)
cs = ~cs & 0xFFFF
return cs
def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
import utime
import uselect
@ -29,7 +31,7 @@ def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
# prepare packet
assert size >= 16, "pkt size too small"
pkt = b'Q'*size
pkt = b"Q" * size
pkt_desc = {
"type": uctypes.UINT8 | 0,
"code": uctypes.UINT8 | 1,
@ -84,9 +86,12 @@ def ping(host, count=4, timeout=5000, interval=10, quiet=False, size=64):
seq = h2.seq
if h2.type == 0 and h2.id == h.id and (seq in seqs): # 0: ICMP_ECHO_REPLY
t_elasped = (utime.ticks_us() - h2.timestamp) / 1000
ttl = ustruct.unpack('!B', resp_mv[8:9])[0] # time-to-live
ttl = ustruct.unpack("!B", resp_mv[8:9])[0] # time-to-live
n_recv += 1
not quiet and print("%u bytes from %s: icmp_seq=%u, ttl=%u, time=%f ms" % (len(resp), addr, seq, ttl, t_elasped))
not quiet and print(
"%u bytes from %s: icmp_seq=%u, ttl=%u, time=%f ms"
% (len(resp), addr, seq, ttl, t_elasped)
)
seqs.remove(seq)
if len(seqs) == 0:
finish = True

View File

@ -20,7 +20,8 @@
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
# Source: improved version of https://github.com/micropython/micropython-lib/blob/master/python-ecosys/urequests/urequests.py
# Source: improved version of:
# https://github.com/micropython/micropython-lib/blob/master/python-ecosys/urequests/urequests.py
# Some useful links for future updates:
# https://www.w3.org/TR/html401/interact/forms.html#h-17.13.4
# https://docs.python-requests.org/en/master/
@ -28,6 +29,7 @@
import usocket
import ubinascii
class Response:
def __init__(self, code, reason, headers=None, content=None):
self.encoding = "utf-8"
@ -46,23 +48,26 @@ class Response:
def json(self):
import ujson
return ujson.loads(self._content)
def readline(s):
l = bytearray()
while True:
try:
l += s.read(1)
if (l[-1] == b'\n'):
if l[-1] == b"\n":
break
except:
break
return l
def socket_readall(s):
buf = b''
buf = b""
while True:
recv = b''
recv = b""
try:
recv = s.recv(1)
except:
@ -72,6 +77,7 @@ def socket_readall(s):
buf += recv
return buf
def request(method, url, data=None, json=None, files=None, headers={}, auth=None, stream=None):
try:
proto, dummy, host, path = url.split("/", 3)
@ -82,6 +88,7 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
port = 80
elif proto == "https:":
import ussl
port = 443
else:
raise ValueError("Unsupported protocol: " + proto)
@ -91,7 +98,9 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
port = int(port)
if auth:
headers['Authorization'] = b'Basic %s'%(ubinascii.b2a_base64('%s:%s' %(auth[0], auth[1]))[0:-1])
headers["Authorization"] = b"Basic %s" % (
ubinascii.b2a_base64("%s:%s" % (auth[0], auth[1]))[0:-1]
)
resp_code = 0
resp_reason = None
@ -107,7 +116,7 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
s.write(b"%s /%s HTTP/1.0\r\n" % (method, path))
if not "Host" in headers:
if "Host" not in headers:
s.write(b"Host: %s\r\n" % host)
# Iterate over keys to avoid tuple alloc
@ -119,6 +128,7 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
if json is not None:
import ujson
data = ujson.dumps(json)
s.write(b"Content-Type: application/json\r\n")
@ -128,7 +138,10 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
s.write(b"Content-Type: multipart/form-data; boundary=%s\r\n" % (boundary))
for name, fileobj in files.items():
data += b"--%s\r\n" % (boundary)
data += b'Content-Disposition: form-data; name="%s"; filename="%s"\r\n\r\n' %(name, fileobj[0])
data += b'Content-Disposition: form-data; name="%s"; filename="%s"\r\n\r\n' % (
name,
fileobj[0],
)
data += fileobj[1].read()
data += b"\r\n"
data += b"\r\n--%s--\r\n" % (boundary)
@ -149,35 +162,40 @@ def request(method, url, data=None, json=None, files=None, headers={}, auth=None
raise ValueError("Unsupported " + l)
elif l.startswith(b"Location:") and not 200 <= status <= 299:
raise NotImplementedError("Redirects not yet supported")
if 'HTTPS' in l or 'HTTP' in l:
if "HTTPS" in l or "HTTP" in l:
sline = l.split(None, 2)
resp_code = int(sline[1])
resp_reason = sline[2].decode().rstrip() if len(sline) > 2 else ""
continue
resp_headers.append(l)
resp_headers = b'\r\n'.join(resp_headers)
content = b'\r\n'.join(response)
resp_headers = b"\r\n".join(resp_headers)
content = b"\r\n".join(response)
except OSError:
s.close()
raise
return Response(resp_code, resp_reason, resp_headers, content)
def head(url, **kw):
return request("HEAD", url, **kw)
def get(url, **kw):
return request("GET", url, **kw)
def post(url, **kw):
return request("POST", url, **kw)
def put(url, **kw):
return request("PUT", url, **kw)
def patch(url, **kw):
return request("PATCH", url, **kw)
def delete(url, **kw):
return request("DELETE", url, **kw)

View File

@ -10,6 +10,7 @@ from uio import IOBase
last_client_socket = None
server_socket = None
# Provide necessary functions for dupterm and replace telnet control characters that come in.
class TelnetWrapper(IOBase):
def __init__(self, socket):
@ -23,7 +24,7 @@ class TelnetWrapper(IOBase):
byte = 0
# discard telnet control characters and
# null bytes
while(byte == 0):
while byte == 0:
byte = self.socket.recv(1)[0]
if byte == 0xFF:
self.discard_count = 2
@ -63,6 +64,7 @@ class TelnetWrapper(IOBase):
def close(self):
self.socket.close()
# Attach new clients to dupterm and
# send telnet control characters to disable line mode
# and stop local echoing
@ -85,6 +87,7 @@ def accept_telnet_connect(telnet_server):
uos.dupterm(TelnetWrapper(last_client_socket))
def stop():
global server_socket, last_client_socket
uos.dupterm(None)
@ -93,6 +96,7 @@ def stop():
if last_client_socket:
last_client_socket.close()
# start listening for telnet connections on port 23
def start(port=23):
stop()

View File

@ -1,98 +1,106 @@
import pyb
VL51L1X_DEFAULT_CONFIGURATION = bytes([
0x00, # 0x2d : set bit 2 and 5 to 1 for fast plus mode (1MHz I2C), else don't touch */
0x00, # 0x2e : bit 0 if I2C pulled up at 1.8V, else set bit 0 to 1 (pull up at AVDD) */
0x00, # 0x2f : bit 0 if GPIO pulled up at 1.8V, else set bit 0 to 1 (pull up at AVDD) */
0x01, # 0x30 : set bit 4 to 0 for active high interrupt and 1 for active low (bits 3:0 must be 0x1), use SetInterruptPolarity() */
0x02, # 0x31 : bit 1 = interrupt depending on the polarity, use CheckForDataReady() */
0x00, # 0x32 : not user-modifiable */
0x02, # 0x33 : not user-modifiable */
0x08, # 0x34 : not user-modifiable */
0x00, # 0x35 : not user-modifiable */
0x08, # 0x36 : not user-modifiable */
0x10, # 0x37 : not user-modifiable */
0x01, # 0x38 : not user-modifiable */
0x01, # 0x39 : not user-modifiable */
0x00, # 0x3a : not user-modifiable */
0x00, # 0x3b : not user-modifiable */
0x00, # 0x3c : not user-modifiable */
0x00, # 0x3d : not user-modifiable */
0xff, # 0x3e : not user-modifiable */
0x00, # 0x3f : not user-modifiable */
0x0F, # 0x40 : not user-modifiable */
0x00, # 0x41 : not user-modifiable */
0x00, # 0x42 : not user-modifiable */
0x00, # 0x43 : not user-modifiable */
0x00, # 0x44 : not user-modifiable */
0x00, # 0x45 : not user-modifiable */
0x20, # 0x46 : interrupt configuration 0->level low detection, 1-> level high, 2-> Out of window, 3->In window, 0x20-> New sample ready , TBC */
0x0b, # 0x47 : not user-modifiable */
0x00, # 0x48 : not user-modifiable */
0x00, # 0x49 : not user-modifiable */
0x02, # 0x4a : not user-modifiable */
0x0a, # 0x4b : not user-modifiable */
0x21, # 0x4c : not user-modifiable */
0x00, # 0x4d : not user-modifiable */
0x00, # 0x4e : not user-modifiable */
0x05, # 0x4f : not user-modifiable */
0x00, # 0x50 : not user-modifiable */
0x00, # 0x51 : not user-modifiable */
0x00, # 0x52 : not user-modifiable */
0x00, # 0x53 : not user-modifiable */
0xc8, # 0x54 : not user-modifiable */
0x00, # 0x55 : not user-modifiable */
0x00, # 0x56 : not user-modifiable */
0x38, # 0x57 : not user-modifiable */
0xff, # 0x58 : not user-modifiable */
0x01, # 0x59 : not user-modifiable */
0x00, # 0x5a : not user-modifiable */
0x08, # 0x5b : not user-modifiable */
0x00, # 0x5c : not user-modifiable */
0x00, # 0x5d : not user-modifiable */
0x01, # 0x5e : not user-modifiable */
0xdb, # 0x5f : not user-modifiable */
0x0f, # 0x60 : not user-modifiable */
0x01, # 0x61 : not user-modifiable */
0xf1, # 0x62 : not user-modifiable */
0x0d, # 0x63 : not user-modifiable */
0x01, # 0x64 : Sigma threshold MSB (mm in 14.2 format for MSB+LSB), use SetSigmaThreshold(), default value 90 mm */
0x68, # 0x65 : Sigma threshold LSB */
0x00, # 0x66 : Min count Rate MSB (MCPS in 9.7 format for MSB+LSB), use SetSignalThreshold() */
0x80, # 0x67 : Min count Rate LSB */
0x08, # 0x68 : not user-modifiable */
0xb8, # 0x69 : not user-modifiable */
0x00, # 0x6a : not user-modifiable */
0x00, # 0x6b : not user-modifiable */
0x00, # 0x6c : Intermeasurement period MSB, 32 bits register, use SetIntermeasurementInMs() */
0x00, # 0x6d : Intermeasurement period */
0x0f, # 0x6e : Intermeasurement period */
0x89, # 0x6f : Intermeasurement period LSB */
0x00, # 0x70 : not user-modifiable */
0x00, # 0x71 : not user-modifiable */
0x00, # 0x72 : distance threshold high MSB (in mm, MSB+LSB), use SetD:tanceThreshold() */
0x00, # 0x73 : distance threshold high LSB */
0x00, # 0x74 : distance threshold low MSB ( in mm, MSB+LSB), use SetD:tanceThreshold() */
0x00, # 0x75 : distance threshold low LSB */
0x00, # 0x76 : not user-modifiable */
0x01, # 0x77 : not user-modifiable */
0x0f, # 0x78 : not user-modifiable */
0x0d, # 0x79 : not user-modifiable */
0x0e, # 0x7a : not user-modifiable */
0x0e, # 0x7b : not user-modifiable */
0x00, # 0x7c : not user-modifiable */
0x00, # 0x7d : not user-modifiable */
0x02, # 0x7e : not user-modifiable */
0xc7, # 0x7f : ROI center, use SetROI() */
0xff, # 0x80 : XY ROI (X=Width, Y=Height), use SetROI() */
0x9B, # 0x81 : not user-modifiable */
0x00, # 0x82 : not user-modifiable */
0x00, # 0x83 : not user-modifiable */
0x00, # 0x84 : not user-modifiable */
0x01, # 0x85 : not user-modifiable */
0x01, # 0x86 : clear interrupt, use ClearInterrupt() */
0x40 # 0x87 : start ranging, use StartRanging() or StopRanging(), If you want an automatic start after VL53L1X_init() call, put 0x40 in location 0x87 */
])
VL51L1X_DEFAULT_CONFIGURATION = bytes(
[
0x00, # 0x2d : set bit 2 and 5 to 1 for fast plus mode (1MHz I2C), else don't touch
0x00, # 0x2e : bit 0 if I2C pulled up at 1.8V, else set bit 0 to 1 (pull up at AVDD)
0x00, # 0x2f : bit 0 if GPIO pulled up at 1.8V, else set bit 0 to 1 (pull up at AVDD)
0x01, # 0x30 : set bit 4 to 0 for active high interrupt and 1 for active low
# (bits 3:0 must be 0x1), use SetInterruptPolarity()
0x02, # 0x31 : bit 1 = interrupt depending on the polarity, use CheckForDataReady()
0x00, # 0x32 : not user-modifiable
0x02, # 0x33 : not user-modifiable
0x08, # 0x34 : not user-modifiable
0x00, # 0x35 : not user-modifiable
0x08, # 0x36 : not user-modifiable
0x10, # 0x37 : not user-modifiable
0x01, # 0x38 : not user-modifiable
0x01, # 0x39 : not user-modifiable
0x00, # 0x3a : not user-modifiable
0x00, # 0x3b : not user-modifiable
0x00, # 0x3c : not user-modifiable
0x00, # 0x3d : not user-modifiable
0xFF, # 0x3e : not user-modifiable
0x00, # 0x3f : not user-modifiable
0x0F, # 0x40 : not user-modifiable
0x00, # 0x41 : not user-modifiable
0x00, # 0x42 : not user-modifiable
0x00, # 0x43 : not user-modifiable
0x00, # 0x44 : not user-modifiable
0x00, # 0x45 : not user-modifiable
0x20, # 0x46 : interrupt configuration 0->level low detection,
# 1-> level high, 2-> Out of window, 3->In window, 0x20-> New sample ready , TBC
0x0B, # 0x47 : not user-modifiable
0x00, # 0x48 : not user-modifiable
0x00, # 0x49 : not user-modifiable
0x02, # 0x4a : not user-modifiable
0x0A, # 0x4b : not user-modifiable
0x21, # 0x4c : not user-modifiable
0x00, # 0x4d : not user-modifiable
0x00, # 0x4e : not user-modifiable
0x05, # 0x4f : not user-modifiable
0x00, # 0x50 : not user-modifiable
0x00, # 0x51 : not user-modifiable
0x00, # 0x52 : not user-modifiable
0x00, # 0x53 : not user-modifiable
0xC8, # 0x54 : not user-modifiable
0x00, # 0x55 : not user-modifiable
0x00, # 0x56 : not user-modifiable
0x38, # 0x57 : not user-modifiable
0xFF, # 0x58 : not user-modifiable
0x01, # 0x59 : not user-modifiable
0x00, # 0x5a : not user-modifiable
0x08, # 0x5b : not user-modifiable
0x00, # 0x5c : not user-modifiable
0x00, # 0x5d : not user-modifiable
0x01, # 0x5e : not user-modifiable
0xDB, # 0x5f : not user-modifiable
0x0F, # 0x60 : not user-modifiable
0x01, # 0x61 : not user-modifiable
0xF1, # 0x62 : not user-modifiable
0x0D, # 0x63 : not user-modifiable
0x01, # 0x64 : Sigma threshold MSB (mm in 14.2 format for MSB+LSB),
# use SetSigmaThreshold(), default value 90 mm
0x68, # 0x65 : Sigma threshold LSB
0x00, # 0x66 : Min count Rate MSB (MCPS in 9.7 format for MSB+LSB), use SetSignalThreshold()
0x80, # 0x67 : Min count Rate LSB
0x08, # 0x68 : not user-modifiable
0xB8, # 0x69 : not user-modifiable
0x00, # 0x6a : not user-modifiable
0x00, # 0x6b : not user-modifiable
0x00, # 0x6c : Intermeasurement period MSB, 32 bits register, use SetIntermeasurementInMs()
0x00, # 0x6d : Intermeasurement period
0x0F, # 0x6e : Intermeasurement period
0x89, # 0x6f : Intermeasurement period LSB
0x00, # 0x70 : not user-modifiable
0x00, # 0x71 : not user-modifiable
0x00, # 0x72 : distance threshold high MSB (in mm, MSB+LSB), use SetD:tanceThreshold()
0x00, # 0x73 : distance threshold high LSB
0x00, # 0x74 : distance threshold low MSB ( in mm, MSB+LSB), use SetD:tanceThreshold()
0x00, # 0x75 : distance threshold low LSB
0x00, # 0x76 : not user-modifiable
0x01, # 0x77 : not user-modifiable
0x0F, # 0x78 : not user-modifiable
0x0D, # 0x79 : not user-modifiable
0x0E, # 0x7a : not user-modifiable
0x0E, # 0x7b : not user-modifiable
0x00, # 0x7c : not user-modifiable
0x00, # 0x7d : not user-modifiable
0x02, # 0x7e : not user-modifiable
0xC7, # 0x7f : ROI center, use SetROI()
0xFF, # 0x80 : XY ROI (X=Width, Y=Height), use SetROI()
0x9B, # 0x81 : not user-modifiable
0x00, # 0x82 : not user-modifiable
0x00, # 0x83 : not user-modifiable
0x00, # 0x84 : not user-modifiable
0x01, # 0x85 : not user-modifiable
0x01, # 0x86 : clear interrupt, use ClearInterrupt()
0x40, # 0x87 : start ranging, use StartRanging() or StopRanging(), If you want
# an automatic start after VL53L1X_init() call, put 0x40 in location 0x87
]
)
class VL53L1X:
def __init__(self, i2c, address=0x29):
self.i2c = i2c
@ -100,7 +108,7 @@ class VL53L1X:
self.reset()
pyb.delay(1)
if self.read_model_id() != 0xEACC:
raise RuntimeError('Failed to find expected ID register values. Check wiring!')
raise RuntimeError("Failed to find expected ID register values. Check wiring!")
# write default configuration
self.i2c.writeto_mem(self.address, 0x2D, VL51L1X_DEFAULT_CONFIGURATION, addrsize=16)
# pyb.delay(100)
@ -111,31 +119,39 @@ class VL53L1X:
def writeReg(self, reg, value):
return self.i2c.writeto_mem(self.address, reg, bytes([value]), addrsize=16)
def writeReg16Bit(self, reg, value):
return self.i2c.writeto_mem(self.address, reg, bytes([(value >> 8) & 0xFF, value & 0xFF]), addrsize=16)
return self.i2c.writeto_mem(
self.address, reg, bytes([(value >> 8) & 0xFF, value & 0xFF]), addrsize=16
)
def readReg(self, reg):
return self.i2c.readfrom_mem(self.address, reg, 1, addrsize=16)[0]
def readReg16Bit(self, reg):
data = self.i2c.readfrom_mem(self.address, reg, 2, addrsize=16)
return (data[0] << 8) + data[1]
def read_model_id(self):
return self.readReg16Bit(0x010F)
def reset(self):
self.writeReg(0x0000, 0x00)
pyb.delay(100)
self.writeReg(0x0000, 0x01)
def read(self):
data = self.i2c.readfrom_mem(self.address, 0x0089, 17, addrsize=16) # RESULT__RANGE_STATUS
range_status = data[0]
# range_status = data[0]
# report_status = data[1]
stream_count = data[2]
dss_actual_effective_spads_sd0 = (data[3]<<8) + data[4]
# stream_count = data[2]
# dss_actual_effective_spads_sd0 = (data[3] << 8) + data[4]
# peak_signal_count_rate_mcps_sd0 = (data[5]<<8) + data[6]
ambient_count_rate_mcps_sd0 = (data[7]<<8) + data[8]
# ambient_count_rate_mcps_sd0 = (data[7] << 8) + data[8]
# sigma_sd0 = (data[9]<<8) + data[10]
# phase_sd0 = (data[11]<<8) + data[12]
final_crosstalk_corrected_range_mm_sd0 = (data[13] << 8) + data[14]
peak_signal_count_rate_crosstalk_corrected_mcps_sd0 = (data[15]<<8) + data[16]
# peak_signal_count_rate_crosstalk_corrected_mcps_sd0 = (data[15] << 8) + data[16]
# status = None
# if range_status in (17, 2, 1, 3):
# status = "HardwareFail"