Python: PaSoRiでSuicaの履歴を読み出す・その後

以前、PaSoRiでSuicaの履歴を読み出すという記事を書いたけど、酷いバグを見つけた。物販で購入時の時刻を間違えるというもの。言い訳になるが、つい最近まで古いSuicaを使っていて、駅の自販機などでSuicaが使えず、テストしていなかったのだ。申し訳ない。

と云うわけで、修正したソースを公開しておく。バグだけ修正しても面白くないので、少し改良した。以前は残高しか出なかったが、使用した金額も併せて表示するようにした。また、以下に示す出力のように日時が先頭に来るように変更した。等々。

2008年xx月xx日 ○○駅 ××駅 540円 4160円 改札機 運賃支払(改札出場) 2008年xx月xx日 ○○駅 +2000円 6160円 券売機等 チャージ 2008年xx月xx日 xx時xx分xx秒 買物 120円 6040円 自販機 物販

以下、ソースコード。まだテストが足りないのでバグがあるかもしれない。見つけ次第、訂正する。

#!/usr/bin/env python # -*- coding: shift_jis -*- """ read_felica.py version 0.02 by nox Suicaの履歴を出力するプログラム. """ from ctypes import * POLLING_ANY = 0xffff POLLING_SUICA = 0x0003 POLLING_EDY = 0xfe00 SERVICE_SUICA = 0x090f SERVICE_EDY = 0x170f # 端末種 TERMINAL = {3: "精算機", 4: "携帯型端末", 5: "車載端末", 7: "券売機", 8: "券売機", 9: "入金機", 18: "券売機", 20: "券売機等", 21: "券売機等", 22: "改札機", 23: "簡易改札機", 24: "窓口端末", 25: "窓口端末", 26: "改札端末", 27: "携帯電話", 28: "乗継精算機", 29: "連絡改札機", 31: "簡易入金機", 70: "VIEW ALTTE", 72: "VIEW ALTTE", 199: "物販端末", 200: "自販機" } # 処理 PROCESS = { 1: "運賃支払(改札出場)", 2: "チャージ", 3: "券購(磁気券購入)", 4: "精算", 5: "精算 (入場精算)", 6: "窓出 (改札窓口処理)", 7: "新規 (新規発行)", 8: "控除 (窓口控除)", 13: "バス (PiTaPa系)", 15: "バス (IruCa系)", 17: "再発 (再発行処理)", 19: "支払 (新幹線利用)", 20: "入A (入場時オートチャージ)", 21: "出A (出場時オートチャージ)", 31: "入金 (バスチャージ)", 35: "券購 (バス路面電車企画券購入)", 70: "物販", 72: "特典 (特典チャージ)", 73: "入金 (レジ入金)", 74: "物販取消", 75: "入物 (入場物販)", 198: "物現 (現金併用物販)", 203: "入物 (入場現金併用物販)", 132: "精算 (他社精算)", 133: "精算 (他社入場精算)" } def read_station_code(fname): global STATION_CODE STATION_CODE = {} data = [l.strip().split(",") for l in file(fname) if l[0] in ("0", "1", "2")] for d in data: STATION_CODE[tuple(map(lambda x: int(x, 16), d[0:3]))] = (d[4], d[5]) def read_felica(): flib = cdll.felicalib flib.pasori_open.restype = c_void_p pasori = flib.pasori_open() flib.pasori_init(pasori) flib.felica_polling.restype = c_void_p felica = flib.felica_polling(pasori, POLLING_SUICA, 0, 0) # Suicaを読む. # 履歴の読み出し. data = [] d = create_string_buffer(16) i = 0 while flib.felica_read_without_encryption02(felica, SERVICE_SUICA, 0, i, d) == 0: data.append(string_at(pointer(d), 16)) i += 1 flib.pasori_close(pasori) return data def parse_data(d, prev=-1): term = ord(d[0]) # 端末種. proc = ord(d[1]) # 処理. date = map(ord, d[4:6]) # 日付. year = (date[0] >> 1) + 2000 month = ((date[0] & 1) << 3) + (date[1] >> 5) day = date[1] & (1<<5) - 1 in_line = ord(d[6]) # 入線区. in_sta = ord(d[7]) # 入駅順. out_line = ord(d[8]) # 出線区. out_sta = ord(d[9]) # 出駅順. balance = map(ord, d[10:12]) # 残高. num = map(ord, d[12:15]) # 連番. region = ord(d[15]) # リージョン. print "%4d年%02d月%02d日" % (year, month, day), if proc in (70, 73, 74, 75, 198, 203): # 物販. hour = in_line >> 3 min = ((in_line & 7) << 3) + (in_sta >> 5) sec = (in_sta & 0x1f) << 1 print "%02d時%02d分%02d秒" % (hour, min, sec), print "買物", elif proc in (13, 15, 31, 35): # バス. out_line = map(ord, d[6:8]) out_sta = map(ord, d[8:10]) print "バス", else: if region == 0: if in_line < 0x80: area = 0 # JR線. else: area = 1 # 関東公営・私鉄. else: area = 2 # 関西公営・私鉄. if in_line not in (0xc7, 0xc8, 0x05): if STATION_CODE.has_key((area, in_line, in_sta)): print "%s駅" % STATION_CODE[(area, in_line, in_sta)][1], else: print "不明", if STATION_CODE.has_key((area, out_line, out_sta)): if not (area == 0 and out_line == 0 and out_sta == 0): print "%s駅" % STATION_CODE[(area, out_line, out_sta)][1], else: print "不明", account = (balance[1] << 8) + balance[0] charge = prev - account if prev < 0: print "---円", elif charge > 0: print "%d円" % charge, elif charge < 0: print "%+d円" % -charge, print "%d円" % account, if TERMINAL.has_key(term): print TERMINAL[term], else: print "不明", if PROCESS.has_key(proc): print PROCESS[proc], else: print "不明", print return account def main(): read_station_code("StationCode.csv") data = read_felica() prev = -1 for d in data[::-1]: prev = parse_data(d, prev) if __name__ == "__main__": main()