-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathscan.py
More file actions
66 lines (52 loc) · 1.95 KB
/
Copy pathscan.py
File metadata and controls
66 lines (52 loc) · 1.95 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
from time import sleep
import threading
class ScanManger:
def __init__(self, servoManager, bltManager, sweepRate=10):
self.servoManager = servoManager
self.bltManager = bltManager
self.sweepRate = sweepRate # angle changes in sec
# distance reading per angle (0..180). -1 means out of range / not yet scanned
self.distances = [-1] * 181
self.angle = 0
self.direction = 1 # +1 sweeps 0 -> 180, -1 sweeps 180 -> 0
self._lock = threading.Lock()
self._running = False
self._thread = None
def _record(self):
"""Move servo to current angle, settle, then store the latest distance."""
self.servoManager.move(self.angle)
sleep(1.0 / self.sweepRate)
d = self.bltManager.latest_distance
with self._lock:
self.distances[self.angle] = -1 if d is None else d
def _step(self):
"""Advance the sweep by one angle, recording a distance, bouncing at the ends."""
self._record()
self.angle += self.direction
if self.angle > 180:
self.angle = 179
self.direction = -1
elif self.angle < 0:
self.angle = 1
self.direction = 1
def _sweep_loop(self):
while self._running:
self._step()
def start(self):
"""Begin sweeping in the background."""
if self._running:
return
self._running = True
self._thread = threading.Thread(target=self._sweep_loop, daemon=True)
self._thread.start()
print("[msg] >>> scanning started.")
def stop(self):
self._running = False
print("[msg] >>> scanning stopped.")
def get_distances(self):
"""Snapshot of the per-angle distances (the display layer reads this)."""
with self._lock:
return list(self.distances)
def get_distance(self, angle):
with self._lock:
return self.distances[angle]