物流2024年問題間近、配車アプリを作ることが出来るAPIを使ってみよう（後編）

はじめに

HEREでエンジニアをしておりますkekishidaと申します。前編では、HEREにおける物流DX向けソリューションの雄であるTour Planning APIのデモサイトについて紹介しました。

後編では、Tour Planning APIをベースにPythonとJupyter notebookを使用して、実際に配車アプリのプロトタイピングをしてみたいと思います。（以下は完成イメージです）

プロトタイピングに必要なもの

今回プロトタイピングを行うにあたって以下を準備する必要があります。

• Python
• Jupyter notebook
• HEREアカウント
  これらの詳細は拙記事である以下のリンクに載せておりますので、そちらをご参照ください。

シミュレーションするシナリオ

物流の最終拠点である営業所から最終届け先に荷物を配送するラストワンマイルのシナリオとして、以下の２種類のシナリオを順次実装します。

（１）前編で紹介したデモサイトのシナリオの再現 (シナリオ1)
（２）再荷積作業の実装 (シナリオ2)

場所

前編のデモサイトのシナリオと同様にエリアは京都とし、配送する場所についても、イメージがし易い有名観光スポットとしました。さらに、配送デポについては京都駅としました。こうすることで、配送順序の妥当性についてイメージがし易いかと思います。（京都に修学旅行に行かれた方は、そのルートを思い出してみてください。）

配送リスト

以下のような配送リストをCSVフォーマットで作成しました。（これから配達する荷物の宛先リストになります。）Start TimeとEnd Timeというフィールドを指定することで、配達の時間指定を制御できます。（ここでは特に制限を加えておりません。）Service Timeというのは配送先毎にかかるサービス時間（分）となります。この配送リストの時点では、配送順序、配送ルート、作業終了時間そして配車プランに関しては確定していません。これらを解決します。

シナリオ１

前提条件

1. 最短時間で荷物を配送する。
2. 大型のオブジェを配送するため、1車両あたり３つしか荷物を配送することができない。
3. 車両及び要員は５人確保されている。

解決したい課題

上記の条件において、必要な車両及び要員と最適な経路を求めることとなります。

配車シミュレーション by Jupyter

以下にjupyterと注釈してありますコードスニペットをそのままJupyter notebook/labのセルにペーストするだけでシミュレーションの実行が可能なように配慮をしておりますが、コードが冗長であること、個別環境に依存する部分がありますことについてはご了承ください。（あくまでもプロトタイピングという体です。）

パッケージのインポート

未インストールのパッケージについては適宜インストールをしてください。

jupyter

import json
import math
import datetime

import matplotlib.cm
import pandas as pd
import geopandas as gpd
import ipywidgets as widgets
import here_map_widget as H

from tqdm.auto import tqdm
from IPython.display import display, clear_output
from here_location_services import LS
from here_location_services.config import routing_config, tour_planning_config

ここで注意点がひとつあります。HEREのGitHubで公開されておりましたhere-location-services-pythonが残念ながらPublic ArchiveとなってしまいHEREでのメンテナンスが止まってしまいました。 そのため、このパッケージでは最新のTour Planning API version3に対応しておりません。

その代わりに、私のgithubリポジトリにフォークして手弁当でTour Planning API version3に対応したものを公開しました。（十分に検証されたものではありませんので、その点はご了承ください。）

次のインストールコマンドでこちらのパッケージをインストールしてください。

bash

pip install git+https://github.com/kekishida/here-location-services-python.git

APIキーとhere-location-service-pythonパッケージの初期化

APIKEYは自身のAPIキーに置き換えてください。

Jupyter

here_api_key = <APIKEY>
ls = LS(api_key=here_api_key)

plotlyの設定

後ほど、配車状況を可視化するために、plotlyをデータの可視化ツールとして使用します。

jupyter

pd.options.plotting.backend = 'plotly'

Basemapの定義

Hereの日本地図を表示するために、事前に以下のようにhere-map-widget-for-jupyterのTileLayer定義を行います。

jupyter

basemap = H.TileLayer(
    provider=H.OMV(
        path='v2/vectortiles/core/mc',
        platform=H.Platform(
            api_key=here_api_key,
            services_config={
                H.ServiceNames.omv: {
                    H.OMVUrl.scheme: 'https',
                    H.OMVUrl.host: 'vector.hereapi.com',
                    H.OMVUrl.path: '/v2/vectortiles/core/mc'
                }
            }
        ),
        style=H.Style(
            config='https://js.api.here.com/v3/3.1/styles/omv/oslo/japan/normal.day.yaml',
            base_url='https://js.api.here.com/v3/3.1/styles/omv/oslo/japan/'
        )
    ),
    style={'max': 22}
)

