/usr/share/sugar/activities/Physics.activity/olpcgames/camera.py is in sugar-physics-activity 7+dfsg-1.2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 | """Accesses OLPC Camera functionality via gstreamer
Depends upon:
pygame
gstreamer (particularly gst-launch)
Activity demonstrating usage:
http://dev.laptop.org/git?p=projects/games-misc;a=tree;f=cameratest.activity;hb=HEAD
"""
import threading, subprocess
import logging
import olpcgames
import time
import os
import pygame
from olpcgames.util import get_activity_root
log = logging.getLogger( 'olpcgames.camera' )
#log.setLevel( logging.DEBUG )
CAMERA_LOAD, CAMERA_LOAD_FAIL = olpcgames.CAMERA_LOAD, olpcgames.CAMERA_LOAD
class Camera(object):
"""A class representing a still-picture camera
Produces a simple gstreamer bus that terminates in a filesink, that is,
it stores the results in a file. When a picture is "snapped" the gstreamer
stream is iterated until it finishes processing and then the file can be
read.
There are two APIs available, a synchronous API which can potentially
stall your activity's GUI (and is NOT recommended) and an
asynchronous API which returns immediately and delivers the captured
camera image via a Pygame event. To be clear, it is recommended
that you use the snap_async method, *not* the snap method.
Note:
The Camera class is simply a convenience wrapper around a fairly
straightforward gst-launch bus. If you have more involved
requirements for your camera manipulations you will probably
find it easier to write your own camera implementation than to
use this one. Basically we provide here the "normal" use case of
snapping a picture into a pygame image.
Note:
With the current camera implementation taking a single photograph
requires about 6 seconds! Obviously we'll need to figure out what's
taking gstreamer so long to process the pipe and fix that.
"""
_aliases = {
'camera': 'v4l2src',
'test': 'videotestsrc',
'testing': 'videotestsrc',
'png': 'pngenc',
'jpeg': 'jpegenc',
'jpg': 'jpegenc',
}
def __init__(self, source='camera', format='png', filename=None, directory = None):
"""Initialises the Camera's internal description
source -- the gstreamer source for the video to capture, useful values:
'v4l2src','camera' -- the camera
'videotestsrc','test' -- test pattern generator source
format -- the gstreamer encoder to use for the capture, useful values:
'pngenc','png' -- PNG format graphic
'jpegenc','jpg','jpeg' -- JPEG format graphic
filename -- the filename to use for the capture, if not specified defaults
to a random UUID + '.' + format
directory -- the directory in which to create the temporary file, defaults
to get_activity_root() + 'tmp'
"""
log.info( 'Creating camera' )
if not filename:
import uuid
filename = '%s.%s'%( uuid.uuid4(), format )
self.source = self._aliases.get( source, source )
self.format = self._aliases.get( format, format )
self.filename = filename
self.directory = directory
SNAP_PIPELINE = 'gst-launch','%(source)s','!','ffmpegcolorspace','!','%(format)s','!','filesink','location="%(filename)s"'
def _create_subprocess( self ):
"""Method to create the gstreamer subprocess from our settings"""
if not self.directory:
path = os.path.join( get_activity_root(), 'tmp' )
try:
os.makedirs( path )
log.info( 'Created temporary directory: %s', path )
except (OSError,IOError), err:
pass
else:
path = self.directory
filename = os.path.join( path, self.filename )
format = self.format
source = self.source
pipeline = [s%locals() for s in self.SNAP_PIPELINE ]
return filename, subprocess.Popen(
pipeline,stderr = subprocess.PIPE
)
def snap(self):
"""Snap a picture via the camera by iterating gstreamer until finished
Note: this is an unsafe implementation, it will cause the whole
activity to hang until the capture finishes. Time to finish is often
measured in whole seconds (3-6s).
It is *strongly* recommended that you use snap_async instead of snap!
"""
log.debug( 'Starting snap' )
filename, pipe = self._create_subprocess()
if not pipe.wait():
log.debug( 'Ending snap, loading: %s', filename )
return self._load_and_clean( filename )
else:
raise IOError( """Unable to complete snapshot: %s""", pipe.stderr.read() )
def _load_and_clean( self, filename ):
"""Use pygame to load given filename, delete after loading/attempt"""
try:
log.info( 'Loading snapshot file: %s', filename )
return pygame.image.load(filename)
finally:
try:
os.remove( filename )
except (IOError,OSError), err:
pass
def snap_async( self, token=None ):
"""Snap a picture asynchronously generating event on success/failure
token -- passed back as attribute of the event which signals that capture
is finished
We return events of type CAMERA_LOAD with an attribute "succeed"
depending on whether we succeed or not. Attributes of the events which
are returned:
success -- whether the loading process succeeded
token -- as passed to this method
image -- pygame image.load result if successful, None otherwise
filename -- the filename in our temporary directory we used to store
the file temporarily (this file will be deleted before the event
is sent, the name is for informational purposes only).
err -- Exception instance if failed, None otherwise
Basically identical to the snap method, save that it posts a message
to the event bus in pygame.event instead of blocking and returning...
Example:
if event == pygame.MOUSEBUTTONDOWN:
camera = Camera( source='test', filename = 'picture32' )
camera.snap_async( myIdentifier )
...
elif event.type == olpcgames.CAMERA_LOAD:
if event.token == myIdentifier:
doSomething( event.image )
"""
log.debug( 'beginning async snap')
t = threading.Thread(target=self._background_snap, args=[token])
t.start()
return token
def _background_snap(
self,
token = None,
):
"""Process gst messages until pipe is finished
pipe -- gstreamer pipe definition for parse_launch, normally it will
produce a file into which the camera should store an image
We consider pipe to be finished when we have had two "state changed"
gstreamer events where the pending state is VOID, the first for when
we begin playing, the second for when we finish.
"""
log.debug( 'Background thread kicking off gstreamer capture begun' )
from pygame.event import Event, post
filename, pipe = self._create_subprocess()
if not pipe.wait():
success = True
log.debug( 'Ending capture, loading: %s', filename )
try:
image = self._load_and_clean( filename )
except Exception, err:
image = None
success = False
else:
err = None
else:
success = False
err = pipe.stderr.read()
image = None
evt = Event(
CAMERA_LOAD,
dict(
filename=filename,
success = success,
token = token,
image=image,
err=err
)
)
post( evt )
def snap():
"""Dump a snapshot from the camera to a pygame surface in background thread
See Camera.snap
"""
return Camera().snap()
def snap_async( token=None, **named ):
"""Dump snapshot from camera return asynchronously as event in Pygame
See Camera.snap_async
"""
return Camera(**named).snap_async( token )
|