missingintegers

普通にやったらもちろんN**2で返ってくる。ので、おそらく普通のやり方らしいが、フラグを作ってそこを積算的に立てていく。と一周+最後の判別ですむ(サイズ(N)で住む)

def solution(A):
    A = sorted(A)
    B = [i for i in A if (0 < i) and (i <= len(A))]
    tester = [0] * len(A)
    for i in B:
        tester[i-1] +=1
    try:
        return tester.index(0) + 1
    except:
        return len(B)+1

Nature Remo

Nature RemoはNature Remo 3, Nature Remo miniなどのスマートリモコンを販売しています。
Nature Remo 3は1万円程度で「温度センサー」「湿度センサー」「照度センサー」「人感センサー」「Alexa」「Google Assistant」「Siri(ショートカット)」に対応しています。

API

Nature Remoの商品はAPI「Nature Remo Cloud API」が公開されており「センサーの情報」「リモコン制御」などがプログラミングから可能です。
この記事でAPIの使い方はわかりやすく出ているので、具体的なコードを見なくてもいい人はリンク先からみてください。
自分はPythonを使ってやっていきます。

https://developer.nature.global/

プログラミング

nature-remoというPythonのライブラリがあるためそちらを使うのが賢いですが、この後の「iOSのショートカット対応」への理解を深めるためにそちらのライブラリは使用しないでurllib,jsonのライブラリを使って進めていきます。
そちらの方法が見たい方は以下の記事を見てください。
https://qiita.com/morinokami/items/6eb2ac6bed48d2c7534b

温度, 湿度, 照度、人感センサー情報取得

人感センサーは最後に動きを検出した日付を返してくれるが少し特殊

##実行結果##
温度 : 23度
湿度 : 54%
明るさは : 57度
最後に検知したのは2020-11-15T10:32:01Zです。

from urllib.parse import urlencode
from urllib.request import urlopen, Request
from urllib.error import HTTPError
from json import loads

api_key = "jmgkkDkFTiMp55iRcVGCRiTU5OUg7FqaQfKYDOECUXI.LPVbfXhH9bqcJfzqsyZ-4" # APIアクセストークン

url = "https://api.nature.global/1/devices/"

headers = {
    "accept" :"application/json",
    "Authorization" :"Bearer " + api_key,
}

request = Request(url, headers=headers)

try:
  with urlopen(request) as response:
    data_byte = response.read()
    data= loads(data_byte)
except HTTPError as e:
    print(e)

device_info = data[0]["newest_events"]

print("温度 : "   + str(device_info["te"]["val"]) + "度")
print("湿度 : "   + str(device_info["hu"]["val"]) + "%")
print("明るさは : " + str(device_info["il"]["val"]) + "度")
print("最後に検知したのは" + str(device_info["mo"]["created_at"]) + "です。")

人感センサー日付 改良版

上のままでは人感センサーの日付がこのままではわかりずらいので、「〇〇分前」という処理をしたいと思います。
datetimeライブラリを使って日付の差分を出していきます。この処理をすることで2倍ほどのコード量になってしまっているので必要なければ上のを使ってください。
24時間以上検知がないと〇日前、1時間以上検知がないと〇時間前、60分内に収まっていれば〇分前になっています。

##実行結果##
温度 : 22.9度
湿度 : 54%
明るさは : 67度
人感センサー :2日前

from urllib.parse import urlencode
from urllib.request import urlopen, Request
from urllib.error import HTTPError
from json import loads
import datetime

api_key = "jmgkkDkFTiMp55iRcVGCRiTU5OUg7FqaQfKYDOECUXI.LPVbfXhH9bqcJfzqsyZ-4" # APIアクセストークン

url = "https://api.nature.global/1/devices/"

headers = {
    "accept" :"application/json",
    "Authorization" :"Bearer " + api_key,
}

request = Request(url, headers=headers)

try:
  with urlopen(request) as response:
    data_byte = response.read()
    data= loads(data_byte)
except HTTPError as e:
    print(e)

device_info = data[0]["newest_events"]

print("温度 : "   + str(device_info["te"]["val"]) + "度")
print("湿度 : "   + str(device_info["hu"]["val"]) + "%")
print("明るさは : " + str(device_info["il"]["val"]) + "度")

detect_date_str = str(device_info["mo"]["created_at"]).split("T")[0].split("-")
detect_time_str = str(device_info["mo"]["created_at"]).split("T")[1].split("Z")[0].split(":")

detect_date = [int(n) for n in detect_date_str]
detect_time = [int(n) for n in detect_time_str]

date = datetime.datetime(year=detect_date[0], month=detect_date[1], day=detect_date[2], hour=detect_time[0], minute=detect_time[1], second=detect_time[2])
now = datetime.datetime.now()
difference_time = now - date

if difference_time.days > 0:
  print("人感センサー :{}日前".format(difference_time.days))
elif difference_time.total_seconds > 3600:
  print("人感センサー : {}時間前".format(difference_time.total_seconds/3600))
else:
  print("人感センサー :{}分前".format(int(difference_time.total_seconds()/60)))

テレビ/ライト操作

• テレビとライトは同じコードで書けるのでまとめています。(applianceを変えることでテレビ/ライトの切り替えが可能です。)
• テレビ/ライトの操作だけでなく、テレビ/ライトのIDの取得に1度アクセスしているので、先にID(device_id)をプログラム内に入力してしまうことでアクセス数を減らすことが可能です。
• ニックネーム(nickname)を指定することで2台同じ家電があっても大丈夫なようにしています。1台であればその処理はいりません。

必要なもの