here-map-widget-for-jupyterの詳細は以下のリンクをご参照下さい。

Global関数の定義

Tour Planning APIに入力する日付のフォーマットを整える関数を定義します。

jupyter

def format_time(time):
    return pd.to_datetime(time) \
             .tz_localize('Asia/Tokyo') \
             .strftime('%Y-%m-%dT%H:%M:%S%z') \
             .replace('+0900', '+09:00')

デポの登録

配送デポを京都駅の緯度経度を設定し、here-map-widget-for-jupyterのMarkerオブジェクトを作成します。

jupyter

depot_lat, depot_lng = 34.98649124332442, 135.75892319474372
depot_marker = H.Marker(lat=depot_lat, lng=depot_lng)

配送リストCSVファイル

先ほど紹介した配送リストをCSV化します。京都.csvというファイル名としました。

CSVファイル

ID,Name,Address,Start Time,End Time,Service Time,Skills,Demand,Job Type
1,東寺,京都府京都市南区九条町１,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送
2,清水寺,京都府京都市東山区清水１丁目２９４,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送
3,金閣寺,京都府京都市北区金閣寺町１,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送
4,銀閣寺,京都府京都市左京区銀閣寺町２,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送
5,渡月橋,京都府京都市右京区嵯峨天龍寺芒ノ馬場町１－５,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送
6,映画村,京都府京都市右京区太秦東蜂岡町１０,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送
7,京都御所,京都府京都市上京区京都御苑３,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送
8,伏見稲荷,京都府京都市伏見区深草藪之内町６８,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送
9,南禅寺,京都府京都市左京区南禅寺福地町８６,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送
10,二条城,京都府京都市中京区二条城町５４１,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送
11,八坂神社,京都府京都市東山区祇園町北側６２５,10.01.22 8:00,10.01.22 20:00,0:00,,1 ,配送

配送リストCSVファイルの読み込み

jupyter

order_df = pd.read_csv('京都.csv')
order_df.ID = order_df.ID.astype(str)
order_df

住所を緯度経度情報に変換

here-location-services-pythonパッケージのgeocodeメソッドを利用して、住所を緯度経度情報に変換し、新たにorders_dfというデータフレームを作成します。同時に時刻フォーマットをTour Planning API向けに変換した列も追加しています。

jupyter

from tqdm import tqdm
from here_location_services import LS

ls = LS(api_key=here_api_key)
addrList = order_df['Address'].tolist()

latlngList = []
for m in tqdm(addrList):
    gc_response = ls.geocode(query=m)
    data = gc_response.to_geojson()
    if len(data["features"]) != 0:
        latlngList.append(data["features"][0]["properties"]["position"])
    else:
        latlngList.append({'lat': 0.0, 'lng': 0.0})
latList = []
lngList = []
for o in latlngList:
    latList.append(o['lat'])
    lngList.append(o['lng'])
idList = []
for p in range(len(addrList)):
    p = p + 1
    idList.append(str(p))

orders_df = order_df.copy()
orders_df['Latitude'] = latList
orders_df['Longitude'] = lngList
orders_df['Service time (s)'] = pd.to_timedelta('0:' + orders_df['Service Time']).dt.total_seconds()
orders_df['Start time (formatted)'] = orders_df['Start Time'].apply(format_time)
orders_df['End time (formatted)'] = orders_df['End Time'].apply(format_time)
orders_df

GeoPandas DataFrameに変換

GeoPandas DataFrameに変換することで、here-map-wdget-for-jupyterにそのままLayerとして重ねるようにします。

jupyter

order_gdf = gpd.GeoDataFrame(orders_df.copy(), geometry=gpd.points_from_xy(x=orders_df.Longitude, y=orders_df.Latitude))
order_gdf

配送場所とデポの場所をプロット

jupyter

order_layer = H.GeoData(geo_dataframe=order_gdf, show_bubble=True, point_style=dict(radius=5))
center = order_gdf.total_bounds.reshape(2, 2).mean(axis=0)[::-1].tolist()
m = H.Map(api_key=here_api_key, center=center, zoom=12, basemap=basemap)
m.add_object(depot_marker)
m.add_layer(order_layer)
m.add_control(H.FullscreenControl())
output = widgets.Output()
layout = widgets.VBox([
    m,
    output,
])
layout

