神戸のデータ活用塾!KDL Data Blog

KDLが誇るデータ活用のプロフェッショナル達が書き連ねるブログです。

【オフィス人数の遠隔取得②】定期自動取得してkintoneに反映してみた編

こんにちは! DataIntelligenceチームの福岡です。

この連載では、以下の要領でオフィスにいる人数の遠隔自動取得に挑戦します!

  1. オフィスの様子を設置したカメラで撮影&ダウンロード
  2. 画像から人だけを検出。人数をカウント
  3. 上記を自動で定期的に実行する
  4. 他サービスと連携して結果をリアルタイムに反映する

前回は、クラウドカメラサービス『ソラカメ』と物体検出モデルYOLOv8を使って、1と2を実装しました。

kdl-di.hatenablog.com

連載2回目となる今回は、3, 4の実装を目指します。

弊社では業務にkintoneを使用しています。

そのため今回は、ソラカメで定期取得した画像から写っている人数をカウントし、kintoneに反映していきます!

実装の流れ

次の手順で実装します。

解説の都合で、他サービス連携を定期実行より先に実装しています。

  1. 画像を保存する機能を実装
  2. 人数カウントする機能を実装
  3. kintoneでアプリを作成し、kintone REST APIでレコードを登録できるようにする
  4. 一連の処理を定期的に実行する

1. 画像を保存する機能を実装

実装を楽にするため、カレントディレクトリにlibrariesファイルを作成し、実装した機能を格納することにします。

  • メインフォルダ
    • ultralytics
    • libraries
      • SoracomAPI.py
      • TimeConberter.py

前回実装した、現在時刻を取得しUNIXタイムに変換するクラスと、指定した時刻の画像を取得するためのクラスも格納しています。

※ ultralyticsには予測結果が保存されます。詳細は前回記事をご参照ください。

ここに、取得した画像を保存する機能を新たに追加します。

#前回実装した2つのクラスも格納
import libraries.TimeConverter as TimeConverter
import libraries.SoracomAPI as SoracomAPI

from pathlib import Path
import requests

class Image:
    def __init__(self, email, password):
        self.time_info = TimeConverter.TimeConverter()
        self.soracom = SoracomAPI.SoracomAPI(email,password)
    
    #指定時間の画像をエクスポート
    def export_image(self):
        get_image_time = self.time_info.get_image_time
        self.soracom.start_image_export(get_image_time)

    def save_image(self):
        #画像URLを取得
        image_URL = self.soracom.get_image_URL()
        print(image_URL)
    
        #YOLOに渡す画像を保存
        # 保存先のパスを設定
  current_directory = Path.cwd()
  save_path = current_directory / "image" / "target.png"

        # 画像をダウンロード
        response = requests.get(image_URL)
        response.raise_for_status()

        # 画像を保存
        with open(save_path, 'wb') as file:
            file.write(response.content)
        print("画像の保存に成功しました")
  • メインフォルダ
    • ultralytics
    • image
      • target.png
    • libraries
      • SoracomAPI.py
      • TimeConberter.py
      • Image.py

2. 人数カウント機能を実装

前回実装した、画像から人間を検出し、映っている人数をカウントする機能もlibrariesに追加します。

from ultralytics import YOLO

class Yolo :
    def __init__(self) :
        self.model = YOLO("yolov8x.pt") 
    
    def predict(self):
        results = self.model("image/target.png", save=False, classes=[0], conf=0.3)
        human_count  =  results[0].boxes.data.shape[0]

        return human_count
  • メインフォルダ
    • ultralytics
    • image
      • target.png
    • libraries
      • SoracomAPI.py
      • TimeConberter.py
      • Image.py
      • Yolo.py

3. kintoneでアプリを作成し、kintone REST APIでレコードを登録できるようにする

以下を参考に、kintone上でアプリを作成します。

アプリをはじめから作成する | kintone ヘルプ

人数列は数値、日時列は日時です。

日時はレコード登録時の日時を初期値にしてあります。

※日時は指定もできます。その場合、”2023-10-25T09:00:00+09:00”のような フォーマットで渡す必要があります。

