ch-10 clean
This commit is contained in:
parent
c632222255
commit
b595f3be54
@ -2,53 +2,54 @@ import time
|
||||
import robot
|
||||
import pid
|
||||
|
||||
|
||||
class FollowObject:
|
||||
def __init__(self):
|
||||
self.max_speed = 0.9
|
||||
self.follow_pid = pid.PID(0.1, 0.1, 0.015, 15)
|
||||
self.last_time = time.monotonic_ns()
|
||||
self.left_dist = 0
|
||||
self.pid_output = 0
|
||||
def __init__(self):
|
||||
self.max_speed = 0.9
|
||||
self.follow_pid = pid.PID(0.1, 0.1, 0.015, 15)
|
||||
self.last_time = time.monotonic()
|
||||
self.left_dist = 0
|
||||
self.pid_output = 0
|
||||
|
||||
def setup_robot(self):
|
||||
robot.left_distance.distance_mode = 1
|
||||
def setup_robot(self):
|
||||
robot.left_distance.distance_mode = 1
|
||||
|
||||
def movement_update(self):
|
||||
# do we have data
|
||||
if robot.left_distance.data_ready:
|
||||
self.left_dist = robot.left_distance.distance
|
||||
|
||||
# calculate time delta
|
||||
new_time = time.monotonic_ns()
|
||||
time_delta = new_time - self.last_time
|
||||
self.last_time = new_time
|
||||
def movement_update(self):
|
||||
# do we have data
|
||||
if robot.left_distance.data_ready:
|
||||
self.left_dist = robot.left_distance.distance
|
||||
|
||||
# get speeds from pid
|
||||
self.pid_output = self.follow_pid.update(self.left_dist, time_delta)
|
||||
speed = min(self.max_speed, self.pid_output)
|
||||
speed = max(-self.max_speed, speed)
|
||||
# calculate time delta
|
||||
new_time = time.monotonic()
|
||||
time_delta = new_time - self.last_time
|
||||
self.last_time = new_time
|
||||
|
||||
# make movements
|
||||
robot.set_left(speed)
|
||||
robot.set_right(speed)
|
||||
# get speeds from pid
|
||||
self.pid_output = self.follow_pid.update(self.left_dist, time_delta)
|
||||
speed = min(self.max_speed, self.pid_output)
|
||||
speed = max(-self.max_speed, speed)
|
||||
|
||||
# reset and loop
|
||||
robot.left_distance.clear_interrupt()
|
||||
# make movements
|
||||
robot.set_left(speed)
|
||||
robot.set_right(speed)
|
||||
|
||||
def main_loop(self):
|
||||
robot.left_distance.start_ranging()
|
||||
self.last_time = time.monotonic()
|
||||
while True:
|
||||
self.movement_update()
|
||||
# reset and loop
|
||||
robot.left_distance.clear_interrupt()
|
||||
|
||||
def main_loop(self):
|
||||
robot.left_distance.start_ranging()
|
||||
while True:
|
||||
self.movement_update()
|
||||
|
||||
def start(self):
|
||||
print("Starting")
|
||||
try:
|
||||
self.setup_robot()
|
||||
self.main_loop()
|
||||
finally:
|
||||
robot.stop()
|
||||
robot.left_distance.clear_interrupt()
|
||||
robot.left_distance.stop_ranging()
|
||||
|
||||
def start(self):
|
||||
print("Starting")
|
||||
try:
|
||||
self.setup_robot()
|
||||
self.main_loop()
|
||||
finally:
|
||||
robot.stop()
|
||||
robot.left_distance.clear_interrupt()
|
||||
robot.left_distance.stop_ranging()
|
||||
|
||||
FollowObject().start()
|
||||
|
||||
@ -1,23 +1,23 @@
|
||||
class PID:
|
||||
def __init__(self, proportional_k, integral_k, differential_k, set_point):
|
||||
self.proportional_k = proportional_k
|
||||
self.integral_k = integral_k
|
||||
self.differential_k = differential_k
|
||||
self.set_point = set_point
|
||||
def __init__(self, proportional_k, integral_k, differential_k, set_point):
|
||||
self.proportional_k = proportional_k
|
||||
self.integral_k = integral_k
|
||||
self.differential_k = differential_k
|
||||
self.set_point = set_point
|
||||
|
||||
self.error_sum = 0
|
||||
self.last_value = 0
|
||||
self.error_sum = 0
|
||||
self.last_value = 0
|
||||
|
||||
def update(self, measurement, time_delta):
|
||||
error_value = measurement - self.set_point
|
||||
proportional = error_value * self.proportional_k
|
||||
def update(self, measurement, time_delta):
|
||||
error_value = measurement - self.set_point
|
||||
proportional = error_value * self.proportional_k
|
||||
|
||||
# calculate integral
|
||||
self.error_sum += error_value * time_delta
|
||||
integral = self.error_sum * self.integral_k
|
||||
self.last_value = error_value
|
||||
# calculate integral
|
||||
self.error_sum += error_value * time_delta
|
||||
integral = self.error_sum * self.integral_k
|
||||
self.last_value = error_value
|
||||
|
||||
differentiated_error = (error_value - self.last_value) / time_delta
|
||||
differential = differentiated_error * self.differential_k
|
||||
differentiated_error = (error_value - self.last_value) / time_delta
|
||||
differential = differentiated_error * self.differential_k
|
||||
|
||||
return proportional + integral + differential
|
||||
return proportional + integral + differential
|
||||
|
||||
@ -54,24 +54,24 @@ send:
|
||||
|
||||
assembled = adafruit_pioasm.assemble(program)
|
||||
|
||||
class QuadratureEncoder:
|
||||
def __init__(self, first_pin, second_pin, reversed=False):
|
||||
"""Encoder with 2 pins. Must use sequential pins on the board"""
|
||||
self.sm = rp2pio.StateMachine(
|
||||
assembled,
|
||||
frequency=0,
|
||||
first_in_pin=first_pin,
|
||||
jmp_pin=second_pin,
|
||||
in_pin_count=2
|
||||
)
|
||||
self.reversed = reversed
|
||||
self._buffer = array.array('i', [0])
|
||||
|
||||
def read(self):
|
||||
while self.sm.in_waiting:
|
||||
self.sm.readinto(self._buffer)
|
||||
if self.reversed:
|
||||
return -self._buffer[0]
|
||||
else:
|
||||
return self._buffer[0]
|
||||
|
||||
class QuadratureEncoder:
|
||||
def __init__(self, first_pin, second_pin, reversed=False):
|
||||
"""Encoder with 2 pins. Must use sequential pins on the board"""
|
||||
self.sm = rp2pio.StateMachine(
|
||||
assembled,
|
||||
frequency=0,
|
||||
first_in_pin=first_pin,
|
||||
jmp_pin=second_pin,
|
||||
in_pin_count=2,
|
||||
)
|
||||
self.reversed = reversed
|
||||
self._buffer = array.array("i", [0])
|
||||
|
||||
def read(self):
|
||||
while self.sm.in_waiting:
|
||||
self.sm.readinto(self._buffer)
|
||||
if self.reversed:
|
||||
return -self._buffer[0]
|
||||
else:
|
||||
return self._buffer[0]
|
||||
|
||||
@ -29,6 +29,7 @@ def stop():
|
||||
motor_B1.duty_cycle = 0
|
||||
motor_B2.duty_cycle = 0
|
||||
|
||||
|
||||
def set_speed(motor, speed):
|
||||
# Swap motor pins if we reverse the speed
|
||||
if speed < 0:
|
||||
@ -36,15 +37,16 @@ def set_speed(motor, speed):
|
||||
speed = -speed
|
||||
else:
|
||||
direction = motor
|
||||
speed = min(speed, 1) # limit to 1.0
|
||||
max_speed = 2**16-1
|
||||
|
||||
speed = min(speed, 1) # limit to 1.0
|
||||
max_speed = 2 ** 16 - 1
|
||||
|
||||
direction[0].duty_cycle = int(max_speed * speed)
|
||||
direction[1].duty_cycle = 0
|
||||
|
||||
|
||||
def set_left(speed):
|
||||
set_speed(left_motor, speed)
|
||||
|
||||
|
||||
def set_right(speed):
|
||||
set_speed(right_motor, speed)
|
||||
|
||||
@ -10,88 +10,91 @@ import pid
|
||||
|
||||
|
||||
class FollowObject:
|
||||
def __init__(self):
|
||||
self.max_speed = 0.9
|
||||
self.follow_pid = pid.PID(0.1, 0.1, 0.015, 15)
|
||||
self.wifi = None
|
||||
self.server = None
|
||||
def __init__(self):
|
||||
self.max_speed = 0.9
|
||||
self.follow_pid = pid.PID(0.1, 0.1, 0.015, 15)
|
||||
self.wifi = None
|
||||
self.server = None
|
||||
|
||||
self.last_time = time.monotonic_ns()
|
||||
self.left_dist = 0
|
||||
self.pid_output = 0
|
||||
self.last_time = time.monotonic()
|
||||
self.left_dist = 0
|
||||
self.pid_output = 0
|
||||
|
||||
def setup_robot(self):
|
||||
robot.left_distance.distance_mode = 1
|
||||
def setup_robot(self):
|
||||
robot.left_distance.distance_mode = 1
|
||||
|
||||
def setup_wifi(self, app):
|
||||
print("Setting up wifi.")
|
||||
self.wifi, esp = robot_wifi.connect_to_wifi()
|
||||
self.server = adafruit_esp32spi_wsgiserver.WSGIServer(
|
||||
80,
|
||||
application=app
|
||||
)
|
||||
adafruit_esp32spi_wsgiserver.set_interface(esp)
|
||||
print("Starting server")
|
||||
def setup_wifi(self, app):
|
||||
print("Setting up wifi.")
|
||||
self.wifi, esp = robot_wifi.connect_to_wifi()
|
||||
self.server = adafruit_esp32spi_wsgiserver.WSGIServer(80, application=app)
|
||||
adafruit_esp32spi_wsgiserver.set_interface(esp)
|
||||
print("Starting server")
|
||||
|
||||
self.server.start()
|
||||
ip_int = ".".join(str(int(n)) for n in esp.ip_address)
|
||||
print(f"IP Address is {ip_int}")
|
||||
self.server.start()
|
||||
ip_int = ".".join(str(int(n)) for n in esp.ip_address)
|
||||
print(f"IP Address is {ip_int}")
|
||||
|
||||
def index(self, request):
|
||||
return 200, [('Content-Type', 'application/json')], [json.dumps(
|
||||
{
|
||||
"last_value": self.follow_pid.last_value,
|
||||
"pid_output": self.pid_output,
|
||||
"time": self.last_time
|
||||
}
|
||||
)]
|
||||
def index(self, request):
|
||||
return (
|
||||
200,
|
||||
[("Content-Type", "application/json")],
|
||||
[
|
||||
json.dumps(
|
||||
{
|
||||
"last_value": self.follow_pid.last_value,
|
||||
"pid_output": self.pid_output,
|
||||
"time": self.last_time,
|
||||
}
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
def movement_update(self):
|
||||
# do we have data
|
||||
if robot.left_distance.data_ready:
|
||||
self.left_dist = robot.left_distance.distance
|
||||
|
||||
# calculate time delta
|
||||
new_time = time.monotonic_ns()
|
||||
time_delta = new_time - self.last_time
|
||||
self.last_time = new_time
|
||||
def movement_update(self):
|
||||
# do we have data
|
||||
if robot.left_distance.data_ready:
|
||||
self.left_dist = robot.left_distance.distance
|
||||
|
||||
# get speeds from pid
|
||||
self.pid_output = self.follow_pid.update(self.left_dist, time_delta)
|
||||
speed = self.pid_output * self.max_speed
|
||||
# calculate time delta
|
||||
new_time = time.monotonic()
|
||||
time_delta = new_time - self.last_time
|
||||
self.last_time = new_time
|
||||
|
||||
# make movements
|
||||
robot.set_left(speed)
|
||||
robot.set_right(speed)
|
||||
# get speeds from pid
|
||||
self.pid_output = self.follow_pid.update(self.left_dist, time_delta)
|
||||
speed = self.pid_output * self.max_speed
|
||||
|
||||
# reset and loop
|
||||
robot.left_distance.clear_interrupt()
|
||||
# make movements
|
||||
robot.set_left(speed)
|
||||
robot.set_right(speed)
|
||||
|
||||
def main_loop(self):
|
||||
robot.left_distance.start_ranging()
|
||||
self.last_time = time.monotonic()
|
||||
while True:
|
||||
try:
|
||||
self.movement_update()
|
||||
self.server.update_poll()
|
||||
except RuntimeError as e:
|
||||
print(f"Server poll error: {type(e)}, {e}")
|
||||
robot.stop()
|
||||
print(f"Resetting ESP...")
|
||||
self.wifi.reset()
|
||||
print("Reset complete.")
|
||||
# reset and loop
|
||||
robot.left_distance.clear_interrupt()
|
||||
|
||||
def main_loop(self):
|
||||
robot.left_distance.start_ranging()
|
||||
while True:
|
||||
try:
|
||||
self.movement_update()
|
||||
self.server.update_poll()
|
||||
except RuntimeError as e:
|
||||
print(f"Server poll error: {type(e)}, {e}")
|
||||
robot.stop()
|
||||
print(f"Resetting ESP...")
|
||||
self.wifi.reset()
|
||||
print("Reset complete.")
|
||||
|
||||
def start(self):
|
||||
app = WSGIApp()
|
||||
app.route("/")(self.index)
|
||||
print("Starting")
|
||||
try:
|
||||
self.setup_robot()
|
||||
self.setup_wifi(app)
|
||||
self.main_loop()
|
||||
finally:
|
||||
robot.stop()
|
||||
robot.left_distance.clear_interrupt()
|
||||
robot.left_distance.stop_ranging()
|
||||
|
||||
def start(self):
|
||||
app = WSGIApp()
|
||||
app.route("/")(self.index)
|
||||
print("Starting")
|
||||
try:
|
||||
self.setup_robot()
|
||||
self.setup_wifi(app)
|
||||
self.main_loop()
|
||||
finally:
|
||||
robot.stop()
|
||||
robot.left_distance.clear_interrupt()
|
||||
robot.left_distance.stop_ranging()
|
||||
|
||||
FollowObject().start()
|
||||
|
||||
@ -2,41 +2,41 @@ from black import err
|
||||
|
||||
|
||||
class PID:
|
||||
def __init__(self, proportional_k, integral_k, differential_k, set_point):
|
||||
self.proportional_k = proportional_k
|
||||
self.integral_k = integral_k
|
||||
self.differential_k = differential_k
|
||||
self.set_point = set_point
|
||||
def __init__(self, proportional_k, integral_k, differential_k, set_point):
|
||||
self.proportional_k = proportional_k
|
||||
self.integral_k = integral_k
|
||||
self.differential_k = differential_k
|
||||
self.set_point = set_point
|
||||
|
||||
self.error_sum = 0
|
||||
self.last_value = 0
|
||||
self.min_output = -1
|
||||
self.max_output = 1
|
||||
self.error_sum = 0
|
||||
self.last_value = 0
|
||||
self.min_output = -1
|
||||
self.max_output = 1
|
||||
|
||||
self.dead_zone = 0.3
|
||||
self.dead_zone = 0.3
|
||||
|
||||
def update(self, measurement, time_delta):
|
||||
error_value = measurement - self.set_point
|
||||
proportional = error_value * self.proportional_k
|
||||
def update(self, measurement, time_delta):
|
||||
error_value = measurement - self.set_point
|
||||
proportional = error_value * self.proportional_k
|
||||
|
||||
# calculate integral
|
||||
self.error_sum += error_value * time_delta
|
||||
# clamp it
|
||||
self.error_sum = min(self.max_output, self.error_sum)
|
||||
self.error_sum = max(self.min_output, self.error_sum)
|
||||
# calculate integral
|
||||
self.error_sum += error_value * time_delta
|
||||
# clamp it
|
||||
self.error_sum = min(self.max_output, self.error_sum)
|
||||
self.error_sum = max(self.min_output, self.error_sum)
|
||||
|
||||
integral = self.error_sum * self.integral_k
|
||||
integral = self.error_sum * self.integral_k
|
||||
|
||||
differentiated_error = (error_value - self.last_value) / time_delta
|
||||
differential = differentiated_error * self.differential_k
|
||||
self.last_value = error_value
|
||||
differentiated_error = (error_value - self.last_value) / time_delta
|
||||
differential = differentiated_error * self.differential_k
|
||||
self.last_value = error_value
|
||||
|
||||
output = proportional + integral + differential
|
||||
# clamp output
|
||||
if abs(output) < self.dead_zone:
|
||||
output = 0
|
||||
else:
|
||||
output = min(self.max_output, output)
|
||||
output = max(self.min_output, output)
|
||||
output = proportional + integral + differential
|
||||
# clamp output
|
||||
if abs(output) < self.dead_zone:
|
||||
output = 0
|
||||
else:
|
||||
output = min(self.max_output, output)
|
||||
output = max(self.min_output, output)
|
||||
|
||||
return output
|
||||
return output
|
||||
|
||||
@ -54,24 +54,24 @@ send:
|
||||
|
||||
assembled = adafruit_pioasm.assemble(program)
|
||||
|
||||
class QuadratureEncoder:
|
||||
def __init__(self, first_pin, second_pin, reversed=False):
|
||||
"""Encoder with 2 pins. Must use sequential pins on the board"""
|
||||
self.sm = rp2pio.StateMachine(
|
||||
assembled,
|
||||
frequency=0,
|
||||
first_in_pin=first_pin,
|
||||
jmp_pin=second_pin,
|
||||
in_pin_count=2
|
||||
)
|
||||
self.reversed = reversed
|
||||
self._buffer = array.array('i', [0])
|
||||
|
||||
def read(self):
|
||||
while self.sm.in_waiting:
|
||||
self.sm.readinto(self._buffer)
|
||||
if self.reversed:
|
||||
return -self._buffer[0]
|
||||
else:
|
||||
return self._buffer[0]
|
||||
|
||||
class QuadratureEncoder:
|
||||
def __init__(self, first_pin, second_pin, reversed=False):
|
||||
"""Encoder with 2 pins. Must use sequential pins on the board"""
|
||||
self.sm = rp2pio.StateMachine(
|
||||
assembled,
|
||||
frequency=0,
|
||||
first_in_pin=first_pin,
|
||||
jmp_pin=second_pin,
|
||||
in_pin_count=2,
|
||||
)
|
||||
self.reversed = reversed
|
||||
self._buffer = array.array("i", [0])
|
||||
|
||||
def read(self):
|
||||
while self.sm.in_waiting:
|
||||
self.sm.readinto(self._buffer)
|
||||
if self.reversed:
|
||||
return -self._buffer[0]
|
||||
else:
|
||||
return self._buffer[0]
|
||||
|
||||
@ -29,6 +29,7 @@ def stop():
|
||||
motor_B1.duty_cycle = 0
|
||||
motor_B2.duty_cycle = 0
|
||||
|
||||
|
||||
def set_speed(motor, speed):
|
||||
# Swap motor pins if we reverse the speed
|
||||
if speed < 0:
|
||||
@ -36,15 +37,16 @@ def set_speed(motor, speed):
|
||||
speed = -speed
|
||||
else:
|
||||
direction = motor
|
||||
speed = min(speed, 1) # limit to 1.0
|
||||
max_speed = 2**16-1
|
||||
|
||||
speed = min(speed, 1) # limit to 1.0
|
||||
max_speed = 2 ** 16 - 1
|
||||
|
||||
direction[0].duty_cycle = int(max_speed * speed)
|
||||
direction[1].duty_cycle = 0
|
||||
|
||||
|
||||
def set_left(speed):
|
||||
set_speed(left_motor, speed)
|
||||
|
||||
|
||||
def set_right(speed):
|
||||
set_speed(right_motor, speed)
|
||||
|
||||
@ -5,21 +5,21 @@ from adafruit_esp32spi import adafruit_esp32spi
|
||||
from adafruit_esp32spi import adafruit_esp32spi_wifimanager
|
||||
|
||||
try:
|
||||
from secrets import secrets
|
||||
from secrets import secrets
|
||||
except ImportError:
|
||||
print("WiFi secrets are kept in secrets.py, please add them there!")
|
||||
raise
|
||||
print("WiFi secrets are kept in secrets.py, please add them there!")
|
||||
raise
|
||||
|
||||
|
||||
def connect_to_wifi():
|
||||
esp32_cs = DigitalInOut(board.GP10)
|
||||
esp32_ready = DigitalInOut(board.GP9)
|
||||
esp32_reset = DigitalInOut(board.GP8)
|
||||
esp32_cs = DigitalInOut(board.GP10)
|
||||
esp32_ready = DigitalInOut(board.GP9)
|
||||
esp32_reset = DigitalInOut(board.GP8)
|
||||
|
||||
spi = busio.SPI(board.GP14, MOSI=board.GP11, MISO=board.GP12)
|
||||
esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset)
|
||||
esp.reset()
|
||||
wifi = adafruit_esp32spi_wifimanager.ESPSPI_WiFiManager(esp, secrets)
|
||||
wifi.connect()
|
||||
spi = busio.SPI(board.GP14, MOSI=board.GP11, MISO=board.GP12)
|
||||
esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset)
|
||||
esp.reset()
|
||||
wifi = adafruit_esp32spi_wifimanager.ESPSPI_WiFiManager(esp, secrets)
|
||||
wifi.connect()
|
||||
|
||||
return wifi, esp
|
||||
return wifi, esp
|
||||
|
||||
@ -1,39 +1,39 @@
|
||||
class PID:
|
||||
def __init__(self, proportional_k, integral_k, differential_k, set_point):
|
||||
self.proportional_k = proportional_k
|
||||
self.integral_k = integral_k
|
||||
self.differential_k = differential_k
|
||||
self.set_point = set_point
|
||||
def __init__(self, proportional_k, integral_k, differential_k, set_point):
|
||||
self.proportional_k = proportional_k
|
||||
self.integral_k = integral_k
|
||||
self.differential_k = differential_k
|
||||
self.set_point = set_point
|
||||
|
||||
self.error_sum = 0
|
||||
self.last_value = 0
|
||||
self.min_output = -1
|
||||
self.max_output = 1
|
||||
self.error_sum = 0
|
||||
self.last_value = 0
|
||||
self.min_output = -1
|
||||
self.max_output = 1
|
||||
|
||||
self.dead_zone = 0.3
|
||||
self.dead_zone = 0.3
|
||||
|
||||
def update(self, measurement, time_delta):
|
||||
error_value = measurement - self.set_point
|
||||
proportional = error_value * self.proportional_k
|
||||
def update(self, measurement, time_delta):
|
||||
error_value = measurement - self.set_point
|
||||
proportional = error_value * self.proportional_k
|
||||
|
||||
# calculate integral
|
||||
self.error_sum += error_value * time_delta
|
||||
# clamp it
|
||||
self.error_sum = min(self.max_output, self.error_sum)
|
||||
self.error_sum = max(self.min_output, self.error_sum)
|
||||
# calculate integral
|
||||
self.error_sum += error_value * time_delta
|
||||
# clamp it
|
||||
self.error_sum = min(self.max_output, self.error_sum)
|
||||
self.error_sum = max(self.min_output, self.error_sum)
|
||||
|
||||
integral = self.error_sum * self.integral_k
|
||||
integral = self.error_sum * self.integral_k
|
||||
|
||||
differentiated_error = (error_value - self.last_value) / time_delta
|
||||
differential = differentiated_error * self.differential_k
|
||||
self.last_value = error_value
|
||||
differentiated_error = (error_value - self.last_value) / time_delta
|
||||
differential = differentiated_error * self.differential_k
|
||||
self.last_value = error_value
|
||||
|
||||
output = proportional + integral + differential
|
||||
# clamp output
|
||||
if abs(output) < self.dead_zone:
|
||||
output = 0
|
||||
else:
|
||||
output = min(self.max_output, output)
|
||||
output = max(self.min_output, output)
|
||||
output = proportional + integral + differential
|
||||
# clamp output
|
||||
if abs(output) < self.dead_zone:
|
||||
output = 0
|
||||
else:
|
||||
output = min(self.max_output, output)
|
||||
output = max(self.min_output, output)
|
||||
|
||||
return output
|
||||
return output
|
||||
|
||||
@ -54,24 +54,24 @@ send:
|
||||
|
||||
assembled = adafruit_pioasm.assemble(program)
|
||||
|
||||
class QuadratureEncoder:
|
||||
def __init__(self, first_pin, second_pin, reversed=False):
|
||||
"""Encoder with 2 pins. Must use sequential pins on the board"""
|
||||
self.sm = rp2pio.StateMachine(
|
||||
assembled,
|
||||
frequency=0,
|
||||
first_in_pin=first_pin,
|
||||
jmp_pin=second_pin,
|
||||
in_pin_count=2
|
||||
)
|
||||
self.reversed = reversed
|
||||
self._buffer = array.array('i', [0])
|
||||
|
||||
def read(self):
|
||||
while self.sm.in_waiting:
|
||||
self.sm.readinto(self._buffer)
|
||||
if self.reversed:
|
||||
return -self._buffer[0]
|
||||
else:
|
||||
return self._buffer[0]
|
||||
|
||||
class QuadratureEncoder:
|
||||
def __init__(self, first_pin, second_pin, reversed=False):
|
||||
"""Encoder with 2 pins. Must use sequential pins on the board"""
|
||||
self.sm = rp2pio.StateMachine(
|
||||
assembled,
|
||||
frequency=0,
|
||||
first_in_pin=first_pin,
|
||||
jmp_pin=second_pin,
|
||||
in_pin_count=2,
|
||||
)
|
||||
self.reversed = reversed
|
||||
self._buffer = array.array("i", [0])
|
||||
|
||||
def read(self):
|
||||
while self.sm.in_waiting:
|
||||
self.sm.readinto(self._buffer)
|
||||
if self.reversed:
|
||||
return -self._buffer[0]
|
||||
else:
|
||||
return self._buffer[0]
|
||||
|
||||
@ -29,6 +29,7 @@ def stop():
|
||||
motor_B1.duty_cycle = 0
|
||||
motor_B2.duty_cycle = 0
|
||||
|
||||
|
||||
def set_speed(motor, speed):
|
||||
# Swap motor pins if we reverse the speed
|
||||
if speed < 0:
|
||||
@ -36,15 +37,16 @@ def set_speed(motor, speed):
|
||||
speed = -speed
|
||||
else:
|
||||
direction = motor
|
||||
speed = min(speed, 1) # limit to 1.0
|
||||
max_speed = 2**16-1
|
||||
|
||||
speed = min(speed, 1) # limit to 1.0
|
||||
max_speed = 2 ** 16 - 1
|
||||
|
||||
direction[0].duty_cycle = int(max_speed * speed)
|
||||
direction[1].duty_cycle = 0
|
||||
|
||||
|
||||
def set_left(speed):
|
||||
set_speed(left_motor, speed)
|
||||
|
||||
|
||||
def set_right(speed):
|
||||
set_speed(right_motor, speed)
|
||||
|
||||
@ -5,21 +5,21 @@ from adafruit_esp32spi import adafruit_esp32spi
|
||||
from adafruit_esp32spi import adafruit_esp32spi_wifimanager
|
||||
|
||||
try:
|
||||
from secrets import secrets
|
||||
from secrets import secrets
|
||||
except ImportError:
|
||||
print("WiFi secrets are kept in secrets.py, please add them there!")
|
||||
raise
|
||||
print("WiFi secrets are kept in secrets.py, please add them there!")
|
||||
raise
|
||||
|
||||
|
||||
def connect_to_wifi():
|
||||
esp32_cs = DigitalInOut(board.GP10)
|
||||
esp32_ready = DigitalInOut(board.GP9)
|
||||
esp32_reset = DigitalInOut(board.GP8)
|
||||
esp32_cs = DigitalInOut(board.GP10)
|
||||
esp32_ready = DigitalInOut(board.GP9)
|
||||
esp32_reset = DigitalInOut(board.GP8)
|
||||
|
||||
spi = busio.SPI(board.GP14, MOSI=board.GP11, MISO=board.GP12)
|
||||
esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset)
|
||||
esp.reset()
|
||||
wifi = adafruit_esp32spi_wifimanager.ESPSPI_WiFiManager(esp, secrets)
|
||||
wifi.connect()
|
||||
spi = busio.SPI(board.GP14, MOSI=board.GP11, MISO=board.GP12)
|
||||
esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset)
|
||||
esp.reset()
|
||||
wifi = adafruit_esp32spi_wifimanager.ESPSPI_WiFiManager(esp, secrets)
|
||||
wifi.connect()
|
||||
|
||||
return wifi, esp
|
||||
return wifi, esp
|
||||
|
||||
@ -11,91 +11,94 @@ import robot_wifi
|
||||
|
||||
|
||||
class FollowWallApp:
|
||||
def __init__(self) -> None:
|
||||
self.speed = 0.6
|
||||
self.max_deflection = 0.4
|
||||
self.follow_pid = pid.PID(0.1, 0.5, 0, 15)
|
||||
self.follow_pid.dead_zone = 0.6
|
||||
def __init__(self) -> None:
|
||||
self.speed = 0.6
|
||||
self.max_deflection = 0.4
|
||||
self.follow_pid = pid.PID(0.1, 0.5, 0, 15)
|
||||
self.follow_pid.dead_zone = 0.6
|
||||
|
||||
self.wifi = None
|
||||
self.server = None
|
||||
self.wifi = None
|
||||
self.server = None
|
||||
|
||||
self.last_time = time.monotonic_ns()
|
||||
self.left_dist = 0
|
||||
self.pid_output = 0
|
||||
self.last_time = time.monotonic()
|
||||
self.left_dist = 0
|
||||
self.pid_output = 0
|
||||
|
||||
def setup_robot(self):
|
||||
robot.left_distance.distance_mode = 1
|
||||
def setup_robot(self):
|
||||
robot.left_distance.distance_mode = 1
|
||||
|
||||
def setup_wifi(self, app):
|
||||
print("Setting up wifi.")
|
||||
self.wifi, esp = robot_wifi.connect_to_wifi()
|
||||
self.server = adafruit_esp32spi_wsgiserver.WSGIServer(
|
||||
80,
|
||||
application=app
|
||||
)
|
||||
adafruit_esp32spi_wsgiserver.set_interface(esp)
|
||||
print("Starting server")
|
||||
def setup_wifi(self, app):
|
||||
print("Setting up wifi.")
|
||||
self.wifi, esp = robot_wifi.connect_to_wifi()
|
||||
self.server = adafruit_esp32spi_wsgiserver.WSGIServer(80, application=app)
|
||||
adafruit_esp32spi_wsgiserver.set_interface(esp)
|
||||
print("Starting server")
|
||||
|
||||
self.server.start()
|
||||
ip_int = ".".join(str(int(n)) for n in esp.ip_address)
|
||||
print(f"IP Address is {ip_int}")
|
||||
self.server.start()
|
||||
ip_int = ".".join(str(int(n)) for n in esp.ip_address)
|
||||
print(f"IP Address is {ip_int}")
|
||||
|
||||
def index(self, request):
|
||||
return 200, [('Content-Type', 'application/json')], [json.dumps(
|
||||
{
|
||||
"last_value": self.follow_pid.last_value,
|
||||
"pid_output": self.pid_output,
|
||||
"time": self.last_time
|
||||
}
|
||||
)]
|
||||
def index(self, request):
|
||||
return (
|
||||
200,
|
||||
[("Content-Type", "application/json")],
|
||||
[
|
||||
json.dumps(
|
||||
{
|
||||
"last_value": self.follow_pid.last_value,
|
||||
"pid_output": self.pid_output,
|
||||
"time": self.last_time,
|
||||
}
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
def movement_update(self):
|
||||
# do we have data
|
||||
if robot.left_distance.data_ready:
|
||||
self.left_dist = robot.left_distance.distance
|
||||
|
||||
# calculate time delta
|
||||
new_time = time.monotonic_ns()
|
||||
time_delta = new_time - self.last_time
|
||||
self.last_time = new_time
|
||||
def movement_update(self):
|
||||
# do we have data
|
||||
if robot.left_distance.data_ready:
|
||||
self.left_dist = robot.left_distance.distance
|
||||
|
||||
# get turn from pid
|
||||
self.pid_output = self.follow_pid.update(self.left_dist, time_delta)
|
||||
deflection = self.pid_output * self.max_deflection
|
||||
# calculate time delta
|
||||
new_time = time.monotonic()
|
||||
time_delta = new_time - self.last_time
|
||||
self.last_time = new_time
|
||||
|
||||
# make movements
|
||||
robot.set_left(self.speed - deflection)
|
||||
robot.set_right(self.speed + deflection)
|
||||
# get turn from pid
|
||||
self.pid_output = self.follow_pid.update(self.left_dist, time_delta)
|
||||
deflection = self.pid_output * self.max_deflection
|
||||
|
||||
# reset and loop
|
||||
robot.left_distance.clear_interrupt()
|
||||
# make movements
|
||||
robot.set_left(self.speed - deflection)
|
||||
robot.set_right(self.speed + deflection)
|
||||
|
||||
def main_loop(self):
|
||||
robot.left_distance.start_ranging()
|
||||
self.last_time = time.monotonic()
|
||||
while True:
|
||||
try:
|
||||
self.movement_update()
|
||||
self.server.update_poll()
|
||||
except RuntimeError as e:
|
||||
print(f"Server poll error: {type(e)}, {e}")
|
||||
robot.stop()
|
||||
print(f"Resetting ESP...")
|
||||
self.wifi.reset()
|
||||
print("Reset complete.")
|
||||
# reset and loop
|
||||
robot.left_distance.clear_interrupt()
|
||||
|
||||
def main_loop(self):
|
||||
robot.left_distance.start_ranging()
|
||||
while True:
|
||||
try:
|
||||
self.movement_update()
|
||||
self.server.update_poll()
|
||||
except RuntimeError as e:
|
||||
print(f"Server poll error: {type(e)}, {e}")
|
||||
robot.stop()
|
||||
print(f"Resetting ESP...")
|
||||
self.wifi.reset()
|
||||
print("Reset complete.")
|
||||
|
||||
def start(self):
|
||||
app = WSGIApp()
|
||||
app.route("/")(self.index)
|
||||
print("Starting")
|
||||
try:
|
||||
self.setup_robot()
|
||||
self.setup_wifi(app)
|
||||
self.main_loop()
|
||||
finally:
|
||||
robot.stop()
|
||||
robot.left_distance.clear_interrupt()
|
||||
robot.left_distance.stop_ranging()
|
||||
|
||||
def start(self):
|
||||
app = WSGIApp()
|
||||
app.route("/")(self.index)
|
||||
print("Starting")
|
||||
try:
|
||||
self.setup_robot()
|
||||
self.setup_wifi(app)
|
||||
self.main_loop()
|
||||
finally:
|
||||
robot.stop()
|
||||
robot.left_distance.clear_interrupt()
|
||||
robot.left_distance.stop_ranging()
|
||||
|
||||
FollowWallApp().start()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user