AEOutput.ThreadProxy

1 ''' 2 Defines a threaded proxy for L{AEOutput.Audio} devices. 3 4 @author: Brett Clippingdale 5 @author: Peter Parente 6 @organization: IBM Corporation 7 @copyright: Copyright (c) 2005, 2007 IBM Corporation 8 @license: The BSD License 9 10 All rights reserved. This program and the accompanying materials are made 11 available under the terms of the BSD license which accompanies 12 this distribution, and is available at 13 U{http://www.opensource.org/licenses/bsd-license.php} 14 ''' 15 16 import threading, Queue 17 import Base 18 import AEConstants 19

20 -class AudioThreadProxy(threading.Thread, Base.AEOutput):

21 ''' 22 Buffers calls to methods on a L{AEOutput.Audio} device in a secondary thread 23 and executes them some time later when that thread runs. 24 25 @ivar device: Device on which to invoke send* methods 26 @type device: L{AEOutput.Audio} 27 @ivar want_stop: Flag indicating a stop is requested 28 @type want_stop: boolean 29 @ivar just_stopped: Flag indicating that the last written command was a stop. 30 Used to avoid unnecessary stops 31 @type just_stopped: boolean 32 @ivar lock: Semaphore used to ensure no commands are buffered while the 33 buffer is being reset after sending a stop 34 @type lock: threading.Sempahore 35 @ivar init_event: Event used to block non-threaded calls to the device until 36 the device has been initialized in the context of the running thread. 37 @type init_event: threading.Event 38 @ivar data_buffer: Buffer of commands to be sent to the output device 39 @type data_buffer: Queue.Queue 40 @ivar alive: Is the thread running or not? 41 @type alive: boolean 42 '''

43 - def __init__(self, device):

44 ''' 45 Initializes the parent class and stores the device reference. Creates a 46 queue that will buffer commands to the speech device. Creates flags used 47 for indicating whether a stop is requested or has been requested recently. 48 Creates a semaphore used to ensure that no commands can be added to the 49 buffer while it is being reset by a stop command. 50 51 @param device: The device reference to use for writing commands 52 @type device: L{AEOutput.Audio} 53 ''' 54 threading.Thread.__init__(self) 55 Base.AEOutput.__init__(self) 56 self.device = device 57 self.want_stop = False 58 self.just_stopped = False 59 self.lock = threading.Semaphore() 60 self.init_event = threading.Event() 61 self.data_buffer = Queue.Queue() 62 self.alive = False

63

64 - def init(self):

65 ''' 66 Called after the instance is created to start the device running. The 67 device's init method is called in the context of the running thread's 68 L{run} method before the thread enters its loop. 69 ''' 70 self.alive = True 71 self.start()

72

73 - def getCapabilities(self):

74 ''' 75 Gets the capabilities of the proxied device. This method is called in the 76 context of the caller, not the running thread. 77 78 @return: List of capability names 79 @rtype: list of string 80 ''' 81 return self.device.getCapabilities()

82

83 - def loadStyles(self, sett_man):

84 ''' 85 Called after the L{init} method by the L{DeviceManager} to ensure that the 86 device is functioning before time is spent unserializing its style data. 87 88 Calls the init method on the default style object and provides it with a 89 reference to this initialized device. Then tries to load the persisted 90 setting values from disk. If that fails, the L{DeviceManager} will try to 91 call L{createDistinctStyles} instead. 92 93 This method is called in the context of the caller, not the running thread. 94 It blocks until the L{init_event} is set indicating the device has been 95 initialized in the second thread. 96 97 @param sett_man: Instance of the settings manager 98 @type sett_man: L{SettingsManager} 99 @raise KeyError: When styles have not previously been persisted for this 100 device 101 @raise OSError: When the profile file cannot be opened or read 102 ''' 103 self.init_event.wait() 104 self.device.loadStyles(sett_man)

105

106 - def saveStyles(self, sett_man):

