7.28.2013

Reading Event Data on Google Calendar in Python

Python: Google Calendar のイベント情報を読み取るスクリプト

 

ライブラリのインストール

google-api-python-client を利用する。

google-api-python-client - Google APIs Client Library for Python - Google Project Hosting

$ sudo pip install google-api-python-client

 

認証用のファイルを準備

このような画面操作でサンプルを手に入れることができる。それを参考にする。

Installation  Google APIs Client Library for Python  Google Developers

認証は、client_secrets.json というJSON形式のファイルを使うのが推奨されているようだ。
このファイルも手打ちする必要はなく、ダウンロードボタン一発で入手できる。 

Client Secrets - Google APIs Client Library for Python — Google Developers

サンプルをたたくと、初回に client_secrets.json を元にFlow オブジェクトが生成される。

このとき、どのAPIにどのレベルでアクセスするかを指定する。
今回はカレンダーの読み取りだけなので、'https://www.googleapis.com/auth/calendar.readonly' があれば十分。

その内容が Storage オブジェクトとなり、ファイルに保存される。
今回は read_calendar.dat というファイル名にした。 

認証情報(credentials)が完成したら、httplib2、build() 関数を使って Calendar API v3 にアクセスすればよい。 

 

コード

今日一日の予定を出力するコード。
プログラムの第一引数に整数を入れれば、n日後の予定を取得できる。

対象のカレンダーIDは予め調べて、ハードコーディング。

予定の開始時刻、終了時刻、内容、作成者をソートして出力するため、OneDayEvent というクラスを作っている。

タイムゾーンの調整はしていない。(手抜き)
もっとコメント書いた方がいいけどこれも手抜き。

Calendar API のサービスを取得した後、
service.events()list(...).execute() で条件に見合ったイベントのリストを取得している。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import httplib2
import os
import sys
import datetime

from apiclient.discovery import build
from oauth2client.file import Storage
from oauth2client.client import AccessTokenRefreshError
from oauth2client.client import flow_from_clientsecrets
from oauth2client.tools import run

# Calendar id for reading.
CALENDAR_ID = 'xxxxxxxxxxxxxxxxxxxxxxxxxx@xxxxx.calendar.google.com'

# Path to the credential files.
BASE_DIR = lambda x: os.path.join(os.path.dirname(__file__), x)

CLIENT_SECRETS_PATH = BASE_DIR('client_secrets.json')
STORAGE_PATH = BASE_DIR('read_calendar.dat')

# Set up a Flow object to be used for authentication.
FLOW = flow_from_clientsecrets(CLIENT_SECRETS_PATH, scope=[
    'https://www.googleapis.com/auth/calendar.readonly',
])


def get_calendar_service():
    storage = Storage(STORAGE_PATH)
    credentials = storage.get()

    if credentials is None or credentials.invalid:
        credentials = run(FLOW, storage)

    http = httplib2.Http()
    http = credentials.authorize(http)

    service = build('calendar', 'v3', http=http)
    return service


class OneDayEvent(object):
    def __init__(self, start_time, end_time, summary, creator):
        if start_time is None:
            assert(end_time is None)
        self.start_time = start_time
        self.end_time = end_time
        self.summary = summary
        self.creator = creator

    def __lt__(self, that):
        if self.start_time is None and that.start_time is None:
            return self.summary < that.summary
        if self.start_time is None or that.start_time is None:
            return self.start_time is None
        return (self.start_time, self.end_time, self.summary, self.creator) < (
            that.start_time, that.end_time, that.summary, that.creator)

    def __str__(self):
        if self.start_time is None:
            tm = u'終日'
        else:
            f = lambda x: datetime.datetime.strftime(x, '%H:%M')
            tm = '%s-%s' % (f(self.start_time), f(self.end_time))
        return u'[%s] %s (%s)' % (tm, self.summary, self.creator)


