Example of objects I used in a pyFirmata project, for controlling a something (a wall in this case) with an actuator.
The arduinos argument of Wall.__init__ was a helper class containing multiple arduino's, but it works the same like the Firmata.Board class. The pin_defs is a dictionary in the form of:
{'position' : 'a:0:i',
'speed' : 'd:3:p',
'direction': 'd:4:o' }
Defining what pin is where on the arduino(s).
| from collections import deque | |
| class LinearSensor(object): | |
| position = None | |
| calibrated = False | |
| upper_boundary = None | |
| lower_boundary = None | |
| position_scale = None | |
| _no_read_warning = False | |
| _readings = deque([0.5], maxlen=10) | |
| def __init__(self, arduinos, pin_defs, parent, log, scale=10): | |
| self.log = log | |
| self.pin = arduinos.get_pin(pin_defs['position']) | |
| self.parent = parent | |
| self.id = "linear_sensor_%s" % self.parent.get_id() | |
| self.SCALE = scale | |
| self.AVG_COUNT = 5 | |
| def get_position(self): | |
| self.update() | |
| return self.position | |
| def update(self): | |
| reading = self.get_reading() | |
| if self.calibrated: | |
| if reading: | |
| old = self.position or 0 | |
| place = reading - self.lower_boundary | |
| self.position = int(round(place / float(self.position_scale) * self.SCALE)) | |
| if old is not self.position: | |
| self.log.debug("Updating position LinearSensor %s. old: %d, reading: %.4f, new: %d" % (self.id, old, reading, self.position)) | |
| def get_reading(self): | |
| # average the readings a bit | |
| # read = self.pin.read() | |
| # if not read: | |
| # if not self._no_read_warning: | |
| # self._no_read_warning = True | |
| # self.log.warning("Could not get any sensor data from %s.pin (%s)" % (self.id, self.pin)) | |
| # self.position = None # so it is clearly out of bounds | |
| # return None | |
| # else: | |
| # self._no_read_warning = False | |
| # self._readings.append(read) | |
| # reading = sum(self._readings) / len(self._readings) | |
| reading = self.pin.read() | |
| return reading | |
| def get_outside_boundaries(self): | |
| if not self.calibrated: | |
| return 0 | |
| reading = self.get_reading() | |
| if reading: | |
| if self.upper_boundary > self.lower_boundary: | |
| if reading > self.upper_boundary: | |
| # too far out | |
| return -1 | |
| elif reading < self.lower_boundary: | |
| # too far in | |
| return 1 | |
| else: | |
| if reading > self.lower_boundary: | |
| # too far in | |
| return 1 | |
| elif reading < self.upper_boundary: | |
| # to far out | |
| return -1 | |
| return 0 | |
| def set_lower(self): | |
| # average a bit, skip low en highest reads | |
| # for i in range(10): | |
| # self.get_reading() | |
| # readings = list(self._readings) | |
| # readings.sort() | |
| # assert sum(readings) > 0, "Could not set lower boundary, readings are 0" | |
| # self.lower_boundary = sum(readings[1:-1]) / (len(self._readings) - 2) | |
| self.lower_boundary = self.pin.read() | |
| def set_upper(self): | |
| # average a bit, skip low en highest reads | |
| # for i in range(10): | |
| # self.get_reading() | |
| # readings = list(self._readings) | |
| # readings.sort() | |
| # assert sum(readings) > 0, "Could not set upper boundary, readings are 0" | |
| # self.upper_boundary = sum(readings[1:-1]) / (len(self._readings) - 2) | |
| self.upper_boundary = self.pin.read() | |
| def can_calibrate(self): | |
| return bool(self.upper_boundary and self.lower_boundary) | |
| def calibrate(self): | |
| assert self.can_calibrate(), "I don't know both boundaries yet!" | |
| self.position_scale = self.upper_boundary - self.lower_boundary | |
| self.update() | |
| self.log.info('Calibrated LinearSensor %s' % self.id) | |
| calibrated = property(lambda self: bool(self.position_scale)) |
| import time | |
| import logging | |
| from .util import LinearSensor | |
| log = logging.getLogger('walls') | |
| INWARD = -1 | |
| OUTWARD = 1 | |
| class Wall(object): | |
| """ Object for controlling a wall""" | |
| position = 0 # 0 = most inward, 10 is most outward | |
| SPEED = 0.5 # 0 = off, 1 = full speed | |
| swap_direction = False | |
| mover = None | |
| target = None | |
| _end = 0 | |
| _direction = 1 | |
| def __init__(self, arduinos, pin_defs, mover_cls, name='N/A'): | |
| self.id = "wall_%s" % name | |
| if type(mover_cls) is str: | |
| mover_cls = globals()[mover_cls] | |
| # abstract for different movers | |
| for name in mover_cls.needed_pins: | |
| setattr(self, "%s_pin" % name, arduinos.get_pin(pin_defs[name])) | |
| self.mover = mover_cls(self) | |
| self.pos_sensor = LinearSensor(arduinos, pin_defs, self, log, dont_load_params) | |
| def get_id(self): | |
| return self.id | |
| def move_for(self, seconds, direction): | |
| self._direction = self.get_direction(direction) | |
| self._end = time.time() + seconds | |
| def move_to(self, target): | |
| """ | |
| Sets a new position for the wall. Moving is delegated to ``update`` | |
| """ | |
| assert self.pos_sensor.calibrated, "You have to calibrate me first, otherwise I have no clue what a position is..." | |
| self.target = target | |
| def stop(self): | |
| """ Stop immediatly. """ | |
| self.target = self.position | |
| self._end = 0 | |
| self.mover.stop() | |
| def update(self, no_move=False): | |
| """ | |
| Updates the position of the wall element. | |
| Give ``no_move`` as True if you just want to update the position | |
| """ | |
| self.position = self.pos_sensor.get_position() | |
| if no_move: | |
| # We just needed an updated position. Set target to position | |
| self.target = self.position | |
| return | |
| # Move for a certain amount of time | |
| if self._end: | |
| if time.time() < self._end: | |
| self.mover.move(self._direction, self.SPEED) | |
| else: | |
| self.mover.stop() | |
| self._end = 0 | |
| # otherwise we think we should move back afterwards | |
| # self.target = self.position | |
| else: | |
| # If we aren't calibrated, doing this doesn't make sense | |
| if self.pos_sensor.calibrated: | |
| # If we are outta bounds, get back in there! | |
| out = self.pos_sensor.get_outside_boundaries() | |
| if out != 0: | |
| self.move_for(1, out) | |
| return | |
| # Ok, save to move to target | |
| if self.position is not self.target: | |
| self.mover.move(self.get_direction(), self.SPEED) | |
| else: | |
| self.mover.stop() | |
| def get_direction(self, direction=None): | |
| """ | |
| Returns 1 or -1 for the direction. Returns 0 when we are at the target | |
| """ | |
| if self.target is None or self.position is None: | |
| return 0 | |
| # get direction by position | |
| if not direction: | |
| if self.target == self.position: | |
| return 0 | |
| direction = int((self.target - self.position) / | |
| abs(self.target - self.position)) | |
| if self.swap_direction: | |
| return direction * -1 | |
| return direction | |
| def is_moving(self): | |
| if self._end and time.time() < self._end: | |
| return True | |
| if self.position is not self.target: | |
| return True | |
| return False | |
| # Calibration methods | |
| def start_calibration(self): | |
| """ | |
| Starts the calibration process. Starts moving in *a* direction. Call | |
| ``is_most_inward`` when the wall is at it's most inward position, or | |
| ``is_most_outward`` when the wall is at it's most outward position. | |
| The wall will then reverse, and after the last call stop. Afterwards, | |
| call ``finish_calibration``. | |
| """ | |
| self.mover.move(INWARD, self.SPEED) | |
| print "Calibration started. Call is_most_inward or is_most_outward when in that position." | |
| def is_most_inward(self): | |
| """ | |
| Method to call when the wall is bent in it's most inward position. | |
| Call ``calibrate`` after you called ``is_most_outward`` as well. | |
| """ | |
| self.pos_sensor.set_lower() | |
| if self.pos_sensor.can_calibrate(): | |
| self.mover.stop() | |
| print "Now call finish_calibration" | |
| else: | |
| self.mover.move(OUTWARD, self.SPEED) | |
| print "Boundary set, moving in the opposite direction" | |
| def is_most_outward(self): | |
| """ | |
| Method to call when the wall is bent in it's most outward position. | |
| Call ``calibrate`` after you called ``is_most_inward`` as well. | |
| """ | |
| self.pos_sensor.set_upper() | |
| if self.pos_sensor.can_calibrate(): | |
| self.mover.stop() | |
| print "Now call finish_calibration" | |
| else: | |
| self.mover.move(OUTWARD, self.SPEED) | |
| print "Boundary set, moving in the opposite direction" | |
| def finish_calibration(self, swap_direction=False): | |
| """ | |
| Calibrates the position and the direction. Call *after* you have | |
| called ``is_most_inward`` and ``is_most_outward``. Give | |
| ``swap_direction`` as True is the wall is moving opposite of how it | |
| should. | |
| """ | |
| assert self.pos_sensor.can_calibrate(), \ | |
| "Call ``is_most_inward`` and ``is_most_outward`` first!" | |
| self.swap_direction = swap_direction | |
| param, created = Parameter.objects.get_or_create(id=self.id) | |
| param.value = swap_direction | |
| param.save() | |
| self.pos_sensor.calibrate() | |
| class BaseMover(object): | |
| needed_pins = () | |
| def __init__(self, wall): | |
| self.wall = wall | |
| def move(self, direction, speed): | |
| """ | |
| Method to move into a certain direction with a certain speed. This | |
| should be overridden in the inheriting class. | |
| :direction arg: -1, 1 or 0. With 0 the object should not move | |
| :speed arg: 0.0 - 1.0 | |
| """ | |
| raise NotImplementedError | |
| def stop(self): | |
| """ | |
| Method to stop the actuator. Override this in the inheriting class. | |
| """ | |
| raise NotImplementedError | |
| class HBridgeMover(BaseMover): | |
| """ | |
| Mover object for the green H-Bridges that use a direction pin and a pwm | |
| pin for speed | |
| """ | |
| needed_pins = ('speed', 'direction') | |
| def move(self, direction, speed): | |
| if direction == 0: | |
| self.stop() | |
| return | |
| # pin should be low when direction is not 1, so make it zero | |
| direction = direction > 0 and 1 or 0 | |
| self.wall.direction_pin.write(direction) | |
| self.wall.speed_pin.write(self.wall.SPEED) | |
| def stop(self): | |
| self._stop = True | |
| self.wall.speed_pin.write(0) | |
| class PulseMover(BaseMover): | |
| """ | |
| Mover object for the servo-like actuator that requires a pulse between 0.8 | |
| and 2.2 ms | |
| """ | |
| needed_pins = ('pulse', 'kill') | |
| min = 800 | |
| max = 2200 | |
| scale = (max - min) / 2 | |
| mid = min + scale | |
| def move(self, direction, speed): | |
| if direction == 0: | |
| self.stop() | |
| return | |
| speed = speed * self.scale | |
| self.wall.pulse_pin.send_pulse(self.mid + speed * direction) | |
| def stop(self): | |
| self._stop = True | |
| self.wall.pulse_pin.send_pulse(self.mid) |