homeauto: service/piNode/devices.py comparison

comparison service/piNode/devices.py @ 1427:6bd36e5e109f

whitespace Ignore-this: 8abcbd0552c4356db0073b464fc3816a darcs-hash:2917f14b52fb149530a2369486ca9876cb0d334a

author	drewp <drewp@bigasterisk.com>
date	Thu, 08 Aug 2019 16:48:40 -0700
parents	83043957c809
children	69a84b3d1dfa

comparison

equal deleted inserted replaced

-:c3c2418d138c
+:6bd36e5e109f
 def hostStateInit(self):
 """
 If you don't want to use __init__, you can use this to set up
 whatever storage you might need for hostStatements
 """
 def description(self):
 return {
 'uri': self.uri,
 'className': self.__class__.__name__,
 'pinNumber': getattr(self, 'pinNumber', None),
 (e.g. light brightness) if its master version is in this
 object. This method is called on /graph requests so it should
 be fast.
 """
 return []
 def watchPrefixes(self):
 """
 subj,pred pairs of the statements that might be returned from
 readFromPoll, so the dashboard knows what it should
 watch. This should be eliminated, as the dashboard should just
 """
 return []
 def poll(self):
 return [] # statements
 def outputPatterns(self):
 """
 Triple patterns, using None as a wildcard, that should be routed
 to sendOutput
 """
 """
 structs to make output widgets on the dashboard. ~1 of these per
 handler you have in sendOutput
 """
 return []
 def sendOutput(self, statements):
 """
 If we got statements that match this class's outputPatterns, this
 will be called with the statements that matched.
 Todo: it would be fine to read back confirmations or
 whatever. Just need a way to collect them into graph statements.
 """
 raise NotImplementedError
 _knownTypes = set()
 def register(deviceType):
 _knownTypes.add(deviceType)
 return deviceType
 @register
 class MotionSensorInput(DeviceType):
 """
 Triggering all the time? Try 5V VCC, per https://electronics.stackexchange.com/a/416295
 0          30s          60s         90s                    10min
 |           |           |           |          ...          |
 Sensor input    ******** ** ******* ****
 :sees output    ........ .. ....... ....
 :seesRecently   .............................................................
 :seesRecently30 ....................................
 :motionStart    x        x  x       x
 self.lastMotionStart90 = 0
 def poll(self):
 motion = self.pi.read(self.pinNumber)
 now  = time.time()
 oneshot = []
 if self.lastRead is not None and motion != self.lastRead:
 oneshot = [(self.uri, ROOM['sees'], ROOM['motionStart'])]
 for v, t in [('lastMotionStart30', 30), ('lastMotionStart90', 90)]:
 if now - getattr(self, v) > t:
 oneshot.append((self.uri, ROOM['sees'], ROOM['motionStart%s' % t]))
 setattr(self, v, now)
 self.lastRead = motion
 return {'latest': [
 (self.uri, ROOM['sees'],
 ROOM['motion'] if motion else ROOM['noMotion']),
 ] + self.recentMotionStatements(now, motion),
 'oneshot': oneshot}
 dt = now - self.lastMotionTime
 return [(self.uri, ROOM['seesRecently'],
 ROOM['motion'] if (dt < 60 * 10) else ROOM['noMotion']),
 (self.uri, ROOM['seesRecently30'],
 ROOM['motion'] if (dt < 30) else ROOM['noMotion'])]
 def watchPrefixes(self):
 return [
 (self.uri, ROOM['sees']),
 (self.uri, ROOM['seesRecently']),
 (self.uri, ROOM['seesRecently30']),
 """3 PWMs for r/g/b on a strip"""
 # pigpio daemon is working fine, but
 # https://github.com/RPi-Distro/python-gpiozero/blob/59ba7154c5918745ac894ea03503667d6473c760/gpiozero/output_devices.py#L213
 # can also apparently do PWM
 deviceType = ROOM['RgbStrip']
 @classmethod
 def findInstances(cls, graph, board, pi):
 for row in graph.query("""SELECT DISTINCT ?dev ?r ?g ?b  WHERE {
 ?board
 :hasPin ?rpin;
 def __init__(self, graph, uri, pi, r, g, b):
 self.graph, self.uri, self.pi = graph, uri, pi
 self.rgb = map(int, [r, g, b])
 self.value = '#000000'
 def setup(self):
 for i in self.rgb:
 setupPwm(self.pi, i)
 def hostStatements(self):
 return [(self.uri, ROOM['color'], Literal(self.value))]
 def outputPatterns(self):
 return [(self.uri, ROOM['color'], None)]
 def _rgbFromHex(self, h):
 rrggbb = h.lstrip('#')
 return [int(x, 16) for x in [rrggbb[0:2], rrggbb[2:4], rrggbb[4:6]]]
 def sendOutput(self, statements):
 assert len(statements) == 1
 assert statements[0][:2] == (self.uri, ROOM['color'])
 rgb = self._rgbFromHex(statements[0][2])
 self.value = statements[0][2]
 for (i, v) in zip(self.rgb, rgb):
 self.pi.set_PWM_dutycycle(i, v)
 def outputWidgets(self):
 return [{
 'element': 'output-rgb',
 'subj': self.uri,
 'pred': ROOM['color'],
 DeviceType.__init__(self, *a, **kw)
 import DHT22
 self.sens = DHT22.sensor(self.pi, self.pinNumber)
 self.recentLowTemp = (0, None) # time, temp
 self.recentPeriodSec = 30
 def poll(self):
 stmts = set()
 now = time.time()
 if self.recentLowTemp[0] < now - self.recentPeriodSec:
 stmts.add((self.uri, RDFS['comment'],
 Literal('No recent DHT response (%.02f sec old)' % self.sens.staleness())))
 if self.recentLowTemp[1] is not None:
 stmts.add((self.uri, ROOM['recentLowTemperatureF'], Literal(self.recentLowTemp[1])))
 self.sens.trigger()
 return stmts
 def watchPrefixes(self):
 return [
 (self.uri, ROOM['temperatureF']),
 (self.uri, ROOM['humidity']),
 ]
 ROOM['press'] if closed else ROOM['release']),
 ]
 else:
 oneshot = []
 self.lastClosed = closed
 return {'latest': [
 (self.uri, ROOM['buttonState'],
 ROOM['pressed'] if closed else ROOM['notPressed']),
 ],
 'oneshot':oneshot}
 def watchPrefixes(self):
 return [
 (self.uri, ROOM['buttonState']),
 ]
 @register
 class OneWire(DeviceType):
 """
 Also see /my/proj/ansible/roles/raspi_io_node/tasks/main.yml for
 some system config that contains the pin number that you want to
 returnValue(stmts)
 except Exception as e:
 log.error(e)
 os.abort()
 def watchPrefixes(self):
 return [(s.uri, ROOM['temperatureF']) for s in self._sensors]
 class FilteredValue(object):
 def __init__(self, setter,
 slew=2.0, # step/sec max slew rate
 accel=5, # step/sec^2 acceleration
 ):
 self.setter = setter
 self.slew, self.accel = slew, accel
 self.x = None
 self.dx = 0
 self.goal = self.x
 self.lastStep = 0
 self.x = nextX
 reactor.callLater(.05, self.step)
 #print "x= %(x)s dx= %(dx)s goal= %(goal)s" % self.__dict__
 self.setter(self.x)
 @register
 class LedOutput(DeviceType):
 deviceType = ROOM['LedOutput']
 def hostStateInit(self):
 class Instant(object):
 def set(self, goal):
 _setPwm(goal)
 self.fv = Instant()
 self.gamma = float(self.graph.value(self.uri, ROOM['gamma'], default=1))
 def setup(self):
 setupPwm(self.pi, self.pinNumber)
 def outputPatterns(self):
 return [(self.uri, ROOM['brightness'], None)]
 def sendOutput(self, statements):
 assert len(statements) == 1
 assert statements[0][:2] == (self.uri, ROOM['brightness'])
 self.value = float(statements[0][2])
 self.fv.set(self.value)
 def _setPwm(self, x):
 v = max(0, min(255, int((x ** self.gamma)* 255)))
 self.pi.set_PWM_dutycycle(self.pinNumber, v)
 def hostStatements(self):
 return [(self.uri, ROOM['brightness'], Literal(self.value))]
 def outputWidgets(self):
 return [{
 'element': 'output-slider',
 'min': 0,
 'max': 1,
 'step': 1 / 255,
 'subj': self.uri,
 'pred': ROOM['brightness'],
 }]
 @register
 class OnboardTemperature(DeviceType):
 deviceType = ROOM['OnboardTemperature']
 pollPeriod = 10
 @classmethod
 def findInstances(cls, graph, board, pi):
 log.debug('graph has any connected devices of type OnboardTemperature on %r?', board)
 for row in graph.query('''SELECT DISTINCT ?uri WHERE {
 ?board :onboardDevice ?uri .
 ?uri a :OnboardTemperature .
 }''', initBindings=dict(board=board)):
 log.debug('  found %s', row.uri)
 yield cls(graph, row.uri, pi, pinNumber=None)
 def poll(self):
 milliC = open('/sys/class/thermal/thermal_zone0/temp').read().strip()
 c = float(milliC) / 1000.
 f = c * 1.8 + 32
 return [
 pixelStats = scales.collection('/rgbPixels',
 scales.PmfStat('updateOutput'),
 scales.PmfStat('currentColors'),
 scales.PmfStat('poll'),
 )
 @register
 class RgbPixels(DeviceType):
 """chain of ws2812 rgb pixels on pin GPIO18"""
 deviceType = ROOM['RgbPixels']
 def hostStateInit(self):
 self.anim = RgbPixelsAnimation(self.graph, self.uri, self.updateOutput)
 log.debug('%s maxIndex = %s', self.uri, self.anim.maxIndex())
 self.neo = rpi_ws281x.Adafruit_NeoPixel(self.anim.maxIndex() + 1, pin=18)
 self.neo.begin()
 colorOrder, stripType = self.anim.getColorOrder(self.graph, self.uri)
 def sendOutput(self, statements):
 self.anim.onStatements(statements)
 @pixelStats.updateOutput.time()
 def updateOutput(self):
 print list(self.anim.currentColors())
 return
 with pixelStats.currentColors.time():
 colors = self.anim.currentColors()
 for idx, (r, g, b) in colors:
 if idx < 4:
 log.debug('out color %s (%s,%s,%s)', idx, r, g, b)
 self.neo.setPixelColorRGB(idx, r, g, b)
 self.neo.show()
 self.anim.step()
 return []
 def hostStatements(self):
 return self.anim.hostStatements()
 def outputPatterns(self):
 return self.anim.outputPatterns()
 def outputWidgets(self):
 return self.anim.outputWidgets()
 @register
 class Lcd8544(DeviceType):
 """PCD8544 lcd (nokia 5110)"""
 deviceType = ROOM['RgbStrip']
 @classmethod
 def findInstances(cls, graph, board, pi):
 for row in graph.query("""
 SELECT DISTINCT ?dev ?din ?clk ?dc ?rst  WHERE {
 ?dev a :Lcd8544 .
 spi=Adafruit_GPIO.SPI.BitBang(
 Adafruit_Nokia_LCD.GPIO.RPiGPIOAdapter(RPi.GPIO),
 sclk=clk,
 mosi=din))
 self.lcd.begin(contrast=60)
 def hostStatements(self):
 return []
 return [(self.uri, ROOM['color'], Literal(self.value))]
 def outputPatterns(self):
 return []
 return [(self.uri, ROOM['color'], None)]
 def sendOutput(self, statements):
 rgb = self._rgbFromHex(statements[0][2])
 self.value = statements[0][2]
 for (i, v) in zip(self.rgb, rgb):
 self.pi.set_PWM_dutycycle(i, v)
 def outputWidgets(self):
 return []
 return [{
 'element': 'output-rgb',
 'subj': self.uri,
 for out in graph.query("""SELECT DISTINCT ?area ?chan WHERE {
 ?dev :output [:area ?area; :channel ?chan] .
 }""", initBindings=dict(dev=row.dev), initNs={'': ROOM}):
 outs[out.area] = out.chan.toPython()
 yield cls(graph, row.dev, pi, outs)
 def __init__(self, graph, dev, pi, outs):
 super(PwmBoard, self).__init__(graph, dev, pi, pinNumber=None)
 import PCA9685
 self.pwm = PCA9685.PWM(pi, bus=1, address=0x40)
 self.pwm.set_frequency(1200)
 assert statements[0][1] == ROOM['brightness'];
 chan = self.outs[statements[0][0]]
 value = float(statements[0][2])
 self.values[statements[0][0]] = value
 self.pwm.set_duty_cycle(chan, value * 100)
 def outputWidgets(self):
 return [{
 'element': 'output-slider',
 'min': 0,
 'max': 1,
 'step': 1 / 255,
 'subj': area,
 'pred': ROOM['brightness'],
 } for area in self.outs]
 def makeDevices(graph, board, pi):
 out = []
 for dt in sorted(_knownTypes, key=lambda cls: cls.__name__):
 out.extend(dt.findInstances(graph, board, pi))
 return out

Mercurial > code > home > repos > homeauto

comparison service/piNode/devices.py @ 1427:6bd36e5e109f