When ctypes comes to the rescue

I recently purchased a DSLR (Canon 1000D) and almost at the same time I read an article about remote controlling your camera using gphoto. I tried it out and thought it was cool. During the holidays I had some time over to spend on hacking and wanted to try controlling my camera from Python. To my disappointment there were no Python bindings included with Ubuntu for gphoto. I did some googling but couldn't find any pre-compiled bindings, what to do?

Well, I could always try doing it with ctypes.

From the docs:

ctypes is a foreign function library for Python. It provides C compatible data types, and allows calling functions in DLLs or shared libraries. It can be used to wrap these libraries in pure Python.

Note: I haven't done any serious stuff with ctypes nor gphoto before so if you find any errors etc please post a comment.

It amazed my how easy it is to use ctypes. Here's a snippet that will take a picture (from the first camera found), download the image to local storage and then delete it from the camera's storage.

import ctypes
import os

# gphoto structures
""" From 'gphoto2-camera.h'
typedef struct {
        char name [128];
        char folder [1024];
} CameraFilePath;
"""
class CameraFilePath(ctypes.Structure):
    _fields_ = [('name', (ctypes.c_char * 128)),
                ('folder', (ctypes.c_char * 1024))]

# gphoto constants
# Defined in 'gphoto2-port-result.h'
GP_OK = 0
# CameraCaptureType enum in 'gphoto2-camera.h'
GP_CAPTURE_IMAGE = 0
# CameraFileType enum in 'gphoto2-file.h'
GP_FILE_TYPE_NORMAL = 1

# Load library
gp = ctypes.CDLL('libgphoto2.so.2')

# Init camera
context = gp.gp_context_new()
camera = ctypes.c_void_p()
gp.gp_camera_new(ctypes.pointer(camera))
gp.gp_camera_init(camera, context)

# Capture image
cam_path = CameraFilePath()
gp.gp_camera_capture(camera,
                     GP_CAPTURE_IMAGE,
                     ctypes.pointer(cam_path),
                     context)

# Download and delete
cam_file = ctypes.c_void_p()
fd = os.open('image.jpg', os.O_CREAT | os.O_WRONLY)
gp.gp_file_new_from_fd(ctypes.pointer(cam_file), fd)
gp.gp_camera_file_get(camera,
                      cam_path.folder,
                      cam_path.name,
                      GP_FILE_TYPE_NORMAL,
                      cam_file,
                      context)
gp.gp_camera_file_delete(camera,
                         cam_path.folder,
                         cam_path.name,
                         context)
gp.gp_file_unref(cam_file)

# Release the camera
gp.gp_camera_exit(camera, context)
gp.gp_camera_unref(camera)

Ok, remember that I haven't done any wrapper or anything, it almost looks like 'C' code. I have also skipped all error checking for brevity.

You can always use a c_void_p if you don't need access to the data in Python (if you only need to pass a pointer between foreign functions). I'm using c_void_p instead of defining ctypes structures for gphoto's data types such as Camera and CameraFile. I still had to define CameraFilePath since I needed access to the data in Python.

I really like ctypes because I don't have to maintain code in C to access native functionality. Maybe you won't get the same performance as with traditional bindings but in this particular case it's not an issue.

FUSE - Filesystem in Userspace part 3 (final)

Finally, as I promised, the last blog post on implementing file systems using FUSE.

I've created a file system, shoutcastfs, which enables you to mount the Shoutcast Radio directory as a file system. The genres are represented as directories and stations as files. Each file contains the station's playlist and the files are suffixed with .pls which makes it possible to load the playlist in a media player such as Amarok by double-clicking the file.

Of course, I'm using pyshoutcast (Python shoutcast API) to access the shoutcast service.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import errno
import fuse
import stat
import os
import shoutcast

fuse.fuse_python_api = (0, 2)

_shoutcastApi = shoutcast.ShoutCast()

class RootInfo(fuse.Stat):
    def __init__(self):
        fuse.Stat.__init__(self)
        self.st_mode = stat.S_IFDIR | 0755
        self.st_nlink = 2
        self._genres = {}

    @property
    def genres(self):
        if not self._genres:
            for g in _shoutcastApi.genres():
                self._genres[g] = GenreInfo(g)
        return self._genres

class GenreInfo(fuse.Stat):
    def __init__(self, name):
        fuse.Stat.__init__(self)
        self.st_mode = stat.S_IFDIR | 0755
        self.st_nlink = 2
        self.name = name
        self._stations = {}

    @property
    def stations(self):
        if not self._stations:
            for s in _shoutcastApi.stations(self.name):
                name = '{0}.pls'.format(s[0])
                name = name.replace('/', '|')
                self._stations[name] = StationInfo(name, s[1])
        return self._stations

