Raspberry Pi + PaSoRi + Python で、勤怠打刻マシンを作ってみた

やってみた IoT

前置き

最近、当社の勤怠システムが更改されるとともに、勤怠打刻のWeb APIも公開されました。

すると、エンジニア間で「わたしのかんがえたさいきょうのきんたい」ブームが起き、いろいろな勤怠打刻方法が生み出されました。

今回は、私の作成した Raspberry Pi + PaSoRi + Python の勤怠打刻マシンをご紹介します。

仕様について

個人的に勤怠打刻に欲しい機能として

  • タイムカードボックスからタイムカードを取り出して打刻
    • 物理タイムレコーダーと同じような操作感がほしいため
  • 「打刻したら音声で挨拶する」「打刻したらSlackで通知する」
    • リモートワークしてても物理出退勤してる感を出したいため

があります。

　
そこで、Raspberry Piと PaSoRi と FeliCa を使って

  1. 無印のタイムカードボックスからFeliCaを取り出す
  2. FeliCaをPaSoRiにタッチする
  3. PaSoRiのつながっているRaspberry Piが反応し、Web APIで打刻する
  4. 打刻に成功したら、Raspberry Piに接続したスピーカーから音声を出す
  5. Slack API で打刻したことを通知する
  6. 出勤 と書かれたタイムカードボックスにFeliCaを入れる

ができるような仕様とします。

用意したもの

  • ハードウェア
    • Raspberry Pi 2 Model B (以降、ラズパイと表記)
      • Raspberry Pi OS, January 28th 2022
    • PaSoRi RC-S380
    • 100均のスピーカー XYZ-22-A
    • FeliCa KURURU
  • ソフトウェア
    • Python 3.9.2 (Raspberry Pi OS付属)
    • nfcpy 1.0.4
      • PasoRiでFeliCaを読むときに使用

勤怠打刻マシンの外観です。

手前の黒いPaSoRiにFeliCaをタッチし、出退勤を打刻します。

設定手順

Slack appの準備

今回、Slack botからSlack通知をするために、Slack appを準備します。

  • Bot tokens を使うSlack appを作成する

ラズパイのセットアップ

勤怠打刻用のラズパイをセットアップします。

  • SSHを可能にする
  • IPアドレスを固定化する
  • mp3ファイルを再生できるよう、 mpg321 をaptでインストールする
  • /home/pi/projects/dakoku_pi/ ディレクトリを作成する
    • このディレクトリに打刻用プログラムファイルを入れる

Pythonまわりの準備

Web APIでの勤怠打刻は、プログラミング言語を問わず利用できるようでした。

そこで、慣れているPythonを使って打刻してみます。

打刻マシンのソフトウェア構成について

もし今後、勤怠システムの更改があったとしても、今回作成する打刻マシンはなるべく変更箇所を少なくしたいです。

そこで今回は、

  • 共通的な処理を行う親クラス
    • Web APIで打刻する機能を呼び出す
    • 音声を出す
    • Slackへ通知する
  • システムごとの処理を行う子クラス
    • Web APIで勤怠打刻する

というクラス構成としました。

また、 dakoku_pi ディレクトリ以下を次のようにしました。

$ tree -L 2
.
├── dakoku/            # 打刻APIに関するPythonスクリプトを入れるディレクトリ
│   ├── base.py        # どの打刻APIであっても共通的に使う機能をまとめたファイル
│   ├── dakoku.py      # 打刻APIの詳細が記載されたファイル
│   └── __init__.py    # main.pyからimportするために使うファイル
├── main.py            # エントリポイント
├── slack.py           # slack-sdkのラッパー
└── voice/             # 音声ファイル用のディレクトリ
    ├── clock_in.mp3   # 出勤する時の音声ファイル
    └── clock_out.mp3  # 退勤する時の音声ファイル

Python環境の準備

必要なライブラリをインストールします。

なお、Slackのトークンなどの秘匿情報はハードコーディングせず、 .env ファイルに記載して python-dotenv で環境変数へロードすることとします。

% pip install requests python-dotenv slack_sdk nfcpy

slack-sdkのラッパーを作成 (slack.py)