107 ''' 108 Persists styles to disk. Called after the L{close} method by the 109 L{DeviceManager} to ensure the device is properly shutdown before 110 serializing its data. This method is called in the context of the caller, 111 not the running thread. 112 113 @param sett_man: Instance of the settings manager 114 @type sett_man: L{SettingsManager} 115 @raise KeyError: When styles have not previously been persisted for this 116 device 117 @raise OSError: When the profile file cannot be opened or read 118 ''' 119 self.device.saveStyles(sett_man)

120

121 - def setStyle(self, key, style):

122 ''' 123 Stores the style object under the given key. The style object should be one 124 previously generated by this device (e.g. using L{createDistinctStyles}) 125 but it is not an enforced requirement. Always makes the style clean before 126 storing it. This method is called in the context of the caller, not the 127 running thread. 128 129 @param key: Any immutable object 130 @type key: immutable 131 @param style: L{AEOutput} subclass of L{AEState} 132 @type style: L{AEState} 133 ''' 134 self.device.setStyle(key, style)

135

136 - def getStyle(self, key):

137 ''' 138 Gets the style object stored under the given key. If the key is unknown, 139 returns an empty flyweight backed by the default style and stores the new 140 style in L{styles}. This method is called in the context of the caller, not 141 the running thread. 142 143 @param key: Any immutable object 144 @type key: immutable 145 @return: L{AEOutput} subclass of L{AEState} 146 @rtype: L{AEState} 147 ''' 148 return self.device.getStyle(key)

149

150 - def createDistinctStyles(self, num_groups, num_layers):

151 ''' 152 Creates up to the given number of styles for this device. This method is 153 called in the context of the caller, not the running thread. It blocks 154 until the L{init_event} is set indicating the device has been initialized 155 in the second thread. 156 157 @param num_groups: Number of sematic groups the requestor would like to 158 represent using distinct styles 159 @type num_groups: integer 160 @param num_layers: Number of content origins (e.g. output originating from 161 a background task versus the focus) the requestor would like to represent 162 using distinct styles 163 @type num_layers: integer 164 @return: New styles 165 @rtype: list of L{AEOutput.Style} 166 ''' 167 self.init_event.wait() 168 return self.device.createDistinctStyles(num_groups, num_layers)

169

170 - def getDefaultStyle(self):

171 ''' 172 Creates up to the given number of styles for this device. This method is 173 called in the context of the caller, not the running thread. It blocks 174 until the L{init_event} is set indicating the device has been initialized 175 in the second thread. 176 177 @return: Default style 178 @rtype: L{AEOutput.Style} 179 ''' 180 self.init_event.wait() 181 return self.device.getDefaultStyle()

182

183 - def close(self):

184 ''' 185 Stops the running thread. Puts a null callable in the queue to wake the 186 thread if it is sleeping. 187 ''' 188 self.alive = False 189 self._put(lambda : None)

190

191 - def getProxy(self):

192 ''' 193 Returns this object as the proxy for itself because a thread proxying for 194 another thread proxying for a device is not supported. 195 196 @return: self 197 @rtype: L{AudioThreadProxy} 198 ''' 199 return self

200

201 - def getName(self):

202 ''' 203 Gives the user displayable (localized) name for this output device. 204 Relevant version and device status should be included. This method is 205 called in the context of the caller, not the running thread. 206 207 @return: The localized name for the device 208 @rtype: string 209 ''' 210 return self.device.getName()

211

212 - def send(self, name, value, style=None):

213 ''' 214 Buffers methods to call for known commands in the context of the thread. 215 216 @param name: Descriptor of the data value sent 217 @type name: object 218 @param value: Content value 219 @type value: object 220 @param style: Style with which this value should be output 221 @type style: L{AEOutput.Style} 222 @return: Return value specific to the given command 223 @rtype: object 224 @raise NotImplementedError: When not overridden in a subclass 225 ''' 226 if name == AEConstants.CMD_STOP: 227 self._put(self.device.sendStop, style) 228 elif name == AEConstants.CMD_TALK: 229 self._put(self.device.sendTalk, style) 230 elif name == AEConstants.CMD_STRING: 231 self._put(self.device.sendString, value, style) 232 elif name == AEConstants.CMD_STRING_SYNC: 233 self.device.sendStringSync(value, style) 234 elif name == AEConstants.CMD_FILENAME: 235 try: 236 # may not be implemented 237 self._put(self.device.sendFilename, value, style) 238 except NotImplementedError: 239 pass 240 elif name == AEConstants.CMD_INDEX: 241 try: 242 # may not be implemented 243 return self._put(self.device.sendIndex, style) 244 except NotImplementedError: 245 pass