class StationInfo(fuse.Stat):
    def __init__(self, name, station_id):
        fuse.Stat.__init__(self)
        self.st_mode = stat.S_IFREG | 0644
        self.st_nlink = 1
        # Hope no playlist exceeds this size
        self.st_size = 4096
        self.name = name
        self.station_id = station_id
        self._content = None

    @property
    def content(self):
        if self._content is None:
            self._content = _shoutcastApi.tune_in(self.station_id).read()
        return self._content

class ShoutcastFS(fuse.Fuse):
    def __init__(self, *args, **kw):
        fuse.Fuse.__init__(self, *args, **kw)
        self.root = RootInfo()

    def split_path(self, path):
        """ Returns genre and station """
        if path == '/':
            return (None, None)

        parts = path.split('/')[1:]
        if len(parts) == 1:
            return (parts[0], None)
        else:
            return parts

    def getattr(self, path):
        genre, station = self.split_path(path)

        if genre is None:
            stat = self.root
        else:
            stat = self.root.genres.get(genre)
            if not stat:
                return -errno.ENOENT

            if station:
                stat = stat.stations.get(station)
                if not stat:
                    return -errno.ENOENT
        return stat

    def readdir(self, path, offset):
        yield fuse.Direntry('.')
        yield fuse.Direntry('..')

        if path == '/':
            entries = self.root.genres.keys()
        else:
            entries = self.root.genres[path[1:]].stations.keys()

        for e in entries:
            yield fuse.Direntry(e)

    def open(self, path, flags):
        # Only support for 'READ ONLY' flag
        access_flags = os.O_RDONLY | os.O_WRONLY | os.O_RDWR
        if flags & access_flags != os.O_RDONLY:
            return -errno.EACCES
        else:
            return 0

    def read(self, path, size, offset):
        genre, station = self.split_path(path)
        info = self.root.genres[genre].stations[station]
        if offset < info.st_size:
            if offset + size > info.st_size:
                size = info.st_size - offset
            return info.content[offset:offset+size]
        else:
            return ''

if __name__ == '__main__':
    fs = ShoutcastFS()
    fs.parse(errex=1)
    fs.multithreaded = False
    fs.main()

To try the file system, run:

$ # Download shoutcast.py
$ wget http://github.com/mariob/pyshoutcast/raw/master/src/shoutcast.py
$ mkdir mnt
$ ./shoutcastfs mnt
$ cd mnt/
$ ls
...A list of genres...
$ cd Samba
$ ls
...A list of 'Samba' stations...
$ cat [station name]
...Playlist data...

Have fun!

FUSE - Filesystem in Userspace part 2

This is the second post on implementing file systems using FUSE. In this post I'll show you how to create a read-only file system that is backed by an XML file. Each tag in the XML file is either a file or a directory. If the tag is representing a directory, it should have a is_dir="1" attribute added. When the tag represents a file it should have its content between the tags. The following XML snippet shows an example:

<root is_dir="1">

  <a is_dir="1">
    <file1.txt>File content for this file</file1.txt>
    <hello>Hello World</hello>
  </a>

  <empty_dir is_dir="1" />

</root>

The example models the following directory tree:

/
|-- a
|   |-- file1.txt
|   `-- hello
`-- empty_dir

There are some limitations in having an XML file representing a file system. You can't for example have filenames starting with a dot because tags are not allowed to start with a dot in XML.

On to the code, the following snippet implements the read-only file system which is backed by an XML file:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import errno
import fuse
import stat
import os
import time
import xml.etree.ElementTree as etree

fuse.fuse_python_api = (0, 2)

# Use same timestamp for all files
_file_timestamp = int(time.time())

class MyStat(fuse.Stat):
    """
    Convenient class for Stat objects.
    Set up the stat object with appropriate
    values depending on constructor args.
    """
    def __init__(self, is_dir, size):
        fuse.Stat.__init__(self)
        if is_dir:
            self.st_mode = stat.S_IFDIR | 0555
            self.st_nlink = 2
        else:
            self.st_mode = stat.S_IFREG | 0444
            self.st_nlink = 1
            self.st_size = size
        self.st_atime = _file_timestamp
        self.st_mtime = _file_timestamp
        self.st_ctime = _file_timestamp

class MyFS(fuse.Fuse):
    def __init__(self, xml_tree, *args, **kw):
        fuse.Fuse.__init__(self, *args, **kw)
        self.tree = xml_tree

    def getattr(self, path):
        # We do not support 'dot' files
        # since xml tags cannot start with a dot.
        if path.find('/.') != -1:
            return -errno.ENOENT

        entry = self.tree.find(path)

        if entry is None:
            return -errno.ENOENT
        else:
            is_dir = entry.get('is_dir', False)
            size = entry.text and len(entry.text.strip()) or 0
            return MyStat(is_dir, size)

    def readdir(self, path, offset):
        yield fuse.Direntry('.')
        yield fuse.Direntry('..')
        for e in self.tree.find(path).getchildren():
            yield fuse.Direntry(e.tag)

    def open(self, path, flags):
        # Only support for 'READ ONLY' flag
        access_flags = os.O_RDONLY | os.O_WRONLY | os.O_RDWR
        if flags & access_flags != os.O_RDONLY:
            return -errno.EACCES
        else:
            return 0

    def read(self, path, size, offset):
        entry = self.tree.find(path)
        content = entry.text and entry.text.strip() or ''
        file_size = len(content)
        if offset < file_size:
            if offset + size > file_size:
                size = file_size - offset
            return content[offset:offset+size]
        else:
            return ''