複数箇所でSlackへの投稿を行うため、python-slack-sdkの薄いラッパーを用意しておきます。

import os

from slack_sdk import WebClient


class Slack:
  def __init__(self):
    self.client = WebClient(token=os.environ.get("SLACK_BOT_TOKEN"))

  def send(self, message):
    self.client.chat_postMessage(
      channel=os.environ.get('DESTINATION_OF_CHANNEL_OR_USER_ID'),
      text=message
    )

共通の機能をまとめた親クラスを作成 (dakoku/base.py)

再掲となりますが、このクラスでは

  • 打刻する を呼び出す
  • 音声を出す
  • Slackへ通知する

の機能を持たせます。

そのため、 打刻する を呼び出すところは

@abstractmethod
def clock(self) -> Optional[AvailableTypes]:
    pass

としておき、子クラスに実装を任せます。

その他の共通的な機能は以下とします。

import os
import pathlib
import subprocess
import time
from abc import ABCMeta, abstractmethod
from enum import Enum
from typing import Optional

BASE_DIR = pathlib.Path(__file__).resolve().parents[1]
VOICE_FILE_OF_CLOCK_IN = f'{BASE_DIR}/voice/clock_in.mp3'
VOICE_FILE_OF_CLOCK_OUT = f'{BASE_DIR}/voice/clock_out.mp3'

class AvailableTypes(Enum):
  CLOCK_IN = 'clock_in'
  CLOCK_OUT = 'clock_out'


class DakokuBase(metaclass=ABCMeta):
  def __init__(self, slack_client, idm, keep_power_on):
    self.api_headers = {
      'Content-Type': 'application/json',
    }

    self.slack_bot_token = os.environ.get('SLACK_BOT_TOKEN')
    self.slack_message_of_clock_in = os.environ.get('SLACK_MESSAGE_OF_CLOCK_IN')
    self.slack_message_of_clock_out = os.environ.get('SLACK_MESSAGE_OF_CLOCK_OUT')

    self.slack_client = slack_client
    self.idm = idm
    self.keep_power_on = keep_power_on

  @abstractmethod
  def clock(self) -> Optional[AvailableTypes]:
    pass

  def run(self) -> None:
    # 打刻する
    result = self.clock()
    # 打刻結果を元に音声を出す
    self.sound(result)
    # 打刻結果を元にSlackへ通知する
    self.notify(result)
    # 必要に応じてシャットダウンする
    self.shutdown_if_needed(result)

  def sound(self, result: Optional[AvailableTypes]) -> None:
    if result == AvailableTypes.CLOCK_IN:
      file = VOICE_FILE_OF_CLOCK_IN

    elif result == AvailableTypes.CLOCK_OUT:
      file = VOICE_FILE_OF_CLOCK_OUT

    else:
      return

    subprocess.call(f'mpg321 {file}', shell=True)

  def notify(self, result: Optional[AvailableTypes]) -> None:
    if result == AvailableTypes.CLOCK_IN:
      self.slack_client.send(self.slack_message_of_clock_in)

    elif result == AvailableTypes.CLOCK_OUT:
      self.slack_client.send(self.slack_message_of_clock_out)

    else:
      return

  def shutdown_if_needed(self, result: Optional[AvailableTypes]) -> None:
    # 退勤の場合のみシャットダウンを行う
    if result == AvailableTypes.CLOCK_OUT and not self.keep_power_on:
      self.slack_client.send('シャットダウンします')

      # Slack通知が終わってからシャットダウンできるよう、ちょっと待つ
      time.sleep(5)
      subprocess.call('sudo shutdown -h now', shell=True)

打刻方法を実装した子クラスを作成 (dakoku/dakoku.py)

DakokuBase を継承し、 clock メソッドを実装します。

社内の勤怠システムに依存するためここでは公開できませんが、 clock メソッドを実装します。

class Dakoku(DakokuBase):
    def clock(self) -> Optional[AvailableTypes]:
        # 打刻処理
        pass

import用の設定 (dakoku/__init__.py)

import階層が深くなるのを避けるため、 __init__.py にimportを追加します。

