mirror of
https://github.com/undera/pylgbst.git
synced 2020-11-18 19:37:26 -08:00
Added tests
This commit is contained in:
parent
6067e4f218
commit
539715af8b
@ -79,6 +79,8 @@ class MoveHub(object):
|
||||
self._handle_port_status(data)
|
||||
elif msg_type == MSG_SENSOR_DATA:
|
||||
self._handle_sensor_data(data)
|
||||
elif msg_type == MSG_SENSOR_SUBSCRIBE_ACK:
|
||||
log.debug("Sensor subscribe ack on port %s", PORTS[get_byte(data, 3)])
|
||||
else:
|
||||
log.warning("Unhandled msg type 0x%x: %s", msg_type, str2hex(orig))
|
||||
else:
|
||||
|
@ -92,24 +92,26 @@ STATUS_FINISHED = 0x0a
|
||||
|
||||
# TILT
|
||||
TILT_SENSOR_MODE_FULL = 0x00
|
||||
TILT_SENSOR_MODE_SOME1 = 0x01
|
||||
TILT_SENSOR_MODE_2AXIS = 0x01
|
||||
TILT_SENSOR_MODE_BASIC = 0x02
|
||||
TILT_SENSOR_MODE_OFF = 0x03
|
||||
|
||||
TILT_HORIZ = 0x00
|
||||
TILT_HORIZONTAL = 0x00
|
||||
TILT_UP = 0x01
|
||||
TILT_DOWN = 0x02
|
||||
TILT_RIGHT = 0x03
|
||||
TILT_LEFT = 0x04
|
||||
TILT_INVERT = 0x05
|
||||
TILT_SOME = 0x09
|
||||
TILT_FRONT = 0x05
|
||||
TILT_SOME1 = 0x07
|
||||
TILT_SOME2 = 0x09
|
||||
|
||||
TILT_STATES = {
|
||||
TILT_HORIZ: "BACK",
|
||||
TILT_HORIZONTAL: "HORIZONTAL",
|
||||
TILT_UP: "UP",
|
||||
TILT_DOWN: "DOWN",
|
||||
TILT_RIGHT: "RIGHT",
|
||||
TILT_LEFT: "LEFT",
|
||||
TILT_INVERT: "FRONT",
|
||||
TILT_SOME: "SOME",
|
||||
TILT_FRONT: "FRONT",
|
||||
TILT_SOME1: "LEFT1",
|
||||
TILT_SOME2: "SOME2",
|
||||
}
|
||||
|
@ -22,7 +22,7 @@ class Peripheral(object):
|
||||
self.parent = parent
|
||||
self.port = port
|
||||
self.working = False
|
||||
self._subscribers = []
|
||||
self._subscribers = set()
|
||||
|
||||
def __repr__(self):
|
||||
return "%s on port %s" % (self.__class__.__name__, PORTS[self.port] if self.port in PORTS else 'N/A')
|
||||
@ -129,14 +129,20 @@ class ColorDistanceSensor(Peripheral):
|
||||
|
||||
|
||||
class TiltSensor(Peripheral):
|
||||
def _switch_mode(self, simple):
|
||||
self._subscribe_on_port(chr(simple) + b'\x01\x00\x00\x00\x01')
|
||||
def __init__(self, parent, port):
|
||||
super(TiltSensor, self).__init__(parent, port)
|
||||
self.mode = TILT_SENSOR_MODE_OFF
|
||||
|
||||
def _switch_mode(self, mode):
|
||||
self.mode = mode
|
||||
self._subscribe_on_port(chr(mode) + b'\x01\x00\x00\x00\x01')
|
||||
|
||||
def subscribe(self, callback, mode=TILT_SENSOR_MODE_BASIC):
|
||||
if mode not in (TILT_SENSOR_MODE_BASIC, TILT_SENSOR_MODE_SOME1, TILT_SENSOR_MODE_FULL):
|
||||
if mode not in (TILT_SENSOR_MODE_BASIC, TILT_SENSOR_MODE_2AXIS, TILT_SENSOR_MODE_FULL):
|
||||
raise ValueError("Wrong tilt sensor mode: 0x%x", mode)
|
||||
|
||||
self._switch_mode(mode)
|
||||
self._subscribers.append(callback) # TODO: maybe join it into `_subscribe_on_port`
|
||||
self._subscribers.add(callback) # TODO: maybe join it into `_subscribe_on_port`
|
||||
|
||||
# 1b0e00 0a00 47 3a020100000001
|
||||
# 1b0e00 0a00 47 3a020100000001
|
||||
@ -146,22 +152,31 @@ class TiltSensor(Peripheral):
|
||||
def unsubscribe(self, callback):
|
||||
self._subscribers.remove(callback)
|
||||
if not self._subscribers:
|
||||
self._switch_mode(3)
|
||||
self._switch_mode(TILT_SENSOR_MODE_OFF)
|
||||
|
||||
def handle_notification(self, data):
|
||||
if len(data) == 5:
|
||||
state = get_byte(data, 4)
|
||||
self._notify_subscribers(state)
|
||||
elif len(data) == 6:
|
||||
# TODO: how to interpret these 2 bytes?
|
||||
self._notify_subscribers(get_byte(data, 4), get_byte(data, 5))
|
||||
if self.mode == TILT_SENSOR_MODE_BASIC:
|
||||
self._notify_subscribers(get_byte(data, 4))
|
||||
elif self.mode == TILT_SENSOR_MODE_FULL:
|
||||
roll = self._byte2deg(get_byte(data, 4))
|
||||
pitch = self._byte2deg(get_byte(data, 5))
|
||||
self._notify_subscribers(roll, pitch)
|
||||
elif self.mode == TILT_SENSOR_MODE_2AXIS:
|
||||
# TODO: figure out right interpreting of this
|
||||
self._notify_subscribers(get_byte(data, 4))
|
||||
else:
|
||||
log.warning("Unexpected length for tilt sensor data: %s", len(data))
|
||||
log.debug("Got tilt sensor data while in finished mode: %s", self.mode)
|
||||
|
||||
def _byte2deg(self, val):
|
||||
if val > 90:
|
||||
return val - 256
|
||||
else:
|
||||
return val
|
||||
|
||||
|
||||
class Button(Peripheral):
|
||||
def __init__(self, parent):
|
||||
super(Button, self).__init__(parent, None)
|
||||
super(Button, self).__init__(parent, 0)
|
||||
|
||||
|
||||
LISTEN_COLOR_SENSOR_ON_C = b' \x0a\x00 \x41\x01 \x08\x01\x00\x00\x00\x01'
|
||||
|
21
tests.py
21
tests.py
@ -5,7 +5,10 @@ from threading import Thread
|
||||
|
||||
from pylgbst import MoveHub, COLOR_RED, LED, EncodedMotor, PORT_AB
|
||||
from pylgbst.comms import Connection, str2hex, hex2str
|
||||
from pylgbst.constants import PORT_LED, TILT_STATES
|
||||
from pylgbst.constants import PORT_LED, TILT_STATES, TILT_SENSOR_MODE_FULL, TILT_SENSOR_MODE_2AXIS, \
|
||||
MOVE_HUB_HARDWARE_HANDLE
|
||||
|
||||
HANDLE = MOVE_HUB_HARDWARE_HANDLE
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
@ -76,7 +79,7 @@ class GeneralTest(unittest.TestCase):
|
||||
|
||||
def test_tilt_sensor(self):
|
||||
hub = HubMock()
|
||||
hub.connection.notifications.append((14, '1b0e00 0f00 04 3a 0128000000000100000001'))
|
||||
hub.connection.notifications.append((HANDLE, '1b0e00 0f00 04 3a 0128000000000100000001'))
|
||||
time.sleep(1)
|
||||
|
||||
def callback(param1, param2=None, param3=None):
|
||||
@ -86,10 +89,20 @@ class GeneralTest(unittest.TestCase):
|
||||
log.debug("Tilt: %s %s %s", param1, param2, param3)
|
||||
|
||||
hub.tilt_sensor.subscribe(callback)
|
||||
hub.connection.notifications.append((14, "1b0e000500453a05"))
|
||||
hub.connection.notifications.append((14, "1b0e000600453a04fe"))
|
||||
hub.connection.notifications.append((HANDLE, "1b0e000500453a05"))
|
||||
hub.connection.notifications.append((HANDLE, "1b0e000a00473a010100000001"))
|
||||
time.sleep(1)
|
||||
hub.tilt_sensor.subscribe(callback, TILT_SENSOR_MODE_2AXIS)
|
||||
|
||||
hub.connection.notifications.append((HANDLE, "1b0e000500453a09"))
|
||||
time.sleep(1)
|
||||
|
||||
hub.tilt_sensor.subscribe(callback, TILT_SENSOR_MODE_FULL)
|
||||
hub.connection.notifications.append((HANDLE, "1b0e000600453a04fe"))
|
||||
time.sleep(1)
|
||||
|
||||
self._wait_notifications_handled(hub)
|
||||
hub.tilt_sensor.unsubscribe(callback)
|
||||
# self.assertEquals("0a01413a000100000001", hub.connection.writes[0][1])
|
||||
|
||||
def test_motor(self):
|
||||
|
Loading…
x
Reference in New Issue
Block a user