246

247 - def isActive(self):

248 ''' 249 Indicates whether the device is active (giving output) or not. 250 251 @return: True when content is buffered or the device is outputing 252 @rtype: boolean 253 @raise NotImplementedError: When not overriden in a subclass 254 ''' 255 return (self.data_buffer.qsize() != 0 or self.device.isActive())

256

257 - def parseString(self, text, style, por, sem):

258 ''' 259 Parses the string using the implementation provided by the proxied device. 260 This method is called in the context of the caller, not the running thread. 261 262 @param text: Text to be parsed 263 @type text: string 264 @param style: Style object defining how the text should be parsed 265 @type style: L{AEOutput.Style} 266 @param por: Point of regard for the first character in the text, or None if 267 the text is not associated with a POR 268 @type por: L{POR} 269 @param sem: Semantic tag for the text 270 @type sem: integer 271 @return: Parsed words 272 @rtype: 3-tuple of lists of string, L{POR}, L{AEOutput.Style} 273 ''' 274 return self.device.parseString(text, style, por, sem)

275

276 - def _put(self, mtd, *args):

277 ''' 278 Buffers any non-stop command in the queue and returns immediately. Sets the 279 L{want_stop} flag for a stop command if neither L{want_stop} nor 280 L{just_stopped} is set. Blocks on entry if the thread is busy clearing out 281 the buffer in response to a previous stop command. Leaves the lock set if 282 a new stop command is buffered. It will be unset when the command is 283 processed. 284 285 @param mtd: Device method to call at a later time 286 @type mtd: callable 287 @param args: Additional positional arguments to be passed to the device 288 when the method is invoked 289 @type args: list 290 ''' 291 self.lock.acquire() 292 if (mtd is self.device.sendStop): 293 if not self.want_stop and not self.just_stopped: 294 # only queue the stop and set the flag if it's not already set 295 self.want_stop = True 296 # add the stop command to the buffer in case the thread is sleeping 297 self.data_buffer.put_nowait((mtd, args)) 298 # IMPORTANT: do not release the semaphore, the thread will do it when 299 # it processes the stop command; it must be held so that further 300 # invocations of this method block until the buffer has been emptied 301 return 302 else: 303 self.data_buffer.put_nowait((mtd, args)) 304 # as soon as something is buffered, stop is no longer last. 305 self.just_stopped = False 306 self.lock.release()

307

308 - def run(self):

309 ''' 310 Runs the thread until alive is not longer True. Sleeps until methods and 311 arguments to be applied are put in the L{data_buffer}. Wakes up and 312 invokes all buffered methods and arguments. Initializes the device before 313 entering the loop and closes it after leaving the loop. 314 ''' 315 self.device.init() 316 self.init_event.set() 317 while self.alive: 318 # Queue.get() blocks for data 319 mtd, args = self.data_buffer.get() 320 321 # if want to stop, it doesn't matter what the buffer element was 322 if self.want_stop: 323 # reset the buffer immediately 324 self.data_buffer = Queue.Queue() 325 # send the device specific stop command 326 self.device.sendStop() 327 # reset the stop flag 328 self.want_stop = False 329 # indicate we've just stopped 330 self.just_stopped = True 331 # IMPORTANT: release the semaphore so the put method can continue 332 self.lock.release() 333 else: 334 # handle the command 335 try: 336 mtd(*args) 337 except NotImplementedError: 338 pass 339 self.device.close()

340

Source Code for Module AEOutput.ThreadProxy