def read_events(date):
    """Returns sorted list of OneDayEvent objects."""

    def format(x):
        return datetime.datetime.strftime(x, '%Y-%m-%dT%H:%M:%SZ')

    time_min = datetime.datetime.combine(date, datetime.time())
    time_max = time_min + datetime.date.resolution

    time_min, time_max = map(format, (time_min, time_max))

    service = get_calendar_service()
    events = service.events().list(
        calendarId=CALENDAR_ID, timeMin=time_min, timeMax=time_max).execute()

    def f(x):
        y = x.get('dateTime')
        if y:
            z = datetime.datetime.strptime(y[:-6], '%Y-%m-%dT%H:%M:%S')
            return datetime.datetime.combine(date, z.time())

    ret = [OneDayEvent(
        f(event['start']), f(event['end']), event['summary'],
        event['creator']['displayName']) for event in events['items']]
    ret.sort()
    return ret


def main(argv):
    offset = int(argv[1]) if len(argv) >= 2 else 0

    for event in read_events(
            datetime.date.today() + datetime.date.resolution * offset):
        print(unicode(event))

if __name__ == '__main__':
    main(sys.argv)

 

 

  • 出力例
    [終日] 有給休暇 (James LaBrie)
    [11:00-12:00] ○○様来社 (John Petrucci)
    [14:00-15:00] [外出] △△訪問 (John Myung)
    [17:30-22:00] □□勉強会 (Jordan Rudess)
    [17:30-23:00] ☆☆飲み会 (Mike Mangini)

 

References

 

Python: Sending IRC Message, Improved

Python: IRCメッセージ送信処理の改善

 

こちらのエントリの改善。
mog project: How to Send an IRC Message with SSL in Python

複数のメッセージを一度に送れるようにした。

あと、disconnect() の時にサーバでエラーが出ていたので削除した。

# -*- coding: utf-8 -*-
 
import ssl
 
SERVER_ADDRESS = 'xxx.xxx.xxx.xxx'
SERVER_PORT = 6667
SERVER_PASS = 'xxxxxx'
 
LOGIN_NAME = 'xxxxxx'
LOGIN_CHANNEL = '#xxxxxx'


def send(message):
    try:
        import irc.client
        import irc.connection

        if isinstance(message, basestring):
            message = [message]

        client = irc.client.IRC()
        server = client.server()
        factory = irc.connection.Factory(wrapper=ssl.wrap_socket)

        c = server.connect(
            SERVER_ADDRESS, SERVER_PORT, LOGIN_NAME, SERVER_PASS,
            connect_factory=factory)

        for m in message:
            c.privmsg(LOGIN_CHANNEL, m)
    except Exception as e:
        print('WARN: Failed to send IRC message. (%s)' % e)

7.22.2013

Counting Bits in Python, C++ and Scala (Pt.2)

各プログラミング言語で バイナリファイルの中にある 1 のビットの数をカウントする (その2)

 

こちらのエントリの続き
mog project: Counting Bits in Python, C++ and Scala (Pt.1)

 

C++ 実測 (3)

方法(1)を途中の段階まで実施して配列を横断して加算していく方法。
Hacker's Delight の実装を参考にした。

8ビット単位まで各個計算を行うバージョンを METHOD 3、16ビット単位のバージョンを METHOD 4 としている。

コード

これまでのコードも含め、全体を掲載する。

#define METHOD 3

#include <fstream>
#include <vector>
#include <string>
#include <ctime>
#include <cassert>

using namespace std;

int const kChunkSize = 1 << 14;

// METHOD 1
int POP(unsigned long long x) {
  unsigned long long const y = 0x3333333333333333ULL;
  x -= (x >> 1) & 0x5555555555555555ULL;
  x = (x & y) + ((x >> 2) & y);
  x = (x + (x >> 4)) & 0x0f0f0f0f0f0f0f0fULL;
  x += x >> 8;
  x += x >> 16;
  x += x >> 32;
  return x & 0x7f;
}

// METHOD 2
int table[1 << 16];

void init_table() {
  for (int i = 0; i < (1 << 16); ++i) table[i] = POP(i);
}

int POP2(unsigned long long x) {
  return table[x & 0xffff] +
      table[(x >> 16) & 0xffff] +
      table[(x >> 32) & 0xffff] +
      table[(x >> 48)];
}

