image_converter/image_flipper.py

   1 #!/usr/bin/python2.4
   2 #
   3 # Copyright 2010 Google Inc. All Rights Reserved.
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17 """Demo for Appengine and pull taskqueue integration with TaskPuller.
  18
  19 This example shows how Google AppEngine apps can use PULL queues and add
  20 tasks to get part of work/processing done outside appengine environment.
  21 This lets the app-engine user use any executable binary (like OCR reader)
  22 and thus complements the Google AppEngine with processing on an external
  23 machine. This demonstrates end-to-end integration of a Google AppEngine app
  24 with gtaskqueue_puller(running on VMs) by putting a task in the taskqueue to
  25 flip the image by the app, this task gets processed(fliping of image) by the
  26 taskqueue puller workers and output from the app is posted back to the same app.
  27 This app handles the posted data and stores the flipped image in datastore.
  28 The images are searchable by name and user can see both uploaded as well as
  29 processed/flipped image. In addition the handler that the workers use to post
  30 data back to AppEngine are OAuth1 protected, and limited to admins of the
  31 AppEngine application.
  32
  33 Example puller command line to acces pull Queue tasks using the REST API on
  34 a worker outside AppEngine:
  35
  36   gtaskqueue_puller --project_name=imageconvertdemo \
  37   --taskqueue_name=imageconvert --lease_secs=30
  38   --executable_binary="convert -annotate 30x20+10+10 google.rocks"
  39   --output_url="http://imageconvertdemo.appspot.com/taskdata?name="
  40   --appengine_access_token_file=./access_token
  41
  42 """
  43
  44 __author__ = 'vibhooti@google.com (Vibhooti Verma)'
  45
  46 import logging
  47 from google.appengine.api import oauth
  48 from google.appengine.ext import db
  49 from google.appengine.ext.webapp.util import run_wsgi_app
  50 from google.appengine.ext import webapp
  51 from google.appengine.api import taskqueue
  52
  53
  54 class SmallImage(db.Model):
  55   """Stores image in datastore."""
  56   image_name = db.StringProperty(required = True)
  57   image = db.BlobProperty(required = True)
  58
  59
  60 class TaskImageMap(db.Model):
  61   """Stores task_id to image_name mapping."""
  62   image_ref = db.ReferenceProperty(SmallImage)
  63   task_name = db.StringProperty(required = True)
  64
  65
  66 class ImageRequestHandler(webapp.RequestHandler):
  67   """Handles requests to upload an image."""
  68
  69   def get(self):
  70     """Get to let the user upload any image."""
  71     self.response.out.write('''
  72     <html>
  73       <body>
  74         <form action="/" method="post" enctype="multipart/form-data">
  75           <p>Name: <input type="text" name="name" /></p>
  76           <div><input type="file" name="upload1"></div>
  77           <div><input type="submit" value="Upload"></div>
  78         </form>
  79       </body>
  80     </html>
  81     ''')
  82
  83   def WriteEntry(self, image_name, value):
  84     """Stores the image in data store as a blob object."""
  85     small_image = SmallImage(key_name=image_name,
  86                              image_name=image_name,
  87                              image=db.Blob(value))
  88     small_image.put()
  89     self.EnqueueTask(small_image, value)
  90
  91   def EnqueueTask(self, image_ref, value):
  92     """Enqueues the task in pull queue to process the image.
  93
  94     Args:
  95      key: Name of the image
  96      value: Image to be sent as payload
  97
  98     This function sets the uploaded image as payload of the task and then puts
  99     the task in appropriate queue (defined in queue.yaml). Project name gets
 100     set automatically as project in which your app is running. task_id gets
 101     automatically generated if you do not sepcify explicitely. After we put the
 102     task in the queue, we also store the task_id to image name mapping which
 103     can be used later when output of the task is posted back.
 104     """
 105     # queuename must be specified in queue.yaml
 106     q = taskqueue.Queue('imageconvert')
 107     task = taskqueue.Task(payload=value, method='PULL')
 108     q.add(task)
 109     # create a taskid to image_name entry and put it in the datastore
 110     task_image_entry = TaskImageMap(key_name=task.name,
 111                                     task_name=task.name,
 112                                     image_ref=image_ref)
 113     task_image_entry.put()
 114
 115   def post(self):
 116     """Stores the user-uploaded image in datastore and enqueues a task in pull
 117     queue to process it.
 118     """
 119     name = self.request.get('name')
 120     for arg in self.request.arguments():
 121       if arg.startswith('upload'):
 122         image_contents = self.request.get(arg)
 123         if len(image_contents) > 0:
 124           self.WriteEntry(name, image_contents)
 125           self.redirect('/imagestatus?name=' + name)
 126           break
 127
 128
 129 class TaskDataHandler(webapp.RequestHandler):
 130   """Handles the output of tasks posted from taskpuller client.
 131
 132   This is responsible for handling the data/output posted from the taskpuller
 133   client. Taskpuller client essentially has a pool of workers to pull
 134   tasks(enqueud by your app) from pull-taskqueues. A worker enqueues a
 135   task, does the desired processing on payload of the task and posts the data
 136   back to your application. This is the handler which handles this output and
 137   takes appropriate action. For example in this app, taskpuller post the image
 138   after processing(flipping) it and we store the processed image in datastore
 139   with original image-name appended with _processed.
 140   """
 141
 142   def GetImageNameFromTaskImageMap(self, key):
 143     """Returns corresponding image_name for the taskid."""
 144     task_image_entry = TaskImageMap.get_by_key_name(key)
 145     if task_image_entry:
 146       return task_image_entry.image_ref.image_name
 147     else:
 148       return None
 149
 150   def post(self):
 151     """Handles the output posted from taskqueue puller.
 152
 153     name is a queryparam that is the task_id of the task whose output is being
 154     posted by the puller. We first find the image_name for the task and store
 155     the processed image as image_name_processed. This helps when user searches
 156     for the image and we can show both uploaded as well as processed image.
 157     This handler is OAuth enabled, so that only the administrators of this app
 158     can call this handler using an OAuth token.
 159     """
 160
 161     try:
 162       user = oauth.get_current_user()
 163       # Currently there is no way to figure out if a given OAuth user is an
 164       # admin. users.is_current_user_admin() only takes care of logged in users,
 165       # not users who are presenting tokens on OAuth. We should fix this!
 166       if user and user.nickname() == 'svivek@google.com':
 167         task_name = self.request.get('name')
 168         name = self.GetImageNameFromTaskImageMap(task_name)
 169         if name:
 170           image_key = name + '_processed'
 171           image_contents = self.request.body
 172           small_image = SmallImage(key_name=image_key,
 173                                    image_name=image_key,
 174                                    image=db.Blob(image_contents))
 175           small_image.put()
 176         else:
 177           logging.error('No image associated with this task')
 178           self.error(404)
 179       else:
 180         logging.error('unauthorized user ' + user.nickname())
 181         self.error(403)
 182     except oauth.OAuthRequestError, e:
 183       logging.error('no oauth token detected')
 184       self.error(403)
 185
 186
 187 class ImageServer(webapp.RequestHandler):
 188   """Serves image from datastore."""
 189
 190   def get(self):
 191     """Get to show the image"""
 192     name = self.request.get('name')
 193     small_image = SmallImage.get_by_key_name(name)
 194     if small_image and small_image.image:
 195       self.response.headers['Content-Type'] = 'image/jpeg'
 196       self.response.out.write(small_image.image)
 197       logging.info('returned %d bytes' % len(small_image.image))
 198     else:
 199       self.error(404)
 200
 201
 202 class ImageStatusHandler(webapp.RequestHandler):
 203   """Lets user search by image name and shows both uploaded and processed
 204   image."""
 205
 206   def ImageExists(self, key):
 207     small_image = SmallImage.get_by_key_name(key)
 208     return small_image is not None
 209
 210   def get(self):
 211     name = self.request.get('name')
 212     logging.info('name is ' + name)
 213     if self.ImageExists(name):
 214       uploaded_url = '/images?name=' + name
 215       processed_url = '/images?name=' + name +'_processed'
 216       uploaded_url_str = '<img src=\"' + uploaded_url + '\" ></img> <br>'
 217       if self.ImageExists(name +"_processed"):
 218         processed_url_str = '<img src=\"' + processed_url + \
 219                             '\" >' + '</img> <br>'
 220       else:
 221         processed_url_str = 'Image is being processed, it will be eventaully\
 222         available at ' + '<a href=\"' + processed_url + '\" >' + 'here'\
 223         + '</a> <br>'
 224     else:
 225       self.response.out.write('No Image Uploaded named ' + name + '. Please Try\
 226                               again!')
 227       return None
 228
 229     self.response.out.write('''
 230                                <html>
 231                                  <body> ''')
 232
 233     self.response.out.write('''
 234                             <TABLE>
 235                             <CAPTION>Image Status for image %s</CAPTION>
 236                             <TR>
 237                             <TD>Uploaded Image : <TD> ''' % name)
 238     self.response.out.write(uploaded_url_str)
 239     self.response.out.write(''' <TR>
 240                             <TD> Processed Image : <TD> ''')
 241     self.response.out.write(processed_url_str)
 242     self.response.out.write(''' </TD>
 243                             </TABLE>''')
 244     self.response.out.write('''
 245       </body>
 246     </html>
 247     ''')
 248
 249
 250 class SearchHandler(webapp.RequestHandler):
 251   def get(self):
 252     """Searches image by the name given during upload."""
 253     self.response.out.write('''
 254     <html>
 255       <body>
 256         <form action="/search" method="post" enctype="multipart/form-data">
 257           <p>Image: <input type="text" name="name" /></p>
 258           <div><input type="submit" value="Search"></div>
 259         </form>
 260       </body>
 261     </html>
 262     ''')
 263
 264   def post(self):
 265     """Searches and shows both uploaded image and processed image. If task has
 266     notfinished yet, it shows the  tentative URL where the processed iamge will
 267     be available after task finishes."""
 268     name = self.request.get('name')
 269     self.redirect('/imagestatus?name=' + name)
 270
 271
 272 def main():
 273   application = webapp.WSGIApplication(
 274       [('/', ImageRequestHandler),
 275        ('/search', SearchHandler),
 276        ('/imagestatus', ImageStatusHandler),
 277        ('/images', ImageServer),
 278        ('/taskdata', TaskDataHandler),
 279       ],
 280       debug=True)
 281
 282   run_wsgi_app(application)
 283
 284 if __name__ == '__main__':
 285   main()