[python] 国土数値情報のAPIを使ってデータを取得してみた

この記事の目的

国土数値情報のAPIを使って、国土数値情報のZIPファイルをダウンロードしてそれを解凍するプログラムを作ります。ちなみにAPI触るの初めててです。

国土数値情報とは

国土数値情報とは、日本の地理情報に関する空間データのことです。国土交通省が公開してます。

1. 国土(水・土地)
   　　海岸線、森林地域、地価データ等
2. 政策区域
   　　行政区域、医療圏、避難施設データ等
3. 地域
   　　警察署、福祉施設データ等
4. 交通
   　　バスルート、駅別乗降客数
5. 各種統計

なんか色々使えそうなデータがあります。

ダウンロード方法

WEBサイトをポチポチクリックして直接ダウンロードする方法とAPIを使ってダウンロードする方法の二つがあります。
どちらの方法もZIPファイルをダウンロードすることになり、中身は同じみたいです。
都道府県をえらんでアンケートに答えて〜って毎回するのもめんどくさいので、今回は前述したとおりAPIを使ってダウンロードします。

国土数値情報API

国土数値情報APIの詳細については仕様書を確認してください。ZIPファイルの中身からなんちゃらも全部確認できます。
[appId=ksjapibeta1, dataformat,identifier]は必須パラメータみたいです。
APIのデモ通り、データ識別子のIDと、データの最新年度、都道府県コードを指定して、データをダウンロードしてみます。

プログラム

modules

from lxml import etree
import requests
import zipfile
import os.path
import os

APIを呼び出す

connect_api

dataformat = 1
identifier = "N03"
prefCodes = range(1, 48)
# prefCodes = [27]
fiscalyear = 2018

    def get_zip_url(self):
        call_url = 'http://nlftp.mlit.go.jp/ksj/api/1.0b/index.php/app/getKSJURL.xml?appId=ksjapibeta1&lang=J&dataformat=%s&identifier=%s&prefCode=%s&fiscalyear=%s'%(dataformat, identifier, prefCode, fiscalyear)
        print("Connect: ", end="")
        print(callurl)

        try:
            r = requests.get(callurl, timeout=5)
        except Exception as e:
            print("erro and retry")
            r = requests.get(callurl, timeout=5)

        root = etree.fromstring(r.content)

        # APIからZIP URLが出力できていないことを表示
        for iter in root.iter('STATUS'):
            if iter.text is "1":
                print("No such a data file /  URL is not wrong")
        return root

ダウンロードURLからZIPファイルをダウンロード

download_file

    def download_file(self, url):
        """
        URL を指定してカレントディレクトリにファイルをダウンロードする
        """

        print("Downloading: ", end="")
        print(url)
        filename = url.split('/')[-1]
        if os.path.isfile(filename):
            print(filename + " aleready exist")
            return filename

        r = requests.get(url, stream=True)
        with open(filename, 'wb') as f:
            for chunk in r.iter_content(chunk_size=1024):
                if chunk:
                    f.write(chunk)
                    f.flush()
            return filename

ZIPファイルを解凍

extract_zip

    def extract_zip(self, filename):
        """
        ファイル名を指定して zip ファイルをtmpディレクトリに展開する
        """
        # 作業ディレクトリ作成
        target_directory = './tmp'

        if os.path.isdir(target_directory) is False:
            os.mkdir('tmp')

        zfile = zipfile.ZipFile(filename)
        zfile.extractall(target_directory)