// METHOD 3
int POP_ARRAY(unsigned long long a[], int n) {
  unsigned long long s, s8, x;
  unsigned long long const y = 0x3333333333333333ULL;

  s = 0;
  for (int i = 0; i < n; i += 31) {
    int lim = min(n, i + 31);
    s8 = 0;
    for (int j = i; j < lim; ++j) {
      x = a[j];
      x -= (x >> 1) & 0x5555555555555555ULL;
      x = (x & y) + ((x >> 2) & y);
      x = (x + (x >> 4)) & 0x0f0f0f0f0f0f0f0fULL;
      s8 += x;
    }
    x = (s8 & 0x00ff00ff00ff00ffULL) + ((s8 >> 8) & 0x00ff00ff00ff00ffULL);
    x += x >> 16;
    x += x >> 32;
    s += x & 0xffff;
  }
  return s;
}

// METHOD 4
int POP_ARRAY2(unsigned long long a[], int n) {
  unsigned long long s, s16, x;
  unsigned long long const y = 0x3333333333333333ULL;

  s = 0;
  for (int i = 0; i < n; i += 4095) {
    int lim = min(n, i + 4095);
    s16 = 0;
    for (int j = i; j < lim; ++j) {
      x = a[j];
      x -= (x >> 1) & 0x5555555555555555ULL;
      x = (x & y) + ((x >> 2) & y);
      x = (x + (x >> 4)) & 0x0f0f0f0f0f0f0f0fULL;
      x = (x + (x >> 8)) & 0x00ff00ff00ff00ffULL;
      s16 += x;
    }
    x = (s16 & 0x0000ffff0000ffffULL) + ((s16 >> 16) & 0x0000ffff0000ffffULL);
    x += x >> 32;
    s += x & 0xffffffff;
  }
  return s;
}

template <typename T, typename U>
double time_it(int iter, int count, T func, U val) {
  double elapsed = 1e30;
  for (int i = 0; i < iter; ++i) {
    clock_t t = clock();
    for (int j = 0; j < count; ++j) {
      func(val);
    }
    clock_t s = clock();
    elapsed = min(elapsed, static_cast<double>(s - t) / CLOCKS_PER_SEC);
  }
  return elapsed;
}

int count_bit(string path) {
  int ret = 0;
  unsigned long long x[kChunkSize];

  ifstream fin(path.c_str(), ios::in | ios::binary);
  while (true) {
    fin.read(reinterpret_cast<char *>(&x), sizeof(unsigned long long) * kChunkSize);
    if (fin.eof()) break;
#if METHOD == 1
    for (int i = 0; i < kChunkSize; ++i) ret += POP2(x[i]);
#elif METHOD == 2
    for (int i = 0; i < kChunkSize; ++i) ret += POP2(x[i]);
#elif METHOD == 3
    ret += POP_ARRAY(x, kChunkSize);
#elif METHOD == 4
    ret += POP_ARRAY2(x, kChunkSize);
#endif
  }
  fin.close();
  return ret;
}

int main(int argc, char *argv[]) {
  string paths[] = {"00.bin", "01.bin", "02.bin", "03.bin", "04.bin", "05.bin",
      "06.bin", "07.bin", "08.bin", "09.bin", "10.bin", "11.bin"};
  string prefix = "../data/";
  int expected[] = {0, 1, 10, 100, 1000, 10000, 100000, 1000000, 10000000,
      100000000, 1000000000, 1 << 30};

  init_table();
  for (int i = 0; i < 12; ++i) {
    string path = prefix + paths[i];
    assert(count_bit(path) == expected[i]);
    printf("%s: %f sec\n", paths[i].c_str(), time_it(3, 10, count_bit, path));
  }
  return 0;
}

 

実行結果
00.bin: 0.571133 sec
01.bin: 0.563557 sec
02.bin: 0.562696 sec
03.bin: 0.559637 sec
04.bin: 0.563524 sec
05.bin: 0.567378 sec
06.bin: 0.567318 sec
07.bin: 0.563231 sec
08.bin: 0.561556 sec
09.bin: 0.563927 sec
10.bin: 0.566986 sec
11.bin: 0.563692 sec

