Mercurial > code > home > repos > homeauto

comparison service/sba/sba.py @ 808:867f59c83dba

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
moved from proj/pixel Ignore-this: 97cafcfb96fd33ea70fcd467659ee31a darcs-hash:20110808045021-312f9-be4547f197b49b893d6e873f9c96f30491d0bdc7.gz
author drewp <drewp@bigasterisk.com>
date Sun, 07 Aug 2011 21:50:21 -0700
parents
children be855a111619
comparison
equal deleted inserted replaced
807:4713bb87e34e 808:867f59c83dba
1 from __future__ import division
2 import serial, time, jsonlib, sys, cgi, argparse
3 import cyclone.web
4 from twisted.python import log
5 from twisted.internet import reactor
6
7 class Sba(object):
8 def __init__(self, port="/dev/ttyACM0"):
9 self.port = port
10 self.reset()
11
12 def reset(self):
13 log.msg("reopening port")
14 self.s = serial.Serial(self.port, baudrate=115200)
15 log.msg(str(self.s.__dict__))
16 self.sendControl()
17
18 def sendControl(self):
19 controlBits = [0, 1,
20 0, 0, 0,
21 1, 1, 1, 1, 1, 1, 1, # b correction
22 0, 0, 0,
23 1, 1, 1, 1, 1, 1, 1, # g correction
24 0,
25 0, 0, # clock mode 00=internal
26 1, 1, 1, 1, 1, 1, 1, # r correction
27 ]
28
29 control = reduce(lambda a, b: a<<1 | b,
30 #controlBits
31 reversed(controlBits)
32 )
33 self.send("C" + hex(control)[2:].zfill(8))
34 self.send("E0")
35
36 def send(self, cmd, getResponse=True):
37 """
38 send a command using the protocol from http://engr.biz/prod/SB-A/
39
40 we will attach the carriage return, cmd is just a string like 'V'
41
42 Returns the response line, like '+OK'
43 """
44 try:
45 self.s.write(cmd + "\r")
46 except OSError:
47 self.reset()
48
49 if getResponse:
50 return self.s.readline().strip()
51
52 def rgbs(self, rgbs):
53 """
54 send a set of full rgb packets. Values are 0..1023.
55 """
56 t1 = time.time()
57 for (r,g,b) in rgbs:
58 packed = (b & 0x3ff) << 20 | (r & 0x3ff) << 10 | (g & 0x3ff)
59 self.send("D%08x" % packed, getResponse=False)
60
61 self.send("L1", getResponse=False)
62 self.send("L0", getResponse=False)
63 sends = time.time() - t1
64 # doing all the reads together triples the transmission rate
65 t2 = time.time()
66 [self.s.readline() for loop in range(2 + len(rgbs))]
67 reads = time.time() - t2
68
69 log.msg("%.1f ms for sends, %.1f ms for reads" % (
70 1000 * sends, 1000 * reads))
71
72 class BriteChain(object):
73 def __init__(self, sba):
74 self.sba = sba
75 self.colors = []
76
77 def setColor(self, pos, color):
78 """color is (r,g,b) 10-bit int. The highest position you ever
79 set is how many channels we'll output"""
80 if len(self.colors) <= pos:
81 self.colors.extend([(0,0,0)]*(pos - len(self.colors) + 1))
82 self.colors[pos] = color
83 self.refresh()
84
85 def getColor(self, pos):
86 try:
87 return self.colors[pos]
88 except IndexError:
89 return (0,0,0)
90
91 def refresh(self):
92 self.sba.rgbs(self.colors[::-1])
93
94 class IndexHandler(cyclone.web.RequestHandler):
95 def get(self):
96 self.set_header("Content-type", "text/html")
97 self.write(open("sba.html").read())
98
99 class BriteHandler(cyclone.web.RequestHandler):
100 """
101 /brite/0 is the first shiftbrite on the chain. Put a text/plain
102 color like #ffffff (8-bit) or a application/json value like
103 {"rgb10":[1023,1023,1023]} (for 10-bit). GET (with accept, to pick
104 your format) to learn the current color.
105
106 /brite/1 affects the second shiftbrite on the chain, etc
107 """
108 def put(self, pos):
109 d = self.request.body
110 ctype = self.request.headers.get("Content-Type")
111 if ';' in ctype:
112 ctype = ctype.split(';')[0].strip()
113 if ctype == 'text/plain':
114 color = decode8bitHexColor(d)
115 elif ctype == 'application/json':
116 color = jsonlib.read(d)['rgb10']
117 elif ctype == 'application/x-www-form-urlencoded':
118 color = decode8bitHexColor(cgi.parse_qs(d)['color'][0])
119 else:
120 self.response.set_status(415, "use text/plain, application/json, "
121 "or application/x-www-form-urlencoded")
122 return
123
124 self.settings.chain.setColor(int(pos), color)
125 self.set_header("Content-Type", "text/plain")
126 self.write("set %s\n" % pos)
127
128 def post(self, pos):
129 self.put(pos)
130 self.redirect("..")
131
132 def get(self, pos):
133 # todo: content neg
134 color = self.settings.chain.getColor(int(pos))
135 self.set_header("Content-Type", "text/plain")
136 self.write(encode8bitHexColor(color))
137
138 def decode8bitHexColor(s):
139 return [4 * int(s.lstrip('#')[i:i+2], 16) for i in [0, 2, 4]]
140 def encode8bitHexColor(color):
141 return "#%02X%02X%02X" % (color[0] // 4, color[1] // 4, color[2] // 4)
142
143 class Application(cyclone.web.Application):
144 def __init__(self, chain):
145 handlers = [
146 (r"/", IndexHandler),
147 (r"/brite/(\d+)", BriteHandler),
148 ]
149
150 settings = {
151 "static_path": "./static",
152 "template_path": "./template",
153 "chain" : chain,
154 }
155
156 cyclone.web.Application.__init__(self, handlers, **settings)
157
158 def main():
159 parser = argparse.ArgumentParser(description='drive sba lights')
160 parser.add_argument('-v', '--verbose', action="store_true", help='logging')
161 args = parser.parse_args()
162
163 try:
164 sba = Sba()
165 except serial.SerialException:
166 sba = Sba("/dev/ttyACM1")
167
168 chain = BriteChain(sba)
169
170 if 0: # todo: stick test patterns like this on some other resource
171 while 1:
172 t1 = time.time()
173 steps = 0
174 for x in range(0, 1024, 5):
175 steps += 1
176 sba.rgbs([(x, x, x)] * 2)
177 print steps / (time.time() - t1)
178
179 if args.verbose:
180 log.startLogging(sys.stdout)
181 reactor.listenTCP(9060, Application(chain))
182 reactor.run()
183
184 if __name__ == "__main__":
185 main()