以下の様に配送場所とデポの場所が地図上にプロットされました。

Tour Planning APIについて

いよいよTour Planning APIを呼び出す部分ですが、入力情報はおおまかに言うと、大きく分けて３つの要素があります。

• 配送リスト
• デポで使用する車両タイプのリスト
• デポで使用する車両プロファイルのリスト

これらの情報はJSON形式でHTTP POSTとして送信されることで、Tour Planning APIの結果として最適な配車プランとして返送されます。詳細については、以下のAPIリファレンスはを参照ください。

このJSONフォーマットを自身でプログラマブルに変更するロジックを作ってもよいのですが、今回は先ほど紹介しましたhere-location-services-python(フォーク版）を利用してこの入力情報を作成します。

Tour Planning API 諸条件の定義

前回のデモサイトで設定したデータを定義します。

jupyter

vehicle_capacity = 3
vehicle_amount = 5
cost_distance = 50
cost_time = 700
cost_fix = 2200
num_of_vehicleTypes = 1

上記の情報は、デモサイトの以下のパラメータに該当します。

• vehicle_capacity (CAPACITY PER VEHICLE)
• vehicle_amout (NO OF VEHICLES)
• cost_diatance (DISTANCE/KM)
• cost_time (DURATION/HR)
• cost_fix (FIXED/TOUR)

配送リストの情報生成

ここでは、先ほど作成したPandas DataFrameを再帰的に読み込んで、JSONフォーマット化します。以下の様に、here-location-services-python(フォーク版）のtour_palnning_configというパッケージを使用します。

jupyter

def func(row):
    return tour_planning_config.Job(
        id=str(row.ID),
        deliveries=[
                tour_planning_config.Task(
                    places=
                    [
                        tour_planning_config.JobPlaces(
                            duration = 3 * 60,
                            lat = row.Latitude,
                            lng = row.Longitude,
                            times=[
                                [row['Start time (formatted)'], row['End time (formatted)']],
                            ],
                        ),
                    ],
                    demand=[row.Demand]
                )
        ],
        skills=[row.Skills or 'any'],
    )
jobs = orders_df.fillna('').apply(func, axis=1).tolist()
plan = tour_planning_config.Plan(jobs=jobs)

デポで使用する車両タイプリストの情報生成

ここでは車両タイプの情報をJSONフォーマット化します。すなわち、コスト、最大積載量、車両数、配送可能時間などになります。今回の例では、車両タイプが１種類しかないため、１種類の車両タイプが作成されます。

jupyter

start_time = "10.1.2022 07:00"
end_time = "10.1.2022 22:00"

def func(i):
    return tour_planning_config.VehicleType(
        id=str(i),
        profile_name='car',
        costs_fixed=cost_fix,
        costs_distance=cost_distance,
        costs_time=cost_time,
        capacity=[vehicle_capacity],
        amount=vehicle_amount,
        shifts = [tour_planning_config.VehicleShift(
            start=dict(time=format_time(start_time), location=dict(lat=depot_lat, lng=depot_lng)),
            end=dict(time=format_time(end_time), location=dict(lat=depot_lat, lng=depot_lng)),
        )],
        limits=dict(maxDistance=3000 * 1000),
        skills=['any'],
    )

vehicle_types = [func(i) for i in range(num_of_vehicleTypes)]

デポで使用する車両プロファイルリストの情報生成

ここでは車両プロファイルの情報をJSONフォーマット化します。すわなち普通車と指定しています。上記の車両タイプで定義されているprofile_nameとこちらのnameがリンクされることで、上記の車両タイプは普通車として適用されることになります。

jupyter

fleet = tour_planning_config.Fleet(
    vehicle_types=vehicle_types,
    vehicle_profiles=[
        tour_planning_config.VehicleProfile(
            name='car',
            type='car',
        ),
    ],
)

configの設定

Tour Planning API自身の設定(タイムアウトなど）をします。

jupyter

configuration=tour_planning_config.Configuration(termination=dict(maxTime=200,stagnationTime=300))

Tour Planing APIの呼び出し

最後に、上記で設定した情報を引数にsoleve_tour_planningメソッドを使用してTour Planning APIを呼び出します。

jupyter

%time res = ls.solve_tour_planning(configuration=configuration, fleet=fleet, plan=plan, is_async=False)
res

このサンプルでは、配送リストの数が少ないため、is_async=Falseにしています。is_async=Trueにすることで非同期モードとし多量の配送リストをカバーすることができます。執筆時点(2023/12/14)では同時に6000件の配送リストの対応が可能となります。詳しくはDevelopper guideをご参照ください。

Tour Planning API結果の分析

PandasとPlotlyを使用して配車結果の分析を行います。
まず、APIのレスポンスJSONを読み込みます。

jupyter

solution = json.loads(res.as_json_string())

それでは、配車結果を確認します。このsolution['tours']をDataFrameとすることで一覧とします。

jupyter

tours_df = pd.json_normalize(solution['tours'])
tours_df

デモサイトの結果と同様に４台に割り当てられています。

割り当てられた車両の各工程を一覧します。少し煩雑なコードですが、以下の様に実現します。

jupyter

vehicleIds = tours_df['vehicleId'].tolist()
jobList = []
for shift in solution['tours']:
    for activity in shift['stops']:
        count =1
        length = len(activity['activities'])
        for jobs in activity['activities']:
            jobs['vehicleId'] = shift['vehicleId']
            if count == length:
                jobs['load'] = activity['load']
            jobList.append(jobs)
            count = count + 1
pd.set_option('display.max_columns', None)
pd.options.display.max_colwidth=100
tours_detail_df = pd.DataFrame(jobList)
for m in vehicleIds:
    display(tours_detail_df[tours_detail_df['vehicleId']==m])

ひとつひとつのテーブルが各車両の工程となります。右端のload列の部分が残荷物数の状況です。ゴールに近づくにつれてその数が減少していく様がわかります。

次にPlotlyを使用して配車具合を視覚的に確認します。

jupyter

columns = ['statistic.cost', 'statistic.distance', 'statistic.duration', 'statistic.times.driving', 'statistic.times.serving']
tours_df[columns].T.plot.bar(barmode='group')

おおよそ当分に割り当てられていることが確認できます。(Distanceのみ１台目が突出していますが、後術する地図を見れば一目瞭然かと思います。）

Tour Planning API結果を地図上に可視化

まずは地図上にルートを描く上でデコレーションする関数を定義し、配送ポイントをカスタムアイコンのMarkerオブジェクトリストとします。

jupyter

colormap = matplotlib.cm.tab10


def get_icon(text, color, size=18):
    bitmap = f'''
        <svg xmlns="http://www.w3.org/2000/svg" width="{size}" height="{size}">
            <circle cx="{size // 2}" cy="{size // 2}" r="{size // 2}" fill="rgba{color}"/>
            <text x="50%" y="50%" font-family="FiraGO" font-size="10pt" text-anchor="middle" dominant-baseline="central" fill="white">
                {text}
            </text>
        </svg>
    '''.strip()
    return H.Icon(bitmap=bitmap, width=size, height=size)


def func(row):
    stops = row['stops'][1:-1]
    color = colormap(row.name, bytes=True)
    markers = [H.Marker(lat=stop['location']['lat'], lng=stop['location']['lng'], icon=get_icon(i + 1, color)) for i, stop in enumerate(stops)]
    return markers


marker_lists = tours_df.apply(func, axis=1)
stop_markers = [marker for marker_list in marker_lists for marker in marker_list]

Tour Planning APIが導き出した配車情報及び各配送ポイントの順序情報に基づいて、Here Routing APIを使用して実際の経路を導き出します。

jupyter

def func(row):
    name = str(row.name)
    color = colormap(row.name, bytes=True)
    stops = row['stops']
    locations = [(stop['location']['lat'], stop['location']['lng']) for stop in stops]
    via = [routing_config.Via(lat=lat, lng=lng) for lat, lng in locations[1:-1]]
    res = ls.car_route(
        origin=locations[0],
        destination=locations[-1],
        via=via,
        return_results=[
            routing_config.ROUTING_RETURN.polyline,
            routing_config.ROUTING_RETURN.elevation
        ],
    )
    layer = H.GeoJSON(data=res.to_geojson(), style=dict(lineWidth=3, strokeColor=f'rgba{color}'), name=name)
    return layer


tqdm.pandas()
route_layers = tours_df.progress_apply(func, axis=1).tolist()

最後に、先ほど作成したMarkerとHERE Routing API結果のLayerとして重ね合わせることで、Tour Planning APIの結果を地図上に可視化します。

jupyter

m = H.Map(api_key=here_api_key, center=center, zoom=12, basemap=basemap)
m.add_layers(route_layers)
m.add_objects(stop_markers)
m.add_object(depot_marker)
m.add_control(H.FullscreenControl())


def on_change(change):
    value = change['new']
    m.objects = []
    m.layers = []
    if value == 'All':
        m.add_object(depot_marker)
        m.add_layers(route_layers)
        m.add_objects(stop_markers)
    else:
        i = int(value)
        m.add_object(depot_marker)
        m.add_layer(route_layers[i])
        m.add_objects(marker_lists[i])


options = ['All'] + tours_df.index.tolist()
dropdown = widgets.Dropdown(
    options=options,
    description='Tour:',
)
dropdown.observe(on_change, 'value')
layout = widgets.VBox([
    dropdown,
    m
])
layout

以上で、デモサイトと同じレベルのことが実現できる様になりました。
次に少し複雑なユースケース、再荷積をプロトタイピングしてみたいと思います。

シナリオ２

前提条件

1. 最短時間で荷物を配送する。
2. 大型のオブジェを配送するため、1車両あたり２つしか荷物を配送することができない。
3. 車両及び要員は５人確保されている。
4. 再荷積を許容する。

解決したい課題

上記の条件において、必要な車両及び要員と最適な経路を求めることとなります。

配車シミュレーション by Jupyter

先ほどのコードにほんの少し手を加えるだけで実現可能です。

Tour Planning API 諸条件の定義

前提条件に合わせて最大積載量を２とします。

最大積載量を２にした理由は再荷積が発生し易い様に荷積量を調整しただけです。

jupyter

- vehicle_capacity = 3
+ vehicle_capacity = 2
vehicle_amount = 5
cost_distance = 50
cost_time = 700
cost_fix = 2200
num_of_vehicleTypes = 1

デポで使用する車両タイプリストの情報生成

車両タイプの情報に、再荷積が可能な時間のシフトを設定します。ここでは配達可能時間を7:00-22:00に対して再荷積可能時間を12:00-16:00と定義しました。さらにshftsパラメータの中にreloads情報を以下の様に定義しています。すなわち、再荷積可能時間、デポの場所などです。

jupyter

start_time = "10.1.2022 07:00"
end_time = "10.1.2022 22:00"
+reload_start_time = "10.1.2022 12:00"
+reload_end_time = "10.1.2022 16:00"

def func(i):
    return tour_planning_config.VehicleType(
        id=str(i),
        profile_name='car',
        costs_fixed=cost_fix,
        costs_distance=cost_distance/1000,
        costs_time=cost_time/3600,
        capacity=[vehicle_capacity],
        amount=vehicle_amount,
        shifts = [tour_planning_config.VehicleShift(
            start=dict(time=format_time(start_time), location=dict(lat=depot_lat, lng=depot_lng)),
            end=dict(time=format_time(end_time), location=dict(lat=depot_lat, lng=depot_lng)),
+            reloads=[dict(
+                times=[
+                    [format_time(reload_start_time), format_time(reload_end_time)],
+                ],
+               location=dict(lat=depot_lat, lng=depot_lng),
+                duration=10 * 60,
+            )],
        )],
        limits=dict(maxDistance=3000 * 1000),
        skills=['any'],
    )

vehicle_types = [func(i) for i in range(num_of_vehicleTypes)]```

以上です！
それでは結果発表！

Tour Planning API結果の分析

３台に配車されました。

それぞれの工程は以下の通りです。それぞれデポに一度戻って再荷積した上で配送作業を再開している状況を伺うことができます。type列を確認するとそれぞれのイベントが確認できます。
(departure->delivery->reload->delivery->arrival)

Tour Planning API結果を地図上に可視化

全体のイメージだけですとわかりにくいので、ひとつの車両に絞って表示してみます。

金閣寺、二条城に配達し、一度京都駅に戻り再荷積をし、八坂神社、清水寺に再配達している状況を確認することができます。

おわりに

いかがでしたでしょうか？　後編では、Tour Planning APIをベースにPython, Jupyter notebookを利用してデモサイトと同レベルあるいはそれ以上の機能をプロトタイピングしてみました。Tour Planning APIの機能はここで紹介したもの以外にも、さらにきめ細やかなシナリオに対応することができます。また、スケーラビリティも大きな特徴です。最後にTour Planning APIのスケーラビリティのイメージについて以下に紹介いたします。(1400強の配送サンプルです。）

是非こちらの記事を参考にして、Tour Planning APIを体験して頂ければ幸いです。