1 回の処理あたり 約 56ms。テーブル検索版よりも速い結果となった。

 

C++ 実測 (4)

実測(3) の配列横断処理を行う前処理の単位を 8ビットから 16ビットに変更して計測。
コードの1行目を #define METHOD 4 とする

実行結果
00.bin: 0.646128 sec
01.bin: 0.645220 sec
02.bin: 0.640301 sec
03.bin: 0.638049 sec
04.bin: 0.636916 sec
05.bin: 0.634473 sec
06.bin: 0.638034 sec
07.bin: 0.649389 sec
08.bin: 0.642163 sec
09.bin: 0.639769 sec
10.bin: 0.635853 sec
11.bin: 0.637499 sec

1 回の処理あたり 約 63ms。8ビット版よりも遅くなってしまった。

 

 

Scala の並列処理の実測はまた後日。

7.21.2013

mogtype - The Simplest Kana Typing Training Tool

mogtype - 最もシンプルなタイピングソフト

 

かな入力で、さらさら日本語の文章が書けるようになりたい!と最近思っているのだが
英語キーボードでも動く「かな入力」の練習ソフトがなかなか見当たらない。(まぁ相当なニッチ需要だけど) 

という訳で、かな入力に特化した、
コマンドラインで動く簡単なタイピングソフトを Python + curses で作ってみた。

Mac でのみ動作確認済み。

  • GitHub
    mogproject/mogtype
  • ヘルプ
    $ ./mogtype.py -h
    Usage: mogtype.py [options]
    
    Options:
      --version             show program's version number and exit
      -h, --help            show this help message and exit
      -f PATH, --file=PATH  path to the sentence database (default: mogtype.txt)
      -c COUNT, --count=COUNT
                            number of exercises (default: 8)
  • 使用方法

    引数(-f)で指定したテキストファイル(省略時は mogtype.txt)の内容がランダムでお題になる。

    $ ./mogtype.py -f ./mogtype.txt
  • 画面サンプル
    Screenshot 7 21 13 22 12 2
    カーソルで現在入力中の位置を表現。

    Screenshot 7 21 13 22 14 4
    ミスをするとこんな感じ。

    なお、画面上のtypoは修正済み。(Accurancy->Accuracy)
     
  • このファイルを変更すれば、異なるキーマッピングにも対応できる
    mogtype/keymap.py at master · mogproject/mogtype 

 

とりあえず動く状態になったけど、今後随時バージョンアップしていきたい。

 

 

References

ひらがな/カタカナの変換はこちらを参考にさせて頂きました。

7.18.2013

Counting Bits in Python, C++ and Scala (Pt.1)

各プログラミング言語で バイナリファイルの中にある 1 のビットの数をカウントする (その1)

 

目的

128 MB のバイナリファイル中にある 1 のビットの数を知りたい。
なお、128 MB = 2^7 MB = 2^17 KB = 2^27 Byte = 2^30 Bit なので求める値は 0 以上 2^30 以下となる。 

これを、並列処理なしの Python, C++, 並列処理ありの Scala でコードを書き、
手元の Mac で所要時間を調べてみる。

 

おことわり
実施環境も処理内容も、(私自身の勉強のために)とりあえず今作れるもので試しただけですので
言語自体の性能を比較する意図はございません。

  

データ作成

実測の前に、1のビットが n 個含まれるランダムなバイナリデータを予め用意しておく。

これは Python で、bitarray ライブラリを利用して行った。
さらに、大きな bitarray に対して random.shuffle を行うと所要時間が恐ろしく長くなるので
2^12 ビットのサイズのバケットを予めランダムに作っておくことで処理時間を短縮した。

参考:
mog project: Python: Random Partition of Integer 

 

bitarray のインストール

bitarray 0.8.1 : Python Package Index

$ sudo pip install bit array
コード
import random
import math
import timeit
from bitarray import bitarray
from random_partition import random_partition