if __name__ == '__main__':
    tree = etree.parse('tree.xml')

    fs = MyFS(tree)
    fs.parse(errex=1)
    fs.main()

I'm using the ElementTree XML API to parse the XML file. The ElementTree API is very easy to use and supports finding elements by specifying a path which fits very well in this context. For example, the following snippet shows how to get the file1.txt element from the example XML file above and extract the content between the tags:

import xml.etree.ElementTree as etree
tree = etree.parse('tree.xml')
el = tree.find('/a/file1.txt')
print el.text.strip() # Remove any white-spaces

The code hasn't change that much since the part 1 post. I've added two new methods which adds support for opening and reading files. The MyStat class is only used as an convenient class to help creating appropriate stat objects. You might notice that I don't do a lot of checking in the code, this is because FUSE do a lot of them for me, this page list some of the assumptions you can make when implementing file system using FUSE.

Did you believe it would be this easy to create a mountable file system that uses an XML file for the layout? I didn't.

In the next (and final) post about FUSE I think I'll create some kind of 'use a service on the Internet' file system which can be useful and not just another toy fs.

FUSE - Filesystem in Userspace part 1

Here's the first part in a series of posts on implementing file systems using FUSE.

Creating a file system can seem to be an intimidating task but with FUSE it's very easy and it gets even easier with python bindings. Any decent Linux distribution should include support for FUSE and be configured such that regular users (or a group) can mount file systems written with FUSE without having root access.

If you are running Ubuntu you need to install the python-fuse package which contains the Python bindings for FUSE.

- Enough, show me the code

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import errno
import fuse
import stat
import time

fuse.fuse_python_api = (0, 2)

class MyFS(fuse.Fuse):
    def __init__(self, *args, **kw):
        fuse.Fuse.__init__(self, *args, **kw)

    def getattr(self, path):
        st = fuse.Stat()
        st.st_mode = stat.S_IFDIR | 0755
        st.st_nlink = 2
        st.st_atime = int(time.time())
        st.st_mtime = st.st_atime
        st.st_ctime = st.st_atime

        if path == '/':
            pass
        else:
            return - errno.ENOENT
        return st

if __name__ == '__main__':
    fs = MyFS()
    fs.parse(errex=1)
    fs.main()

The example above shows how easy it is to create a mountable file system, it does nothing but is still a valid file system. The MyFS class implements a file system with no entries, it will always return ENOENT (No such file or directory) except for the root path. You should take a look at the FUSE Python Reference page which contains details about FusePython and, among other things, describes the stat object returned from the getattr method.

Paste the example to a file, MyFS.py, and make the file executable:

$ chmod 755 MyFS.py

Now create a directory which will be used as mount point and mount the file system:

$ mkdir myfs
$ ls -l
totalt 4
drwxr-xr-x 2 mario mario 4096 2009-11-15 15:59 myfs
-rwxr-xr-x 1 mario mario  648 2009-11-15 15:59 MyFS.py
$ # This will mount the file system, try ./MyFS.py --help
$ ./MyFS.py myfs
$ ls -l
totalt 0
drwxr-xr-x 2 root  root    0 2009-11-15 16:10 myfs
-rwxr-xr-x 1 mario mario 648 2009-11-15 15:59 MyFS.py

Here we can see that the owner, size and time stamp changed on myfs after we mounted the file system. You can cd into myfs, but if you try to list the directory (file system root) you'll get an error:

$ cd myfs
$ ls
ls: reading directory .: Function not implemented

You got this error because we haven't implemented the readdir method which is invoked when listing a directory. To make this work we need to add a readdir method which returns (at least) the '.' and '..' entries. Add the following snippet to the class:

    def readdir(self, path, offset):
        for e in '.', '..':
            yield fuse.Direntry(e) 

Before you continue you should unmount the file system:

$ fusermount -u myfs

This will unmount your FUSE file system as a regular user and doesn't require root access.

Retry to mount and cd into the myfs directory:

$ ./MyFS.py myfs
$ cd myfs
$ ls -a
.  ..

Alright, I think I'll stop for today. Oh, just one more thing... You won't get any print outputs nor tracebacks from your code if you don't start the file system in foreground mode. This is done by adding '-f' as argument when starting the file system.

$ ./MyFS.py -f myfs

It can be hard to tell why something doesn't work if you can't see tracebacks and print outputs.