appliance:家電を選択(テレビ:TV, ライト: LIGHT)
nickname:操作したい家電のニックネーム(例:Two-Storied-TV)
api_access_key: APIアクセストークン(APIアクセストークンは「https://home.nature.global/」で発行したもの)

##出力結果(テレビ)##
操作するボタンを数字で選んでください。
0 : TV_power
1 : TV_source
2 : TV_schedule
3 : TV_mute
4 : TV_terrestrial
⌇
47 : TV_subtitle
48 : TV_exit
49 : TV_rewind_10sec
50 : TV_forward_30sec
入力:48
成功です。

##出力結果(ライト)##
操作するボタンを数字で選んでください。
0 : Light_on
1 : Light_off
2 : Light_all
3 : Light_night
4 : Light_bright
5 : Light_dark
入力:1
成功です。

from urllib.parse import urlencode
from urllib.request import urlopen, Request
from urllib.error import HTTPError
from json import loads

appliance = "TV" #操作する家電の種類["TV", "LIGHT"]
nickname = "Two-Storied-TV"   #テレビ/ライトのニックネーム
#nickname = "Atrium-light"   #ライトのニックネーム
api_access_key = "jmgkkDkFTiMp55iRcVGCRiTU5OUg7FqaQfKYDOECUXI.LPVbfXhH9bqcJfzqsyZ-4" # APIアクセストークン

url = "https://api.nature.global"

headers = {
    "Authorization" :"Bearer " + api_access_key,
    "accept" :"application/json",
    "Content-Type" :"application/x-www-form-urlencoded"
}

#全ての家電情報を取得
request = Request(url + "/1/appliances/", headers=headers) 

try:
  with urlopen(request) as response:
    data = response.read()
    devices = loads(data)
except HTTPError as e:
    print(e)


#デバイスID探索
device_id = "" 

for device in devices:
  if device["type"] == appliance and device["nickname"] == nickname:
    device_id = device["id"]
    buttons = device[appliance.lower()]["buttons"]


#各ボタンの表示
print("操作するボタンを数字で選んでください。")
num = 0
for button in buttons:
  if button["label"] != "":#空白のボタンがあるためif文で処理
    print(str(num) + " :\t"  + button["label"])
    num += 1

selected_button = int(input("入力:"))
signal = buttons[select_button]["name"]

#データ送信
request = Request(url + "/1/appliances/" + device_id + "/" + appliance.lower(), headers=headers)

data = {
    "button": signal
}

data = urlencode(data).encode("utf-8")

try:
  urlopen(request, data)
  print("成功です。")
except HTTPError as e:
    print(e)

エアコン操作

• エアコンの操作だけでなく、エアコンのIDの取得に1度アクセスしているので、先にID(AirCon_device_id )をプログラム内に入力してしまうことでアクセス数を減らすことが可能です。
• ニックネーム(nickname)を指定することで2台エアコンがあっても大丈夫なようにしています。1台であればその処理はいりません。

from urllib.parse import urlencode
from urllib.request import urlopen, Request
from urllib.error import HTTPError
from json import loads

temperature = list(range(16, 31)) #温度
operation_mode = ["cool", "warm", "dry", "blow", "auto"] #モード
air_volume = ["1", "2", "3", "4", "5", "auto"] #風量
air_direction = ["1", "2", "3", "4", "5", "auto"] #風向き
button = ["", "power-off"] #電源 空白は電源オン

nickname = "Two-Stroried-AirCon"   #エアコンのニックネーム
api_key = "jmgkkDkFTiMp55iRcVGCRiTU5OUg7FqaQfKYDOECUXI.LPVbfXhH9bqcJfzqsyZ-4" # APIアクセストークン

url = "https://api.nature.global"

headers = {
    "accept" :"application/json",
    "Authorization" :"Bearer " + api_key,
    "Content-Type" :"application/x-www-form-urlencoded"
}

req = Request(url + "/1/appliances/", headers=headers)

try:
  with urlopen(req) as response:
    data = response.read()
    devices = loads(data)
except HTTPError as e:
    print(e)

AirCon_device_id = "" 

for device in devices:
  if device["type"] == "AC" and device["nickname"] == nickname:
    AirCon_device_id = device["id"]
    buttons = device["aircon"]

request = Request(url + "/1/appliances/" + AirCon_device_id + "/aircon_settings", headers=headers)

data = {
    "temperature": temperature[10],
    "operation_mode": operation_mode[3],
    "air_volume": air_volume[4],
    "air_direction": air_direction[2],
    "button": button[1],
}

data = urlencode(data).encode("utf-8")

try:
  response = urlopen(request, data)
  print("成功です。")
except HTTPError as e:
    print(e)

政府CIOポータルのオープンデータのオープンデータ伝道師一覧のpptxの表をスクレイピング

wget https://cio.go.jp/sites/default/files/uploads/documents/opendata-dendoushi_ichiran.pptx -O ichiran.pptx
pip install python-pptx

import pptx
import pandas as pd

prs = pptx.Presentation("ichiran.pptx")

dfs = []

for page in prs.slides:

    data = [[cell.text for cell in row.cells] for row in page.shapes[1].table.rows]

    dfs.append(pd.DataFrame(data[1:], columns=data[0]))

df = pd.concat(dfs).set_index("No.")

df["所属団体等"] = df["所属団体等"].str.replace("\n", "", regex=True)

df1 = df.join(
    df["氏名"].str.split("\n", expand=True).rename(columns={0: "ふりがな", 1: "名前"})
).drop("氏名", axis=1)

df2 = df1.reindex(columns=["名前", "ふりがな", "主な活動エリア", "これまでの主な実績等", "所属団体等"])

df2.to_csv("ichiran.csv", encoding="utf_8_sig")

政府CIOポータルのオープンデータのオープンデータ伝道師一覧のpptxの表をスクレイピング

wget https://cio.go.jp/sites/default/files/uploads/documents/opendata-dendoushi_ichiran.pptx -O ichiran.pptx
pip install python-pptx

import pptx
import pandas as pd

prs = pptx.Presentation("ichiran.pptx")

dfs = []

for page in prs.slides:

    data = [[cell.text for cell in row.cells] for row in page.shapes[1].table.rows]

    dfs.append(pd.DataFrame(data[1:], columns=data[0]))

df = pd.concat(dfs).set_index("No.")

df["所属団体等"] = df["所属団体等"].str.replace("\n", "", regex=True)

df1 = df.join(
    df["氏名"].str.split("\n", expand=True).rename(columns={0: "ふりがな", 1: "名前"})
).drop("氏名", axis=1)

df2 = df1.reindex(columns=["名前", "ふりがな", "主な活動エリア", "これまでの主な実績等", "所属団体等"])

df2.to_csv("ichiran.csv", encoding="utf_8_sig")

はじめに

こんなことはありませんか?
toBの企業情報は取得しているけどCRMには法人番号はない。
あるのは、法人名と設立年度などの基本情報。
こんな時に他のリストのデータセットと結合する際に、何を結合keyにすればいいのか？
悩むと思います。

行政の定める法人番号さえあれば、あらゆる企業情報を結合し、1つの情報にすることができます。
例えば,以下のように、テーブルAとテーブルBを結合したい時がありうるとします。

テーブルA(企業情報)

会社名	設立年	都道府県
hoge	fuga	3

テーブルB(コンタクト情報)

会社名	見込み金額	受注状況
hoge	3000	初回接続

gbizinfoを使ってみる

sqlなどデータベース言語に触れることがある人ならわかると思います。
文字列同士の結合はDB負荷がかかるという事に。
法人番号さえあれば共通の結合Keyとして使用できます。
法人番号を取得する場合に、便利なのが、経産省のgbizinfoです
なんとこのサービスREST APIを備えているのです。
なので、とっても法人番号の取得が簡単です。
https://info.gbiz.go.jp/api/index.html

のちにheadersの情報でX-hojinInfo-api-tokenが必要になるので
あらかじめAPIの利用申請が必要です。

pythonで呼んでみる

サンプルデータ

以下のように、会社名と設立年度だけが入ったデータがあるとします。
このデータに法人番号を入れたいです。
リクエストの際のメソッドはGETです

会社名	設立年
楽天モバイル株式会社	2018
株式会社松屋フーズ	2018

リクエストしてみる

request.py

import json
import pandas as pd
import requests

class CorporateNumbers:
    def __init__(self):
        self.headers = {
            "Accept": "application/json",
            "X-hojinInfo-api-token": "###token###"
        }
        self.endpoint_url = 'https://info.gbiz.go.jp/hojin/v1/hojin'
    def _create_taeger_company_dataframe(self):
        df = pd.read_clipboard()
        return df
    def _get_corporate_number(self,df):
        #df = self._create_taeger_company_dataframe()
        name = df.name
        founded_year = df.founded
        results = []
        for name,founded in zip(name,founded_year):
            data = {
                'name':name,
                'founded':founded
            }
            res = requests.get(
                url = self.endpoint_url,
                headers = self.headers,
                params = data
            )
            json = res.json()['hojin-infos']
            results.extend(json)
            df = pd.io.json.json_normalize(results)
        return df
    def _merge_dataframe(self):
        df1 = self._create_taeger_company_dataframe()
        df2 = self._get_corporate_number(df = df1)
        df3 = pd.merge(df1,df2,on='name',how='left')
        return df3

結果

ウィジェット

Frame
概要：ウィジェットを格納する枠組みを作る。

Label
概要：文字列やイメージを表示する。

Message
概要：複数行の文字列を表示する。

Button
概要：ボタンを作る。

Radiobutton
概要：ラジオボタンを作る。

Listbox
概要：リストボックスを作る。

Scrollbar
概要：スクールバーを作る。

Scale
概要：スケールを作る。

Entry
概要：１行の文字列を入力と編集。

Menu
概要：メニューボタンを作る。

Menubutton
概要：メニューボタンを作る。

Bitmap
概要：ビットマップを作る。

Canvas
概要：キャンバスを作る。

Text
概要：テキストの入力と編集。

LabelFrame
概要：ラベル付きフレーム。

スピンボックス
概要：スピンボックスを作る。

PanedWindow
概要：ペインウィンドウを作る。

---

ウィジェットのオプション

foreground (fg)
文字や線を描くのに使用する色を指定

background (bg)
背景色の指定

text
ウィジェット内に表示されるテキスト

textvariable
テキストを格納するオブジェクトを指定

image
ウィジェット内に表示されるイメージ

bitmap
ウィジェット内に表示されるビットマップ

borderwidth (bd)
ウィジェットの枠の幅

relief
ウィジェットの枠のスタイル

height
ウィジェットの高さ

width
ウィジェットの幅

anchor
ウィジェットや表示されるデータの位置を指定

はじめに

「〜やないか」と「〜ちゃうやないか」が繰り返される点が、プログラミング言語の条件分岐に落とし込みやすいと閃いてしまったので（？）、書かずにいられませんでした。

コーンフレークネタを再現するならこんな感じでしょうか。

コード

class Breakfast:
    def is_sweet(self):
        return True

    def is_crunchy(self):
        return True

    def is_eaten_with_milk(self):
        return True

    def is_good_for_dinner(self):
        return True

    def is_eaten_by_monk(self):
        return True

    def exists_at_bottom_of_parfait(self):
        return True

    def is_chinese(self):
        return True

class Mother:
    def good_for_the_last_supper(self, breakfast):
        return True

    def knows_why_pentagon_is_big(self, breakfast):
        return False

    def used_to_adored(self, breakfast):
        return True

    def knows_who_to_thank(self, breakfast):
        return False

    def corn_flakes(self, breakfast):
        return False

class Father:
    def predict(self, breakfast, corn_flakes):
        if not corn_flakes:
            return "鯖の塩焼"

mother = Mother()
breakfast = Breakfast()
corn_flakes = False

if breakfast.is_sweet() and breakfast.is_crunchy() and breakfast.is_eaten_with_milk():
    corn_flakes = True

if mother.good_for_the_last_supper(breakfast):
    corn_flakes = False

if not mother.knows_why_pentagon_is_big(breakfast):
    corn_flakes = True

if breakfast.is_good_for_dinner():
    corn_flakes = False

if mother.used_to_adored(breakfast):
    corn_flakes = True

if breakfast.is_eaten_by_monk():
    corn_flakes = False

if breakfast.exists_at_bottom_of_parfait():
    corn_flakes = True

if breakfast.is_chinese():
    corn_flakes = False

if not mother.knows_who_to_thank(breakfast):
    corn_flakes = True

if not mother.corn_flakes(breakfast):
    corn_flakes = False

father = Father()
print(father.predict(breakfast, corn_flakes))

実行結果

$ python milkboy.py 
鯖の塩焼

目次

1. 概要
2. 前提条件
3. データアップロードの仕組み
4. データ連携の自動化
5. まとめ

概要

最近、Code for Africaという団体が運営しているopenAFRICAというアフリカのオープンデータのポータルサイトと、自身がルワンダの水道公社WASACと共同でメンテナンスしている水道ベクトルタイルデータの自動連携機能を、Pythonで実装した。

日本の自治体のオープンデータサイトでも多く使われていると思われるCKANというAPIを用いているので、自組織が持っているファイルなどのオープンデータをAPI経由で自動連携させたい場合に活用できると思うので、共有したいと思う。

前提条件

• CKAN APIを使っているオープンデータプラットフォームに自組織のアカウントを持っている
• Githubでオープンデータを管理している

この記事を通して、Githubに置いてあるオープンデータを更新したタイミングで、Github Actionを用いて、CKAN経由でプラットフォーム上のデータを自動連携させるようにします。

ちなみにルワンダの水道公社の水道ベクトルタイルのオープンデータのopenAFRICAのページは以下のリンクにあります。
https://open.africa/dataset/rw-water-vectortiles

また水道ベクトルタイルのGithubリポジトリは以下のリンクにあり、毎週水道公社内のサーバーからGithubに自動更新されます。
https://github.com/WASAC/vt

データアップロードの仕組み

リポジトリのダウンロードとインストール

pipenvがインストールされていない場合は、まず設定を行ってください。

git clone https://github.com/watergis/open-africa-uploader
cd open-africa-uploader
pipenv install
pipenv shell

CKAN APIを用いたファイルのアップロードの仕組み

まずリポジトリ内のOpenAfricaUploader.pyのソースコード全文を載せます。

import os
import ckanapi
import requests


class OpanAfricaUploader(object):
  def __init__(self, api_key):
    """Constructor

    Args:
        api_key (string): CKAN api key
    """
    self.data_portal = 'https://africaopendata.org'
    self.APIKEY = api_key
    self.ckan = ckanapi.RemoteCKAN(self.data_portal, apikey=self.APIKEY)

  def create_package(self, url, title):
    """create new package if it does not exist yet.

    Args:
        url (str): the url of package eg. https://open.africa/dataset/{package url}
        title (str): the title of package
    """
    package_name = url
    package_title = title
    try:
        print ('Creating "{package_title}" package'.format(**locals()))
        self.package = self.ckan.action.package_create(name=package_name,
                                            title=package_title,
                                            owner_org = 'water-and-sanitation-corporation-ltd-wasac')
    except (ckanapi.ValidationError) as e:
        if (e.error_dict['__type'] == 'Validation Error' and
          e.error_dict['name'] == ['That URL is already in use.']):
            print ('"{package_title}" package already exists'.format(**locals()))
            self.package = self.ckan.action.package_show(id=package_name)
        else:
            raise

  def resource_create(self, data, path, api="/api/action/resource_create"):
    """create new resource, or update existing resource

    Args:
        data (object): data for creating resource. data must contain package_id, name, format, description. If you overwrite existing resource, id also must be included.
        path (str): file path for uploading
        api (str, optional): API url for creating or updating. Defaults to "/api/action/resource_create". If you want to update, please specify url for "/api/action/resource_update"
    """
    self.api_url = self.data_portal + api
    print ('Creating "{}"'.format(data['name']))
    r = requests.post(self.api_url,
                      data=data,
                      headers={'Authorization': self.APIKEY},
                      files=[('upload', open(path, 'rb'))])

    if r.status_code != 200:
        print ('Error while creating resource: {0}'.format(r.content))
    else:
      print ('Uploaded "{}" successfully'.format(data['name']))

  def resource_update(self, data, path):
    """update existing resource

    Args:
        data (object): data for creating resource. data must contain id, package_id, name, format, description.
        path (str): file path for uploading
    """
    self.resource_create(data, path, "/api/action/resource_update")

  def upload_datasets(self, path, description):
    """upload datasets under the package

    Args:
        path (str): file path for uploading
        description (str): description for the dataset
    """
    filename = os.path.basename(path)
    extension = os.path.splitext(filename)[1][1:].lower()

    data = {
      'package_id': self.package['id'],
      'name': filename,
      'format': extension,
      'description': description
    }

    resources = self.package['resources']
    if len(resources) > 0:
      target_resource = None
      for resource in reversed(resources):
        if filename == resource['name']:
          target_resource = resource
          break

      if target_resource == None:
        self.resource_create(data, path)
      else:
        print ('Resource "{}" already exists, it will be overwritten'.format(target_resource['name']))
        data['id'] = target_resource['id']
        self.resource_update(data, path)
    else:
      self.resource_create(data, path)

OpenAfricaUploader.pyを呼び出してファイルをアップロードするソースコードは以下のような感じです。

import os
from OpenAfricaUploader import OpanAfricaUploader

uploader = OpanAfricaUploader(args.key)
uploader.create_package('rw-water-vectortiles','Vector Tiles for rural water supply systems in Rwanda')
uploader.upload_datasets(os.path.abspath('../data/rwss.mbtiles'), 'mbtiles format of Mapbox Vector Tiles which was created by tippecanoe.')

---

一個ずつ説明していきます。

コンストラクタ

このモジュールはあらかじめopenAFRICAにアップロードするためにベースとなるポータルサイトのURLをコンストラクタ内で設定しています。

self.data_portal = 'https://africaopendata.org'の部分のURLを自組織が利用しているCKAN APIのURLと置き換えてください。

  def __init__(self, api_key):
    """Constructor

    Args:
        api_key (string): CKAN api key
    """
    self.data_portal = 'https://africaopendata.org'
    self.APIKEY = api_key
    self.ckan = ckanapi.RemoteCKAN(self.data_portal, apikey=self.APIKEY)

コンストラクタの呼び出しは次のようになります。args.keyにご自身のアカウントのCKAN APIキーを指定してください。

uploader = OpanAfricaUploader(args.key)

パッケージの作成

package_createというAPIを利用してパッケージを作成します。その際引数には以下を指定します。

• name=ここで指定した文字列がパッケージのURLになります
• title=パッケージのタイトルです
• owner_org=CKANのポータル上の対象組織のIDです

作成に成功すると、パッケージ情報が戻り値として返って来ます。既にある場合はエラーになるため、例外処理の中で既存のパッケージ情報を取得する処理を書いています。

  def create_package(self, url, title):
    """create new package if it does not exist yet.

    Args:
        url (str): the url of package eg. https://open.africa/dataset/{package url}
        title (str): the title of package
    """
    package_name = url
    package_title = title
    try:
        print ('Creating "{package_title}" package'.format(**locals()))
        self.package = self.ckan.action.package_create(name=package_name,
                                            title=package_title,
                                            owner_org = 'water-and-sanitation-corporation-ltd-wasac')
    except (ckanapi.ValidationError) as e:
        if (e.error_dict['__type'] == 'Validation Error' and
          e.error_dict['name'] == ['That URL is already in use.']):
            print ('"{package_title}" package already exists'.format(**locals()))
            self.package = self.ckan.action.package_show(id=package_name)
        else:
            raise

この関数の呼び出し方は以下の通りになります

uploader.create_package('rw-water-vectortiles','Vector Tiles for rural water supply systems in Rwanda')

リソースの作成及び更新

リソースの作成はresource_createという関数で行っています。/api/action/resource_createというREST APIを使用して、アップロード対象のバイナリデータやファイル情報などもろもろを渡してあげれば良いです。

def resource_create(self, data, path, api="/api/action/resource_create"):
    self.api_url = self.data_portal + api
    print ('Creating "{}"'.format(data['name']))
    r = requests.post(self.api_url,
                      data=data,
                      headers={'Authorization': self.APIKEY},
                      files=[('upload', open(path, 'rb'))])

    if r.status_code != 200:
        print ('Error while creating resource: {0}'.format(r.content))
    else:
      print ('Uploaded "{}" successfully'.format(data['name']))

但し、resource_createだけだとリソースの追加だけしかできず、更新するたびにどんどん数が増えてしまいますので、/api/action/resource_updateというAPIを使って、既存のリソースがあったら更新してあげるようにします。

resource_updateの使い方は基本的にresource_createと同じで、違いはdataのなかにresource_idがあるかないかだけです

def resource_update(self, data, path):
    self.resource_create(data, path, "/api/action/resource_update")

resource_createとresource_updateをいい感じに組み合わせて、既存のリソースがあったら更新し、なかったら新規作成するという処理にしたのがupload_datasetsという関数です。

def upload_datasets(self, path, description):
    # ファイル名を拡張子と分離します
    filename = os.path.basename(path)
    extension = os.path.splitext(filename)[1][1:].lower()

    # リソース作成用のデータを作ります
    data = {
      'package_id': self.package['id'], #パッケージのID
      'name': filename,                 #更新対象のファイル名
      'format': extension,              #フォーマット（ここでは拡張子にしています）
      'description': description        #ファイルの説明
    }

    # 既にパッケージ内にリソースがあった場合はアップロード対象のファイル名と同じ名前のリソースがあるかないかチェックする。
    resources = self.package['resources']
    if len(resources) > 0:
      target_resource = None
      for resource in reversed(resources):
        if filename == resource['name']:
          target_resource = resource
          break

      if target_resource == None:
        # 同じ名前のリソースがない場合はresource_createを呼び出す
        self.resource_create(data, path)
      else:
        # リソースがある場合はdataにIDを設定してresource_updateを呼び出す
        print ('Resource "{}" already exists, it will be overwritten'.format(target_resource['name']))
        data['id'] = target_resource['id']
        self.resource_update(data, path)
    else:
      # リソースがない場合はresource_createを呼び出す
      self.resource_create(data, path)

upload_datasets関数の呼び出し方は以下のようになります。

 uploader.upload_datasets(os.path.abspath('../data/rwss.mbtiles'), 'mbtiles format of Mapbox Vector Tiles which was created by tippecanoe.')

アップロードのソースをコマンドラインから呼べるようにする

upload2openafrica.pyでコマンドラインから呼べるようにしています。

import os
import argparse
from OpenAfricaUploader import OpanAfricaUploader

def get_args():
  prog = "upload2openafrica.py"
  usage = "%(prog)s [options]"
  parser = argparse.ArgumentParser(prog=prog, usage=usage)
  parser.add_argument("--key", dest="key", help="Your CKAN api key", required=True)
  parser.add_argument("--pkg", dest="package", help="Target url of your package", required=True)
  parser.add_argument("--title", dest="title", help="Title of your package", required=True)
  parser.add_argument("--file", dest="file", help="Relative path of file which you would like to upload", required=True)
  parser.add_argument("--desc", dest="description", help="any description for your file", required=True)
  args = parser.parse_args()

  return args

if __name__ == "__main__":
  args = get_args()

  uploader = OpanAfricaUploader(args.key)
  uploader.create_package(args.package,args.title)
  uploader.upload_datasets(os.path.abspath(args.file), args.description)

実際に使う際は以下のような感じになります。upload_mbtiles.shというシェルスクリプトを作っています。環境変数にCKAN_API_KEYを設定するようにしてください。

#!/bin/bash

pipenv run python upload2openafrica.py \
  --key ${CKAN_API_KEY} \
  --pkg rw-water-vectortiles \
  --title "Vector Tiles for rural water supply systems in Rwanda" \
  --file ../data/rwss.mbtiles \
  --desc "mbtiles format of Mapbox Vector Tiles which was created by tippecanoe."

これでCKAN APIを使ってオープンデータをアップロードできるようになりました。

データ連携の自動化

でも毎回手動でCKANと連携するのは面倒なので、Github Actionで自動化します。ワークフローファイルは以下のような感じです。

name: openAFRICA upload

on:
  push:
    branches: [ master ]
    # ここではdataフォルダ以下が更新された場合にワークフローが走るようにしています
    paths:
      - "data/**"

jobs:
  build:

    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v2
    - name: Set up Python 3.8
      uses: actions/setup-python@v2
      with:
        python-version: 3.8
    - name: Install dependencies
      # ここでまずPipenvの初期設定をします
      run: |
        cd scripts
        pip install pipenv
        pipenv install
    - name: upload to openAFRICA
      # GithubのリポジトリのSettingsページのSecretsでCKAN_API_KEYという名前で登録しておけば次のようにして環境変数を使うことができます
      env:
        CKAN_API_KEY: ${{secrets.CKAN_API_KEY}}
      # その上で、シェルスクリプトを呼んであげるようにします
      run: |
        cd scripts
        ./upload_mbtiles.sh

これだけでGithubにファイルがアップロードされたらオープンデータプラットフォームに自動連携できるようになりました。次の画像はルワンダの水道公社のGithub Acitonが実行された際の画面です。

まとめ

CKAN APIは国内外の様々なオープンソースプラットフォームで使用されています。そのCKAN APIはPythonを用いることで比較的簡単にデータ連携を実装することが可能です。またオープンデータを管理しているのがGithub上なら、Github Actionを用いてさらに容易に自動連携することができます。

今回openAFRICA向けに作成したモジュールが国内外の他のCKANを使ったオープンデータの利活用に役に立つことを願っています。

目次

1. 概要
2. 前提条件
3. データアップロードの仕組み
4. データ連携の自動化
5. まとめ

概要

最近、Code for Africaという団体が運営しているopenAFRICAというアフリカのオープンデータのポータルサイトと、自身がルワンダの水道公社WASACと共同でメンテナンスしている水道ベクトルタイルデータの自動連携機能を、Pythonで実装した。

日本の自治体のオープンデータサイトでも多く使われていると思われるCKANというAPIを用いているので、自組織が持っているファイルなどのオープンデータをAPI経由で自動連携させたい場合に活用できると思うので、共有したいと思う。

前提条件

• CKAN APIを使っているオープンデータプラットフォームに自組織のアカウントを持っている
• Githubでオープンデータを管理している

この記事を通して、Githubに置いてあるオープンデータを更新したタイミングで、Github Actionを用いて、CKAN経由でプラットフォーム上のデータを自動連携させるようにします。

ちなみにルワンダの水道公社の水道ベクトルタイルのオープンデータのopenAFRICAのページは以下のリンクにあります。
https://open.africa/dataset/rw-water-vectortiles

また水道ベクトルタイルのGithubリポジトリは以下のリンクにあり、毎週水道公社内のサーバーからGithubに自動更新されます。
https://github.com/WASAC/vt

データアップロードの仕組み

リポジトリのダウンロードとインストール

pipenvがインストールされていない場合は、まず設定を行ってください。

git clone https://github.com/watergis/open-africa-uploader
cd open-africa-uploader
pipenv install
pipenv shell

CKAN APIを用いたファイルのアップロードの仕組み

まずリポジトリ内のOpenAfricaUploader.pyのソースコード全文を載せます。

import os
import ckanapi
import requests


class OpanAfricaUploader(object):
  def __init__(self, api_key):
    """Constructor

    Args:
        api_key (string): CKAN api key
    """
    self.data_portal = 'https://africaopendata.org'
    self.APIKEY = api_key
    self.ckan = ckanapi.RemoteCKAN(self.data_portal, apikey=self.APIKEY)

  def create_package(self, url, title):
    """create new package if it does not exist yet.

    Args:
        url (str): the url of package eg. https://open.africa/dataset/{package url}
        title (str): the title of package
    """
    package_name = url
    package_title = title
    try:
        print ('Creating "{package_title}" package'.format(**locals()))
        self.package = self.ckan.action.package_create(name=package_name,
                                            title=package_title,
                                            owner_org = 'water-and-sanitation-corporation-ltd-wasac')
    except (ckanapi.ValidationError) as e:
        if (e.error_dict['__type'] == 'Validation Error' and
          e.error_dict['name'] == ['That URL is already in use.']):
            print ('"{package_title}" package already exists'.format(**locals()))
            self.package = self.ckan.action.package_show(id=package_name)
        else:
            raise

  def resource_create(self, data, path, api="/api/action/resource_create"):
    """create new resource, or update existing resource

    Args:
        data (object): data for creating resource. data must contain package_id, name, format, description. If you overwrite existing resource, id also must be included.
        path (str): file path for uploading
        api (str, optional): API url for creating or updating. Defaults to "/api/action/resource_create". If you want to update, please specify url for "/api/action/resource_update"
    """
    self.api_url = self.data_portal + api
    print ('Creating "{}"'.format(data['name']))
    r = requests.post(self.api_url,
                      data=data,
                      headers={'Authorization': self.APIKEY},
                      files=[('upload', open(path, 'rb'))])

    if r.status_code != 200:
        print ('Error while creating resource: {0}'.format(r.content))
    else:
      print ('Uploaded "{}" successfully'.format(data['name']))

  def resource_update(self, data, path):
    """update existing resource

    Args:
        data (object): data for creating resource. data must contain id, package_id, name, format, description.
        path (str): file path for uploading
    """
    self.resource_create(data, path, "/api/action/resource_update")

  def upload_datasets(self, path, description):
    """upload datasets under the package

    Args:
        path (str): file path for uploading
        description (str): description for the dataset
    """
    filename = os.path.basename(path)
    extension = os.path.splitext(filename)[1][1:].lower()

    data = {
      'package_id': self.package['id'],
      'name': filename,
      'format': extension,
      'description': description
    }

    resources = self.package['resources']
    if len(resources) > 0:
      target_resource = None
      for resource in reversed(resources):
        if filename == resource['name']:
          target_resource = resource
          break

      if target_resource == None:
        self.resource_create(data, path)
      else:
        print ('Resource "{}" already exists, it will be overwritten'.format(target_resource['name']))
        data['id'] = target_resource['id']
        self.resource_update(data, path)
    else:
      self.resource_create(data, path)

OpenAfricaUploader.pyを呼び出してファイルをアップロードするソースコードは以下のような感じです。

import os
from OpenAfricaUploader import OpanAfricaUploader

uploader = OpanAfricaUploader(args.key)
uploader.create_package('rw-water-vectortiles','Vector Tiles for rural water supply systems in Rwanda')
uploader.upload_datasets(os.path.abspath('../data/rwss.mbtiles'), 'mbtiles format of Mapbox Vector Tiles which was created by tippecanoe.')

---

一個ずつ説明していきます。

コンストラクタ

このモジュールはあらかじめopenAFRICAにアップロードするためにベースとなるポータルサイトのURLをコンストラクタ内で設定しています。

self.data_portal = 'https://africaopendata.org'の部分のURLを自組織が利用しているCKAN APIのURLと置き換えてください。

  def __init__(self, api_key):
    """Constructor

    Args:
        api_key (string): CKAN api key
    """
    self.data_portal = 'https://africaopendata.org'
    self.APIKEY = api_key
    self.ckan = ckanapi.RemoteCKAN(self.data_portal, apikey=self.APIKEY)

コンストラクタの呼び出しは次のようになります。args.keyにご自身のアカウントのCKAN APIキーを指定してください。

uploader = OpanAfricaUploader(args.key)

パッケージの作成

package_createというAPIを利用してパッケージを作成します。その際引数には以下を指定します。

• name=ここで指定した文字列がパッケージのURLになります
• title=パッケージのタイトルです
• owner_org=CKANのポータル上の対象組織のIDです

作成に成功すると、パッケージ情報が戻り値として返って来ます。既にある場合はエラーになるため、例外処理の中で既存のパッケージ情報を取得する処理を書いています。

  def create_package(self, url, title):
    """create new package if it does not exist yet.

    Args:
        url (str): the url of package eg. https://open.africa/dataset/{package url}
        title (str): the title of package
    """
    package_name = url
    package_title = title
    try:
        print ('Creating "{package_title}" package'.format(**locals()))
        self.package = self.ckan.action.package_create(name=package_name,
                                            title=package_title,
                                            owner_org = 'water-and-sanitation-corporation-ltd-wasac')
    except (ckanapi.ValidationError) as e:
        if (e.error_dict['__type'] == 'Validation Error' and
          e.error_dict['name'] == ['That URL is already in use.']):
            print ('"{package_title}" package already exists'.format(**locals()))
            self.package = self.ckan.action.package_show(id=package_name)
        else:
            raise

この関数の呼び出し方は以下の通りになります

uploader.create_package('rw-water-vectortiles','Vector Tiles for rural water supply systems in Rwanda')

リソースの作成及び更新

リソースの作成はresource_createという関数で行っています。/api/action/resource_createというREST APIを使用して、アップロード対象のバイナリデータやファイル情報などもろもろを渡してあげれば良いです。

def resource_create(self, data, path, api="/api/action/resource_create"):
    self.api_url = self.data_portal + api
    print ('Creating "{}"'.format(data['name']))
    r = requests.post(self.api_url,
                      data=data,
                      headers={'Authorization': self.APIKEY},
                      files=[('upload', open(path, 'rb'))])

    if r.status_code != 200:
        print ('Error while creating resource: {0}'.format(r.content))
    else:
      print ('Uploaded "{}" successfully'.format(data['name']))

但し、resource_createだけだとリソースの追加だけしかできず、更新するたびにどんどん数が増えてしまいますので、/api/action/resource_updateというAPIを使って、既存のリソースがあったら更新してあげるようにします。

resource_updateの使い方は基本的にresource_createと同じで、違いはdataのなかにresource_idがあるかないかだけです

def resource_update(self, data, path):
    self.resource_create(data, path, "/api/action/resource_update")

resource_createとresource_updateをいい感じに組み合わせて、既存のリソースがあったら更新し、なかったら新規作成するという処理にしたのがupload_datasetsという関数です。

def upload_datasets(self, path, description):
    # ファイル名を拡張子と分離します
    filename = os.path.basename(path)
    extension = os.path.splitext(filename)[1][1:].lower()

    # リソース作成用のデータを作ります
    data = {
      'package_id': self.package['id'], #パッケージのID
      'name': filename,                 #更新対象のファイル名
      'format': extension,              #フォーマット（ここでは拡張子にしています）
      'description': description        #ファイルの説明
    }

    # 既にパッケージ内にリソースがあった場合はアップロード対象のファイル名と同じ名前のリソースがあるかないかチェックする。
    resources = self.package['resources']
    if len(resources) > 0:
      target_resource = None
      for resource in reversed(resources):
        if filename == resource['name']:
          target_resource = resource
          break

      if target_resource == None:
        # 同じ名前のリソースがない場合はresource_createを呼び出す
        self.resource_create(data, path)
      else:
        # リソースがある場合はdataにIDを設定してresource_updateを呼び出す
        print ('Resource "{}" already exists, it will be overwritten'.format(target_resource['name']))
        data['id'] = target_resource['id']
        self.resource_update(data, path)
    else:
      # リソースがない場合はresource_createを呼び出す
      self.resource_create(data, path)

upload_datasets関数の呼び出し方は以下のようになります。

 uploader.upload_datasets(os.path.abspath('../data/rwss.mbtiles'), 'mbtiles format of Mapbox Vector Tiles which was created by tippecanoe.')

アップロードのソースをコマンドラインから呼べるようにする

upload2openafrica.pyでコマンドラインから呼べるようにしています。

import os
import argparse
from OpenAfricaUploader import OpanAfricaUploader

def get_args():
  prog = "upload2openafrica.py"
  usage = "%(prog)s [options]"
  parser = argparse.ArgumentParser(prog=prog, usage=usage)
  parser.add_argument("--key", dest="key", help="Your CKAN api key", required=True)
  parser.add_argument("--pkg", dest="package", help="Target url of your package", required=True)
  parser.add_argument("--title", dest="title", help="Title of your package", required=True)
  parser.add_argument("--file", dest="file", help="Relative path of file which you would like to upload", required=True)
  parser.add_argument("--desc", dest="description", help="any description for your file", required=True)
  args = parser.parse_args()

  return args

if __name__ == "__main__":
  args = get_args()

  uploader = OpanAfricaUploader(args.key)
  uploader.create_package(args.package,args.title)
  uploader.upload_datasets(os.path.abspath(args.file), args.description)

実際に使う際は以下のような感じになります。upload_mbtiles.shというシェルスクリプトを作っています。環境変数にCKAN_API_KEYを設定するようにしてください。

#!/bin/bash

pipenv run python upload2openafrica.py \
  --key ${CKAN_API_KEY} \
  --pkg rw-water-vectortiles \
  --title "Vector Tiles for rural water supply systems in Rwanda" \
  --file ../data/rwss.mbtiles \
  --desc "mbtiles format of Mapbox Vector Tiles which was created by tippecanoe."

これでCKAN APIを使ってオープンデータをアップロードできるようになりました。

データ連携の自動化

でも毎回手動でCKANと連携するのは面倒なので、Github Actionで自動化します。ワークフローファイルは以下のような感じです。

name: openAFRICA upload

on:
  push:
    branches: [ master ]
    # ここではdataフォルダ以下が更新された場合にワークフローが走るようにしています
    paths:
      - "data/**"

jobs:
  build:

    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v2
    - name: Set up Python 3.8
      uses: actions/setup-python@v2
      with:
        python-version: 3.8
    - name: Install dependencies
      # ここでまずPipenvの初期設定をします
      run: |
        cd scripts
        pip install pipenv
        pipenv install
    - name: upload to openAFRICA
      # GithubのリポジトリのSettingsページのSecretsでCKAN_API_KEYという名前で登録しておけば次のようにして環境変数を使うことができます
      env:
        CKAN_API_KEY: ${{secrets.CKAN_API_KEY}}
      # その上で、シェルスクリプトを呼んであげるようにします
      run: |
        cd scripts
        ./upload_mbtiles.sh

これだけでGithubにファイルがアップロードされたらオープンデータプラットフォームに自動連携できるようになりました。次の画像はルワンダの水道公社のGithub Acitonが実行された際の画面です。

まとめ

CKAN APIは国内外の様々なオープンソースプラットフォームで使用されています。そのCKAN APIはPythonを用いることで比較的簡単にデータ連携を実装することが可能です。またオープンデータを管理しているのがGithub上なら、Github Actionを用いてさらに容易に自動連携することができます。

今回openAFRICA向けに作成したモジュールが国内外の他のCKANを使ったオープンデータの利活用に役に立つことを願っています。

問題

Anaconda で　install したはずの　パッケージが使用できず　下記のようなエラーが発生しました．

ModuleNotFoundError: No module named 'selenium'

今回はselenium をinstall したはずなのに　上記のようなエラーが出ました．

原因

anacondaプロンプトで python プログラム(****.py) を動作するときにpy.exeを使用していた．

対策

起動するときに　以下のように記述した．

前: py *********.py (エラーが出た)

後: python *********.py (動いた)

感想

pyrhon.exe と　py.exe 2種は違うものだと知った(今までは同じだと思っていた)
py.exe とはいったい何なのか，新たな疑問が生まれた
解決してみれば非常に単純な問題だが，時間がかかった
単純なミス過ぎて，ネットで検索しても出てこなかった．この記事が誰かの役に立とうれしいです．

解決までの道

自分に対するメモです
1.問題に気づく
2.py と　python パッケージをどこからimportしているか調べる

import sys,pprint
pprint.pprint(sys.path)

結果が違うことに気づく!!(計　2時間程度)

3.解決 (力尽きて作業進まず)

Tensorflow Serving を使い倒す

有望なディープラーニングのライブラリは Tensorflow と PyTorch で勢力が二分されている現状です。それぞれに強み弱みがあり、以下のような特徴があると思います。

• Tensorflow：Tensorflow Serving や Tensorflow Lite のような豊富な推論エンジン、Keras の便利な学習 API
• PyTorch：Define by Run による強力な学習、TorchVision による便利な画像処理

研究や学習では PyTorch が圧倒的になっていますが、推論器を動かすとなると Tensorflow のほうが有力な機能を提供していると思います。PyTorch は ONNX で推論することが可能ですが、モバイル向けや End-to-end なパイプラインサポートとなると、Tensorflow Lite や TFX 含めて Tensorflow が便利です。

本ブログでは Tensorflow Serving を用いた推論器とクライアントの作り方を説明します。Tensorflow Serving を動かすだけであれば多様な記事がありますが、本ブログではデータの入力から前処理、推論、後処理、出力まで、End-to-end で Tensorflow でカバーする方法を紹介します。

今回書いたコード：https://github.com/shibuiwilliam/e2e_tensorflow_serving

問題意識

ディープラーニングでモデルを学習した後、モデルは saved model や ONNX 形式で出力できても、前処理や出力が学習時の Python コードしかなく、推論へ移行するときに書き直すことになります。

学習も推論も Python で、Python コードをそのまま使い回せるなら良いですが（それでも間違うことが多々ありますが）、本番システムは Java や Golang、Node.js で Python を組み込む基盤や運用がないということがあります。Python 以外の言語で画像やテーブルデータの処理が Python ほど豊富であるとは限りませんし、Python で実行している前処理をそのまま動かすことができるとは限りません。

解決策のひとつは、機械学習の推論プロセスをサポートする推論器を作ることです。推論プロセスのすべてを Tensorflow の saved model に組み込んでしまい、Tensorflow Serving へ生データをリクエストすれば推論結果がレスポンスされる API を作れば、連携するバックエンドは REST クライアントや GRPC クライアントとして Tensorflow Serving にリクエストを送るだけで良くなります。
Tensorflow の Operator はニューラルネットワークだけでなく、画像のデコードやリサイズ、テーブルデータの One Hot 化等、機械学習に必須な処理が可能になっています。従来であれば Python の Pillow や Scikit-learn に依存していた処理が Tensorflow の計算グラフに組み込まれているため、推論のデータ入力から推論結果の出力まで、全工程を Tensorflow Serving でカバーすることができます。

本ブログでは Tensorflow Serving による画像分類、テキストの感情分析、テーブルデータの 2 値分類を使い、Tensorflow Serving の可能性を示していきたいと思います。

Tensorflow Serving

Tensorflow Serving は Tensorflow や Keras のモデルを推論器として稼働させるためのシステムです。Tensorflow の saved model を Web API（GRPC と REST API）として稼働させることができます。また単なる Web API だけでなく、バッチシステムとして動かすこともできます。複数バージョンのモデルを同一の Serving に組み込み、エンドポイントを分けることも可能です。Tensorflow Serving は Docker で起動させることが一般的です。

画像分類

ディープラーニングの重要な使い途の一つが画像処理です。今回はInception V3を使った画像分類を Tensorflow Serving で動かします。
画像分類のプロセスは以下になります。

1. 生データの画像ファイルを入力データとして受け取る。
2. 画像をデコードする。
3. 画像をリサイズして Inception V3 の入力 Shape である(299,299,3)に変換する。
4. Inception V3 で推論し、Softmax を得る。
5. 各ラベルに Softmax の確率をマッピングする。
6. 最も確率の高いラベルを出力する。

Inception V3 が担うのは常勤お 4 のみで、1,2,3,5,6 は前処理や後処理として周辺システムでカバーする必要があります。学習時は Python で Pillow や OpenCV、Numpy 等々を使って書きますが、推論時に同様のライブラリを使えるとは限りません。特に Python 以外の言語で構築する場合、OpenCV を使うことはできるかもしれませんが、他の Pillow や Numpy は他のライブラリで代替するか、自作する必要があります。
しかし Tensorflow であれば、1,2,3,5,6 も Tensor Operation に組み込み、推論の全行程をカバーすることができます。そのためには tf.function に前処理（1,2,3）と後処理（5,6）の Operation を記述します。
以下のdef serfing_fnがその Operation になります。Pillow や Numpy でも同様の処理を書くことがあると思いますが、記述量も複雑さも大差ない実装が可能です。

from typing import List
import tensorflow as tf
from tensorflow import keras

class InceptionV3Model(tf.keras.Model):
    def __init__(self, model: tf.keras.Model, labels: List[str]):
        super().__init__(self)
        self.model = model  # Inception V3 model
        self.labels = labels  # ImageNet labels in list

    @tf.function(
        input_signature=[tf.TensorSpec(shape=[None], dtype=tf.string, name="image")]
    )
    def serving_fn(self, input_img: str) -> tf.Tensor:
        def _base64_to_array(img):
            img = tf.io.decode_base64(img)
            img = tf.io.decode_jpeg(img)
            img = tf.image.convert_image_dtype(img, tf.float32)
            img = tf.image.resize(img, (299, 299))
            img = tf.reshape(img, (299, 299, 3))
            return img

        img = tf.map_fn(_base64_to_array, input_img, dtype=tf.float32)
        predictions = self.model(img)

        def _convert_to_label(candidates):
            max_prob = tf.math.reduce_max(candidates)
            idx = tf.where(tf.equal(candidates, max_prob))
            label = tf.squeeze(tf.gather(self.labels, idx))
            return label

        return tf.map_fn(_convert_to_label, predictions, dtype=tf.string)

    def save(self, export_path="./saved_model/inception_v3/0/"):
        signatures = {"serving_default": self.serving_fn}
        tf.keras.backend.set_learning_phase(0)
        tf.saved_model.save(self, export_path, signatures=signatures)

上記InceptionV3Modelクラスのインスタンスを saved model として保存し、Tensorflow Serving として起動することができます。起動した Tensorflow Serving は GRPC として 8500 ポート、REST API として 8501 ポートが開放されます。

docker run -t -d --rm \
-p 8501:8501 \
-p 8500:8500 \
--name inception_v3 \
-v $(pwd)/saved_model/inception_v3:/models/inception_v3 \
-e MODEL_NAME=inception_v3 \
tensorflow/serving:2.3.0

エンドポイントの定義は以下のようになっています。inputs以下が入力定義で、outputs以下が出力定義です。inputsではimageタグのデータを取ります。Shape が-1となっていますが、これは画像の base64 エンコードされたデータを入力とするためです。この時点で Tensorflow Serving への入力は(299,299,3)次元の配列ではなく、画像データそのものとなっています。

curl localhost:8501/v1/models/inception_v3/versions/0/metadata

$ curl localhost:8501/v1/models/inception_v3/versions/0/metadata
{
"model_spec":{
 "name": "inception_v3",
 "signature_name": "",
 "version": "0"
}
,
"metadata": {"signature_def": {
 "signature_def": {
  "serving_default": {
   "inputs": {
    "image": {
     "dtype": "DT_STRING",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_image:0"
    }
   },
   "outputs": {
    "output_0": {
     "dtype": "DT_STRING",
     "tensor_shape": {
      "dim": [],
      "unknown_rank": true
     },
     "name": "StatefulPartitionedCall:0"
    }
   },
   "method_name": "tensorflow/serving/predict"
  },
  "__saved_model_init_op": {
   "inputs": {},
   "outputs": {
    "__saved_model_init_op": {
     "dtype": "DT_INVALID",
     "tensor_shape": {
      "dim": [],
      "unknown_rank": true
     },
     "name": "NoOp"
    }
   },
   "method_name": ""
  }
 }
}
}
}

リクエストは以下のように実行することができます。GRPC と REST API の例を書いていますが、どちらも画像をバイナリデータとして読み込み、base64 エンコードして Tensorflow Serving のエンドポイントにリクエストします。クライアントは前処理することなく Tensorflow Serving にデータをリクエストします。
注意点は Tensorflow のtf.io.decode_base64がbase64.urlsafe_b64encodeされたデータでないとデコードできないという点です。

def read_image(image_file: str = "./a.jpg") -> bytes:
    with open(image_file, "rb") as f:
        raw_image = f.read()
    return raw_image

# GRPC
def request_grpc(
    image: bytes,
    model_spec_name: str = "inception_v3",
    signature_name: str = "serving_default",
    address: str = "localhost",
    port: int = 8500,
    timeout_second: int = 5,
) -> str:
    serving_address = f"{address}:{port}"
    channel = grpc.insecure_channel(serving_address)
    stub = prediction_service_pb2_grpc.PredictionServiceStub(channel)
    base64_image = base64.urlsafe_b64encode(image)

    request = predict_pb2.PredictRequest()
    request.model_spec.name = model_spec_name
    request.model_spec.signature_name = signature_name
    request.inputs["image"].CopyFrom(tf.make_tensor_proto([base64_image]))
    response = stub.Predict(request, timeout_second)

    prediction = response.outputs["output_0"].string_val[0].decode("utf-8")
    return prediction

#REST
def request_rest(
    image: bytes,
    model_spec_name: str = "inception_v3",
    signature_name: str = "serving_default",
    address: str = "localhost",
    port: int = 8501,
    timeout_second: int = 5,
):
    serving_address = f"http://{address}:{port}/v1/models/{model_spec_name}:predict"
    headers = {"Content-Type": "application/json"}
    base64_image = base64.urlsafe_b64encode(image).decode("ascii")
    request_dict = {"inputs": {"image": [base64_image]}}
    response = requests.post(
        serving_address,
        json.dumps(request_dict),
        headers=headers,
    )
    return dict(response.json())["outputs"][0]

推論結果は以下のようになります。

# GRPC
$ python request_inceptionv3.py -f GRPC
Siamese cat

# REST API
$ python request_inceptionv3.py -f REST
Siamese cat

テキストの感情分析

続いてテキスト分類です。テキスト処理も画像と同様で、入力、前処理、後処理、出力になる箇所を
Tensorflow でカバーします。

今回はサンプルデータとしてKaggle にある感情分析の NLP データを使用します。感情分析の英文データで、[anger, fear, joy, love, sadness, surprise]の 6 クラス分類となっています。

• anger: i felt anger when at the end of a telephone call
• fear: i pay attention it deepens into a feeling of being invaded and helpless
• joy: i am feeling totally relaxed and comfy
• love: i want each of you to feel my gentle embrace
• sadness: i realized my mistake and i m really feeling terrible and thinking that i shouldn't do that
• surprise: i feel shocked and sad at the fact that there are so many sick people

Tensorflow のテキスト処理で使えるライブラリは複数あります。

• Tensorflow Text
• Tensorflow Transform
• Tensorflow Keras Preprocessing
• Tensorflow Keras Layers Preprocessing

今回はTensorflow Keras Layers Preprocessingを使います。これを選んだのは API が使いやすいという理由です。
テキスト分類では以下の手順をたどります。前処理はテキストや目的次第ですが、今回は簡単のために tfidf を使います。

1. 生データのテキストを入力データとして受け取る。
2. テキストを前処理してベクターにする。
3. ニューラルネットワーク で推論し、Softmax を得る。
4. 各ラベルに Softmax の確率をマッピングする。
5. 最も確率の高いラベルを出力する。

Tensorflow Keras Layers PreprocessingではTextVectorizationで テキストデータの tfidf のベクター化が可能です。
以下は TextVectorization を使用したサンプルコードです。TextVectorization.adaptでテキストデータに対して変換マップを作ることができます。adapt した TextVectorization はtf.keras.layerとして Keras Model の 1 レイヤーに組み込むことができます。今回は入力レイヤーに使います。

def make_text_vectorizer(
    data: np.ndarray,
) -> tf.keras.layers.experimental.preprocessing.TextVectorization:
    text_vectorizer = tf.keras.layers.experimental.preprocessing.TextVectorization(
        output_mode="tf-idf", ngrams=2
    )
    text_vectorizer.adapt(data)
    return text_vectorizer

def define_model(
    text_vectorizer: tf.keras.layers.experimental.preprocessing.TextVectorization,
    optimizer: str = "adam",
    loss: str = "categorical_crossentropy",
    metrics: List[str] = ["accuracy"],
) -> tf.keras.Model:
    inputs = keras.Input(shape=(1,), dtype="string")
    x = text_vectorizer(inputs)
    x = layers.Dense(1)(x)
    x = layers.Dense(256, activation="relu")(x)
    x = layers.Dense(256, activation="relu")(x)
    outputs = layers.Dense(6, activation="softmax")(x)

    model = keras.Model(inputs, outputs)
    model.compile(optimizer=optimizer, loss=loss, metrics=metrics)
    return model

fit したモデルを使って saved model を作成します。今回は TextVectorization が入力データの前処理を担うため、後処理（手順 4,5）の分類部分のみ追加実装しています。

class TextModel(tf.keras.Model):
    def __init__(self, model: tf.keras.Model, labels: List[str]):
        super().__init__(self)
        self.model = model
        self.labels = labels

    @tf.function(
        input_signature=[tf.TensorSpec(shape=[None], dtype=tf.string, name="text")]
    )
    def serving_fn(self, text: str) -> tf.Tensor:
        predictions = self.model(text)

        def _convert_to_label(candidates):
            max_prob = tf.math.reduce_max(candidates)
            idx = tf.where(tf.equal(candidates, max_prob))
            label = tf.squeeze(tf.gather(self.labels, idx))
            return label

        return tf.map_fn(_convert_to_label, predictions, dtype=tf.string)

    def save(self, export_path="./saved_model/text/"):
        signatures = {"serving_default": self.serving_fn}
        tf.keras.backend.set_learning_phase(0)
        tf.saved_model.save(self, export_path, signatures=signatures)

保存した saved model で Tensorflow Serving を起動します。

docker run -t -d --rm \
-p 8501:8501 \
-p 8500:8500 \
--name text \
-v $(pwd)/saved_model/text:/models/text \
-e MODEL_NAME=text \
tensorflow/serving:2.3.0

Tensorflow Serving のメタデータは以下のとおりになっています。入力としてtextフィールドにテキストデータを入れてリクエストします。出力はoutout_0に推論結果のラベルがレスポンスされます。

curl localhost:8501/v1/models/text/versions/0/metadata

$ curl localhost:8501/v1/models/text/versions/0/metadata
{
"model_spec":{
 "name": "text",
 "signature_name": "",
 "version": "0"
}
,
"metadata": {"signature_def": {
 "signature_def": {
  "serving_default": {
   "inputs": {
    "text": {
     "dtype": "DT_STRING",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_text:0"
    }
   },
   "outputs": {
    "output_0": {
     "dtype": "DT_STRING",
     "tensor_shape": {
      "dim": [],
      "unknown_rank": true
     },
     "name": "StatefulPartitionedCall:0"
    }
   },
   "method_name": "tensorflow/serving/predict"
  },
  "__saved_model_init_op": {
   "inputs": {},
   "outputs": {
    "__saved_model_init_op": {
     "dtype": "DT_INVALID",
     "tensor_shape": {
      "dim": [],
      "unknown_rank": true
     },
     "name": "NoOp"
    }
   },
   "method_name": ""
  }
 }
}
}
}

今回も GRPC と REST のリクエスト例を示します。テキストデータをそのままリクエストに入れることができます。事前に前処理する必要はありません。

def read_text(text_file: str = "./text.txt") -> str:
    with open(text_file, "r") as f:
        text = f.read()
    return text

# GRPC
def request_grpc(
    text: str,
    model_spec_name: str = "text",
    signature_name: str = "serving_default",
    address: str = "localhost",
    port: int = 8500,
    timeout_second: int = 5,
) -> str:
    serving_address = f"{address}:{port}"
    channel = grpc.insecure_channel(serving_address)
    stub = prediction_service_pb2_grpc.PredictionServiceStub(channel)

    request = predict_pb2.PredictRequest()
    request.model_spec.name = model_spec_name
    request.model_spec.signature_name = signature_name
    request.inputs["text"].CopyFrom(tf.make_tensor_proto([text]))
    response = stub.Predict(request, timeout_second)

    prediction = response.outputs["output_0"].string_val[0].decode("utf-8")
    return prediction

# REST API
def request_rest(
    text: str,
    model_spec_name: str = "text",
    signature_name: str = "serving_default",
    address: str = "localhost",
    port: int = 8501,
    timeout_second: int = 5,
):
    serving_address = f"http://{address}:{port}/v1/models/{model_spec_name}:predict"
    headers = {"Content-Type": "application/json"}
    request_dict = {"inputs": {"text": [text]}}
    response = requests.post(
        serving_address,
        json.dumps(request_dict),
        headers=headers,
    )
    return dict(response.json())["outputs"][0]

テーブルデータ 2 値分類

最後にテーブルデータです。
モデル自体は Tensorflow のサンプルで公開されているClassify structured data with feature columnsを使用します。以下のようなデータ構成になっています。

テーブルデータの前処理は tensorflow.feature_columnで各種データの変換をサポートしています。tensorflow.feature_columnを使用した推論の流れは以下のようになります。

1. データを入力データとして受け取る。
2. データをカラムに応じて前処理する。
3. ニューラルネットワーク で推論し、Sigmoid を得る。
4. 陽性の確率を出力する。

前処理含めて学習時にカラムの前処理を定義することができます。使い方はシンプルで、データの特徴に応じて変換方法を適用するだけで使えます。

from tensorflow import feature_column
from tensorflow.keras import layers

feature_columns = []

for header in ["age", "trestbps", "chol", "thalach", "oldpeak", "slope", "ca"]:
    feature_columns.append(feature_column.numeric_column(header))

age = feature_column.numeric_column("age")
age_buckets = feature_column.bucketized_column(
    age, boundaries=[18, 25, 30, 35, 40, 45, 50, 55, 60, 65]
)
feature_columns.append(age_buckets)

thal = feature_column.categorical_column_with_vocabulary_list(
    "thal", ["fixed", "normal", "reversible"]
)
thal_one_hot = feature_column.indicator_column(thal)
feature_columns.append(thal_one_hot)

thal_embedding = feature_column.embedding_column(thal, dimension=8)
feature_columns.append(thal_embedding)

crossed_feature = feature_column.crossed_column(
    [age_buckets, thal], hash_bucket_size=1000
)
crossed_feature = feature_column.indicator_column(crossed_feature)
feature_columns.append(crossed_feature)

feature_columnで定義したデータの前処理をモデルの入力レイヤーとして活用することが可能です。

def define_model(
    feature_columns: List[Any],
    optimizer: str = "adam",
    loss: str = "binary_crossentropy",
    metrics: List[str] = ["accuracy"],
) -> tf.keras.Model:
    feature_layer = tf.keras.layers.DenseFeatures(feature_columns)
    model = tf.keras.Sequential(
        [
            feature_layer,
            layers.Dense(128, activation="relu"),
            layers.Dense(128, activation="relu"),
            layers.Dense(1, activation="sigmoid"),
        ]
    )

    model.compile(optimizer=optimizer, loss=loss, metrics=metrics)
    return model

これでモデルは完成です。モデルを保存して saved model とし、Tensorflow Serving として起動することができます。

docker run -t -d --rm \
-p 8501:8501 \
-p 8500:8500 \
--name table_data \
-v $(pwd)/saved_model/table_data:/models/table_data \
-e MODEL_NAME=table_data \
tensorflow/serving:2.3.0

Tensorflow Serving への入力データはカラム毎にフィールドを指定する形式になります。metadata を取ると以下のようになっています。長くなっていますが、各カラムで入力フィールドを定義しており、受け付けるデータ型や Shape が明示されています。

curl localhost:8501/v1/models/table_data/versions/0/metadata

$ curl localhost:8501/v1/models/table_data/versions/0/metadata
{
"model_spec":{
 "name": "table_data",
 "signature_name": "",
 "version": "0"
}
,
"metadata": {"signature_def": {
 "signature_def": {
  "serving_default": {
   "inputs": {
    "oldpeak": {
     "dtype": "DT_DOUBLE",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_oldpeak:0"
    },
    "restecg": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_restecg:0"
    },
    "trestbps": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_trestbps:0"
    },
    "slope": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_slope:0"
    },
    "sex": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_sex:0"
    },
    "ca": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_ca:0"
    },
    "exang": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_exang:0"
    },
    "fbs": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_fbs:0"
    },
    "chol": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_chol:0"
    },
    "thalach": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_thalach:0"
    },
    "thal": {
     "dtype": "DT_STRING",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_thal:0"
    },
    "cp": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_cp:0"
    },
    "age": {
     "dtype": "DT_INT64",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "serving_default_age:0"
    }
   },
   "outputs": {
    "output_1": {
     "dtype": "DT_FLOAT",
     "tensor_shape": {
      "dim": [
       {
        "size": "-1",
        "name": ""
       },
       {
        "size": "1",
        "name": ""
       }
      ],
      "unknown_rank": false
     },
     "name": "StatefulPartitionedCall_1:0"
    }
   },
   "method_name": "tensorflow/serving/predict"
  },
  "__saved_model_init_op": {
   "inputs": {},
   "outputs": {
    "__saved_model_init_op": {
     "dtype": "DT_INVALID",
     "tensor_shape": {
      "dim": [],
      "unknown_rank": true
     },
     "name": "NoOp"
    }
   },
   "method_name": ""
  }
 }
}
}
}