ARRAY_SIZE = 1 << 30
CHUNK_SIZE = 1 << 12

TEST_COUNTS = [
    (0, '00.bin'), (1, '01.bin'), (10, '02.bin'), (100, '03.bin'),
    (1000, '04.bin'), (10000, '05.bin'), (100000, '06.bin'),
    (1000000, '07.bin'), (10000000, '08.bin'), (100000000, '09.bin'),
    (1000000000, '10.bin'), (1 << 30, '11.bin')]


def make_bitarray(size, count):
    assert(count <= size)

    ret = bitarray(size)
    if count == size:
        ret.setall(True)
    else:
        ret.setall(False)
        if count != 0:
            ret[:count] = True
            random.shuffle(ret)
    return ret


def make_file(count, path):
    # Create buckets.
    num_buckets = int(math.ceil(float(ARRAY_SIZE) / CHUNK_SIZE))
    buckets = random_partition(num_buckets, count, CHUNK_SIZE)

    # Write to file.
    with open(path, 'wb') as fh:
        for b in buckets:
            make_bitarray(CHUNK_SIZE, b).tofile(fh)
    print('Created: %s' % path)


if __name__ == '__main__':
    timeit.__dict__.update(make_file=make_file)
    for t in TEST_COUNTS:
        time = timeit.Timer('make_file(%d, "%s")' % (t[0], t[1])).timeit(1)
        print('%f sec' % time)

 

実行結果

AWS (t1.micro) で一晩コトコト煮込んだデータ。(約7.2時間)
gzip 圧縮して全部で 41 MB。 

$ python ./make_test_file.py
Created: 00.bin
3.845612 sec
Created: 01.bin
3.803929 sec
Created: 02.bin
3.650161 sec
Created: 03.bin
4.306574 sec
Created: 04.bin
10.518001 sec
Created: 05.bin
47.784271 sec
Created: 06.bin
213.532356 sec
Created: 07.bin
716.523029 sec
Created: 08.bin
4907.577745 sec
Created: 09.bin
8589.099633 sec
Created: 10.bin
11360.644969 sec
Created: 11.bin
30.384724 sec

Python 実測

bitarray の count 関数を使うだけ。

コード
import timeit
from bitarray import bitarray


TEST_COUNTS = [
    (0, '00.bin'), (1, '01.bin'), (10, '02.bin'), (100, '03.bin'),
    (1000, '04.bin'), (10000, '05.bin'), (100000, '06.bin'),
    (1000000, '07.bin'), (10000000, '08.bin'), (100000000, '09.bin'),
    (1000000000, '10.bin'), (1 << 30, '11.bin')]
PATH_PREFIX = '../data/'


def count_bit(path):
    x = bitarray()
    with open(path, 'rb') as fh:
        x.fromfile(fh)
    return x.count()

if __name__ == '__main__':
    for t in TEST_COUNTS:
        path = PATH_PREFIX + t[1]
        assert(count_bit(path) == t[0])
        timeit.__dict__.update(count_bit=count_bit)
        time = min(timeit.Timer('count_bit("%s")' % path).repeat(3, 10))
        print('%s: %f sec' % (t[1], time))

 

実行結果
00.bin: 1.858225 sec
01.bin: 1.860906 sec
02.bin: 1.858306 sec
03.bin: 1.846741 sec
04.bin: 1.853473 sec
05.bin: 1.840856 sec
06.bin: 1.836229 sec
07.bin: 1.852755 sec
08.bin: 1.864767 sec
09.bin: 1.851672 sec
10.bin: 1.839153 sec
11.bin: 1.824429 sec

1 回の処理あたり 約 185ms。
ちなみにそのおよそ半分がファイル読み込みに要する時間。

 

C++ 実測 (1)

まずは普通のカウント方法で。
一度にストリームから読み込むサイズは試行錯誤の末、2^14 (個の long long int) に落ち着いた。

if stream は read() した後で EOFチェックしないと正しい値にならない。

コード
#include <fstream>
#include <vector>
#include <string>
#include <ctime>
#include <cassert>

using namespace std;