from dakoku.dakoku import Dakoku

エントリポイントを作成 (main.py)

今まで作成してきたファイルと nfcpy を使い、FeliCaを読み込むと打刻できるよう実装します。

なお、開発用にコマンドライン引数も用意しておきます。

import argparse
import binascii
import os
import pathlib

import nfc
from dotenv import load_dotenv

from dakoku import Dakoku
from slack import Slack

BASE_DIR = pathlib.Path(__file__).resolve().parent

parser = argparse.ArgumentParser()
parser.add_argument('-d', '--debug', action='store_true', help='FeliCaは読み込まれたものとして実行する')
parser.add_argument('-k', '--keep_power_on', action='store_true', help='退勤時もラズパイを起動させたままにする')


def on_connect(tag):
    idm = binascii.hexlify(tag._nfcid).decode('utf-8')
    print(f'FeliCa IDm: {idm}')

    slack_client = Slack()
    if idm != os.environ.get('FELICA_IDM_OF_CLOCK'):
        slack_client.send(f'このカードでは打刻できません: {idm}')
        return

    Dakoku(slack_client, keep_power_on=args.keep_power_on).run()


if __name__ == "__main__":
    args = parser.parse_args()
    load_dotenv()

    if args.debug:
        Dakoku(Slack(), keep_power_on=args.keep_power_on).run()

    else:
        try:
            print('読み取り開始')

            with nfc.ContactlessFrontend('usb:054c:06c3') as cf:
                cf.connect(rdwr={'on-connect': on_connect})

            print('終了')

        except Exception as e:
            print(e)

            with open(f'{BASE_DIR}/error.log', 'w') as f:
                f.write(str(e))

音声ファイルを準備する

打刻した時にスピーカーから音声を出すため、mp3形式のファイルを2つ(出勤・退勤)用意します。

文字から音声を作るサービスで作成したり、自分で録音したりしてください。

作成したら voice ディレクトリの中に

  • clock_in.mp3
  • clock_out.mp3

として保存します。

systemdまわりの準備

次に、systemdを使い「ラズパイへPaSoRiを挿入した時に上記スクリプトを実行することで、常時FeliCaのタッチを待ち受けている」状態にします。

udevによりUSBデバイス挿入を認識するよう設定

udev を使い、ラズパイのUSBポートへのPaSoRi接続を認識するよう設定します。

まず、 udev の rules を作成するため、PaSoRiの idVendor と idProduct を確認します。

$ dmesg | grep usb
...
[    4.353996] usb 1-1.4: New USB device found, idVendor=054c, idProduct=06c3, bcdDevice= 1.11
[    4.354035] usb 1-1.4: New USB device strings: Mfr=1, Product=2, SerialNumber=4
[    4.354057] usb 1-1.4: Product: RC-S380/P
[    4.354077] usb 1-1.4: Manufacturer: SONY
...

List of USB ID's - linux-usb.orgを見たところ、そのベンダーIDとデバイスIDの組み合わせが FeliCa S380 [PaSoRi] で間違いなさそうでした。

次に、 /etc/udev/rules.d/90-rc-s380.rules を以下の内容で作成し、serviceと関連付けます。

なお、serviceに指定した rc-s380.service は後ほど作成します。

SUBSYSTEM=="usb", ACTION=="add", ATTRS{idProduct}=="06c3", ATTRS{idVendor}=="054c", GROUP="plugdev", TAG+="systemd", ENV{SYSTEMD_WANTS}+="rc-s380.service", NAME="pasori380"

systemdのserviceを追加　

続いて、systemd用のservice として /etc/systemd/system/rc-s380.service を作成します。

[Unit]
Description=Pasori RC-S380 service
Requires=dev-bus-usb-001-004.device
After=dev-bus-usb-001-004.device

[Service]
Type=simple
User=pi
Restart=always
RestartSec=5
ExecStart=/usr/bin/python /home/pi/projects/dakoku_pi/main.py

以上で完成です。

おわりに

新しい勤怠システムがリリースされてからラズパイ勤怠を利用していますが、特に問題は発生していません。

また、打刻し忘れることもなく、安定して運用できています。