たとえば json でリクエストする場合、以下のようなデータでリクエストすることができます。

{
  "age": [[71]],
  "sex": [[0]],
  "cp": [[4]],
  "trestbps": [[112]],
  "chol": [[149]],
  "fbs": [[0]],
  "restecg": [[0]],
  "thalach": [[125]],
  "exang": [[0]],
  "oldpeak": [[1.6]],
  "slope": [[2]],
  "ca": [[0]],
  "thal": [["normal"]]
}

PythonでGRPC、RESTでリクエストする場合は以下になります。

def request_grpc(
    data: Dict[str, Any],
    model_spec_name: str = "inception_v3",
    signature_name: str = "serving_default",
    address: str = "localhost",
    port: int = 8500,
    timeout_second: int = 5,
) -> str:
    serving_address = f"{address}:{port}"
    channel = grpc.insecure_channel(serving_address)
    stub = prediction_service_pb2_grpc.PredictionServiceStub(channel)

    request = predict_pb2.PredictRequest()
    request.model_spec.name = model_spec_name
    request.model_spec.signature_name = signature_name

    age = np.array(data["age"], dtype=np.int64)
    sex = np.array(data["sex"], dtype=np.int64)
    cp = np.array(data["cp"], dtype=np.int64)
    trestbps = np.array(data["trestbps"], dtype=np.int64)
    chol = np.array(data["chol"], dtype=np.int64)
    fbs = np.array(data["fbs"], dtype=np.int64)
    restecg = np.array(data["restecg"], dtype=np.int64)
    thalach = np.array(data["thalach"], dtype=np.int64)
    exang = np.array(data["exang"], dtype=np.int64)
    oldpeak = np.array(data["oldpeak"], dtype=np.float64)
    slope = np.array(data["slope"], dtype=np.int64)
    ca = np.array(data["ca"], dtype=np.int64)
    thal = np.array(data["thal"], dtype=str)

    request.inputs["age"].CopyFrom(tf.make_tensor_proto(age))
    request.inputs["sex"].CopyFrom(tf.make_tensor_proto(sex))
    request.inputs["cp"].CopyFrom(tf.make_tensor_proto(cp))
    request.inputs["trestbps"].CopyFrom(tf.make_tensor_proto(trestbps))
    request.inputs["chol"].CopyFrom(tf.make_tensor_proto(chol))
    request.inputs["fbs"].CopyFrom(tf.make_tensor_proto(fbs))
    request.inputs["restecg"].CopyFrom(tf.make_tensor_proto(restecg))
    request.inputs["thalach"].CopyFrom(tf.make_tensor_proto(thalach))
    request.inputs["exang"].CopyFrom(tf.make_tensor_proto(exang))
    request.inputs["oldpeak"].CopyFrom(tf.make_tensor_proto(oldpeak))
    request.inputs["slope"].CopyFrom(tf.make_tensor_proto(slope))
    request.inputs["ca"].CopyFrom(tf.make_tensor_proto(ca))
    request.inputs["thal"].CopyFrom(tf.make_tensor_proto(thal))
    response = stub.Predict(request, timeout_second)

    prediction = response.outputs["output_1"].float_val[0]
    return prediction


def request_rest(
    data: Dict[str, Any],
    model_spec_name: str = "table_data",
    signature_name: str = "serving_default",
    address: str = "localhost",
    port: int = 8501,
    timeout_second: int = 5,
):
    serving_address = f"http://{address}:{port}/v1/models/{model_spec_name}:predict"
    headers = {"Content-Type": "application/json"}
    request_dict = {"inputs": data}
    response = requests.post(
        serving_address,
        json.dumps(request_dict),
        headers=headers,
    )
    return dict(response.json())["outputs"][0][0]

まとめ

Tensorflow の Operation を活用すれば、ディープラーニングのモデルだけでなく、データ入力から前処理、後処理までを計算グラフに組み込むことができます。学習から推論器へとシステムを移管する際、コードの書き換えが発生して非効率なシステム開発や設計になることがあります。End-to-end でテンソル演算に組み込んで Tensorflow Serving で推論することで、機械学習の学習時と同様の推論 API を構築することできます。