int const kChunkSize = 1 << 14;

int POP(unsigned long long x) {
  unsigned long long const y = 0x3333333333333333ULL;
  x -= (x >> 1) & 0x5555555555555555ULL;
  x = (x & y) + ((x >> 2) & y);
  x = (x + (x >> 4)) & 0x0f0f0f0f0f0f0f0fULL;
  x += x >> 8;
  x += x >> 16;
  x += x >> 32;
  return x & 0x7f;
}

template <typename T, typename U>
double time_it(int iter, int count, T func, U val) {
  double elapsed = 1e30;
  for (int i = 0; i < iter; ++i) {
    clock_t t = clock();
    for (int j = 0; j < count; ++j) {
      func(val);
    }
    clock_t s = clock();
    elapsed = min(elapsed, static_cast<double>(s - t) / CLOCKS_PER_SEC);
  }
  return elapsed;
}

int count_bit(string path) {
  int ret = 0;
  unsigned long long x[kChunkSize];

  ifstream fin(path.c_str(), ios::in | ios::binary);
  while (true) {
    fin.read(reinterpret_cast<char *>(&x), sizeof(unsigned long long) * kChunkSize);
    if (fin.eof()) break;
    for (int i = 0; i < kChunkSize; ++i) ret += POP(x[i]);
  }
  fin.close();
  return ret;
}

int main(int argc, char *argv[]) {
  string paths[] = {"00.bin", "01.bin", "02.bin", "03.bin", "04.bin", "05.bin",
      "06.bin", "07.bin", "08.bin", "09.bin", "10.bin", "11.bin"};
  string prefix = "../data/";
  int expected[] = {0, 1, 10, 100, 1000, 10000, 100000, 1000000, 10000000,
      100000000, 1000000000, 1 << 30};

  for (int i = 0; i < 12; ++i) {
    string path = prefix + paths[i];
    assert(count_bit(path) == expected[i]);
    printf("%s: %f sec\n", paths[i].c_str(), time_it(3, 10, count_bit, path));
  }
  return 0;
}

 

実行結果
00.bin: 0.821018 sec
01.bin: 0.808333 sec
02.bin: 0.809245 sec
03.bin: 0.809044 sec
04.bin: 0.809715 sec
05.bin: 0.809076 sec
06.bin: 0.813601 sec
07.bin: 0.813362 sec
08.bin: 0.812772 sec
09.bin: 0.808916 sec
10.bin: 0.811492 sec
11.bin: 0.811594 sec

1 回の処理あたり 約 81ms。

 

C++ 実測 (2)

テーブル検索を行うバージョン。

コード
int table[1 << 16];

void init_table() {
  for (int i = 0; i < (1 << 16); ++i) table[i] = POP(i);
}

int POP2(unsigned long long x) {
  return table[x & 0xffff] +
      table[(x >> 16) & 0xffff] +
      table[(x >> 32) & 0xffff] +
      table[(x >> 48)];
}

 

実行結果
00.bin: 0.608648 sec
01.bin: 0.598598 sec
02.bin: 0.597612 sec
03.bin: 0.601898 sec
04.bin: 0.600394 sec
05.bin: 0.596191 sec
06.bin: 0.595793 sec
07.bin: 0.600928 sec
08.bin: 0.615904 sec
09.bin: 0.622860 sec
10.bin: 0.629954 sec
11.bin: 0.600881 sec

1 回の処理あたり 約 60ms。だいぶ速くなった。

 

 

残り(C++で配列中のビットを累積的に調べる/Scala実測を行う予定)はまた次回。

7.16.2013

Installing tmux with Fabric

Fabric で tmux をインストール

yum でインストールするだけだけど。

# -*- coding: utf-8 -*-
"""
Installing tmux.
"""

from fabric.api import *
from fabric.decorators import runs_once, roles, task

env.user = 'ec2-user'
env.use_ssh_config = True

@task
def install():
    """Install tmux."""
    sudo('yum -y install tmux')

これで、寝ている間も安心して AWS に仕事を任せられる。

 

References

tmux 公式