このような形で作れます。

ここに、人数と日時をどんどん追加していきます。

後ほど必要なアプリIDも確認しておきましょう。

https://onl.la/4EFCbpd

作成したアプリにレコードを追加できるようにします。

kintone REST APIのリファレンスを参考にします。

kintone REST API 一覧 - cybozu developer network

取得した人数と、そのときの日時をレコードに追加するAPIを作成します。

※コード上の「人数」はフィールドコートと呼ばれます。 アプリの設定→フォーム→(列ごとの)設定から、確認・編集ができます。

import requests

class KintoneAPI:

    def __init__(self, api_token, content_type="application/json"):
        self.url = "https://kdl.cybozu.com/k/v1/record.json"
        self.headers = {
            "X-Cybozu-API-Token": api_token,
            "Content-Type": content_type
        }

    def post_record(self, human_count, nitiji):
        data = {
            "app": app-ID,
            "record": {
                "人数": {
                    "value": human_count
                }
            }
        }

        response = requests.post(self.url, headers=self.headers, json=data)
        return response
  • メインフォルダ
    • ultralytics
    • image
    • libraries
      • SoracomAPI.py
      • TimeConberter.py
      • Image.py
      • Yolo.py
      • KintoneAPI.py.py

これでOKです!

4. 一連の処理を定期的に実行する

最後に、ここまでの処理をまとめて定期的に実行しましょう。

定期実行したい処理をrunner.pyにまとめます。

画像のエクスポートには数秒かかるので、time.sleep()を使用して処理を一時休止しています。

#メールアドレス等の定義
email ="E_MAIL@ADDRES"
password = "PASSWORD"
X_Cybozu_API_Token= "MY_API_TOKUN"
Content_Type =  "application/json"

import time
import libraries.Image as Image
import libraries.Yolo as Yolo
import libraries.KintoneAPI as KintoneAPI 

image = Image.Image(email,password)
yolo = Yolo.Yolo()
kintone = KintoneAPI.KintoneAPI(X_Cybozu_API_Token)

image.export_image()

#少し待たないとエクスポートが完了しない
time.sleep(15)
image.save_image()

human_count = yolo.predict()
print(human_count)
kintone.post_record(human_count)
  • メインフォルダ
    • ultralytics
    • image
    • libraries
      • SoracomAPI.py
      • TimeConberter.py
      • Image.py
      • Yolo.py
      • KintoneAPI.py.py -runner.py

このrunner.pyを一定時間ごとに実行すれば

・オフィス画像の取得

・画像から人数のカウント

・カウントした人数をkintoneに反映

ができます。

処理の定期実行にはAPSchedulerを使用します。

平日9時から17時、1時間ごとに実行しましょう。

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
import subprocess
import os


def run_script():
    print(f"Running script at {datetime.now()}")
    process = subprocess.run(["python", os.path.join("runner.py")], capture_output=True, text=True)

    # 標準出力と標準エラー出力を取得
    stdout = process.stdout
    stderr = process.stderr

    if stderr :
        print(f"Error: {stderr}")
    print(f"Output: {stdout}")


scheduler = BlockingScheduler()

# 平日の9時から17時まで1時間ごとに実行するスケジュールを追加
scheduler.add_job(run_script, 'cron', day_of_week='mon-fri', hour='9-17')

print(f"Script scheduled to run every hour from 9 AM to 5 PM on weekdays: {datetime.now()}")
scheduler.start()

kintoneのアプリに戻りましょう

反映できてます!!!

この日は、観測範囲では2人から8人まで出社人数に幅があるようです。

時刻では10時、14時、17時が多くなっていますね。

毎日観測すれば、より詳しいことがわかるかもしれません。

いかがでしたか? 今回は、

・ソラカメからの画像の取得とAIによる人数のカウントを定期実行する

・結果をkintoneに反映する

を実装しました!

色々な技術・サービスが繋がる様子が楽しいですね。

kintone以外にも、例えばLINEやX(旧Twitter)などに反映しても面白いかもしれません。

今後機会があればやってみたいと思います。

最後まで読んで頂いてありがとうございました!!