AWS: EC2 Restore Script

AWS: EC2インスタンスをスナップショットから復元するスクリプト

 

スナップショットのリストアに伴う一連の操作をスクリプト化。

デバイスが EBS 1個だけの場合にのみ対応。

コード

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import subprocess
import time
try:
    import json
except ImportError:
    print('You need python 2.6 or later to run this script.')
    sys.exit(1)


def usage():
    print('Usage: %s <instance-id> <snapshot-id>' % sys.argv[0])
    sys.exit(2)


def run_command(*args):
    output = subprocess.check_output(['aws', 'ec2'] + list(args))
    return json.loads(output)


def get_volume_id(instance):
    assert(len(instance['BlockDeviceMappings']) == 1)
    return instance['BlockDeviceMappings'][0]['Ebs']['VolumeId']


def get_zone(instance):
    return instance['Placement']['AvailabilityZone']


def get_device(instance):
    assert(len(instance['BlockDeviceMappings']) == 1)
    return instance['BlockDeviceMappings'][0]['DeviceName']


def is_running(instance):
    return instance['State']['Code'] == 16


def is_stopped(instance):
    return instance['State']['Code'] == 80


def get_instance(instance_id):
    j = run_command('describe-instances', '--instance-ids', instance_id)
    return j['Reservations'][0]['Instances'][0]


def start_instance(instance_id):
    sys.stdout.write('Starting instance: %s ...' % instance_id)
    run_command('start-instances', '--instance-ids', instance_id)

    for i in range(20):
        sys.stdout.write('.')
        time.sleep(10)
        if is_running(get_instance(instance_id)):
            break
    else:
        raise(RunTimeError('Timed out for waiting.'))
    print('OK')


def stop_instance(instance_id):
    sys.stdout.write('Stopping instance: %s ...' % instance_id)
    run_command('stop-instances', '--instance-ids', instance_id)

    for i in range(20):
        sys.stdout.write('.')
        time.sleep(10)
        if is_stopped(get_instance(instance_id)):
            break
    else:
        raise(RunTimeError('Timed out for waiting.'))
    print('OK')


def detach_volume(volume_id):
    sys.stdout.write('Detaching volume: %s ...' % volume_id)
    run_command('detach-volume', '--volume-id', volume_id)
    print('OK')


def create_volume(zone, snapshot):
    sys.stdout.write('Creating volume from snapshot: %s ...' % snapshot)
    j = run_command(
        'create-volume', '--availability-zone', zone,
        '--snapshot-id', snapshot)
    print('OK')
    return j['VolumeId']


def attach_volume(volume_id, instance_id, device):
    sys.stdout.write('Attaching volume: %s ...' % volume_id)
    run_command(
        'attach-volume', '--volume-id', volume_id,
        '--instance-id', instance_id, '--device', device)
    print('OK')


def delete_volume(volume_id):
    sys.stdout.write('Deleting volume: %s ...' % volume_id)
    run_command('delete-volume', '--volume-id', volume_id)
    print('OK')


if __name__ == '__main__':
    if len(sys.argv) != 3:
        usage()

    instance_id = sys.argv[1]
    snapshot = sys.argv[2]

    sys.stdout.write('Checking instance: %s ...' % instance_id)
    ins = get_instance(instance_id)

    old_vol = get_volume_id(ins)
    print('OK')

    stop_instance(instance_id)
    detach_volume(old_vol)
    new_vol = create_volume(get_zone(ins), snapshot)
    attach_volume(new_vol, instance_id, get_device(ins))
    start_instance(instance_id)
    delete_volume(old_vol)

実行例

$ ./aws_restore.py i-xxxxxxxx snap-XXXXXXXX
Checking instance: i-xxxxxxxx ...OK
Stopping instance: i-xxxxxxxx .......OK
Detaching volume: vol-yyyyyyyy ...OK
Creating volume from snapshot: snap-XXXXXXXX ...OK
Attaching volume: vol-zzzzzzzz ...OK
Starting instance: i-xxxxxxxx .....OK
Deleting volume: vol-yyyyyyyy ...OK