初めに
先日, 私はFlipsに入門し, 関数型っぽい感じで数理最適化with F#を扱うことが出来るようになりました.
そこで, 何かしらを作ってみたいな？と言うことで, Flipsを用いた数理最適化スケジューラをコアに据えた, シフトスケジューリングシステムを簡単に作ってみました.
本記事では, フロントエンド側に当たる, Excelファイルのアップロード, 受け取りデータの整形, データ送受信に関する処理を, streamlit, requestsを用いて作成してみた過程をまとめていきます.
アーキテクチャ設計など, 大量にツッコミどころはあるかと思います. 伸び代があると強引に言い換えて, 暖かく見てもらえたら幸いです.
目次
事前の決め事
このスケジューリングシステムは, 架空の居酒屋のシフト表作成を行います.
想定するユースケース
ユースケース図を描いてみます. 堅苦しく言いますが, イメージとしては, 必要なExcelファイルをアップロードしたら, そのデータに従ったシフト表が返ってきます.
Excelデータの形式
2種類のデータが必要です. 希望休のExcelファイルと, 人事的な情報が入った名簿Excelファイルです.
希望休ファイル
シート1: 希望休テーブル
希望休の表に関しては, こいこいさんの動画を参考にしました. 分かりやすくて良い表ですよね. イメージとしては, 以下のような表です.
| 2026 | 年 | シフト表 | ||||||||||||||||||||||||||||||
| 7 | 月 | |||||||||||||||||||||||||||||||
| 水 | 木 | 金 | 土 | 日 | 月 | 火 | 水 | 木 | 金 | 土 | 日 | 月 | 火 | 水 | 木 | 金 | 土 | 日 | 月 | 火 | 水 | 木 | 金 | 土 | 日 | 月 | 火 | 水 | 木 | 金 | ||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 社員ID | 名前 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15 | 16 | 17 | 18 | 19 | 20 | 21 | 22 | 23 | 24 | 25 | 26 | 27 | 28 | 29 | 30 | 31 |
| 1 | 佐藤 | ◎ | ◎ | |||||||||||||||||||||||||||||
| 2 | 鈴木 | ◎ | ◎ | |||||||||||||||||||||||||||||
| 3 | 高橋 | |||||||||||||||||||||||||||||||
| 4 | 田中 | ◎ | ◎ | |||||||||||||||||||||||||||||
| 5 | 伊藤 | ◎ | ||||||||||||||||||||||||||||||
| 6 | 渡辺 | |||||||||||||||||||||||||||||||
| 7 | 山本 | ◎ | ||||||||||||||||||||||||||||||
| 8 | 中村 | ◎ | ||||||||||||||||||||||||||||||
| 9 | 小林 | ◎ | ||||||||||||||||||||||||||||||
| 10 | 加藤 | ◎ | ◎ | |||||||||||||||||||||||||||||
シート2: 最低人数
続いて, 各日付と最低人数を記載したシートです.
| 2026 | 年 | ||
| 7 | 月 | ||
| 日 | 曜日 | 最低人数 | 備考 |
|---|---|---|---|
| 1 | 木 | 4 | 通常 |
| 2 | 金 | 7 | 繁忙 |
| 3 | 土 | 7 | 繁忙 |
| 4 | 日 | 4 | 通常 |
| 5 | 月 | 4 | 通常 |
| 6 | 火 | 4 | 通常 |
| 7 | 水 | 0 | 定休日 |
| 8 | 木 | 4 | 通常 |
| 9 | 金 | 7 | 繁忙 |
| 10 | 土 | 7 | 繁忙 |
| 11 | 日 | 4 | 通常 |
| 12 | 月 | 4 | 通常 |
| 13 | 火 | 4 | 通常 |
| 14 | 水 | 0 | 定休日 |
| 15 | 木 | 4 | 通常 |
| 16 | 金 | 7 | 繁忙 |
| 17 | 土 | 7 | 繁忙 |
| 18 | 日 | 7 | 繁忙 |
| 19 | 月 | 4 | 通常 |
| 20 | 火 | 4 | 通常 |
| 21 | 水 | 0 | 定休日 |
| 22 | 木 | 4 | 通常 |
| 23 | 金 | 7 | 繁忙 |
| 24 | 土 | 7 | 繁忙 |
| 25 | 日 | 4 | 通常 |
| 26 | 月 | 4 | 通常 |
| 27 | 火 | 4 | 通常 |
| 28 | 水 | 0 | 定休日 |
| 29 | 木 | 4 | 通常 |
| 30 | 金 | 7 | 繁忙 |
| 31 | 火 | 4 | 通常 |
シート3: 出勤可能最大日数
最後に, 出勤可能な最大日数です.
| 社員ID | 名前 | 出勤可能最大日数 |
|---|---|---|
| 1 | 佐藤 | 26 |
| 2 | 鈴木 | 26 |
| 3 | 高橋 | 26 |
| 4 | 田中 | 26 |
| 5 | 伊藤 | 26 |
| 6 | 渡辺 | 26 |
| 7 | 山本 | 26 |
| 8 | 中村 | 26 |
| 9 | 小林 | 26 |
| 10 | 加藤 | 26 |
名簿Excelファイル
次が名簿Excelファイルです. 架空の居酒屋の経営なので, リーダーを任せられる人材に関する情報を付記しています.
| idf | 名前 | リーダー |
|---|---|---|
| 1 | 佐藤 | 〇 |
| 2 | 鈴木 | 〇 |
| 3 | 高橋 | 〇 |
| 4 | 田中 | |
| 5 | 伊藤 | |
| 6 | 渡辺 | |
| 7 | 山本 | |
| 8 | 中村 | |
| 9 | 小林 | |
| 10 | 加藤 |
ざっくり設計
今回は, 計算データの保存等, DBを扱いません. 今後, 気が向いたら作ろうかな？とは考えています.
データモデル
まずはデータモデルです. Pydantic.BaseModelを使用します. サクッと従属関係を示します. 矢印の方向が従属する先です.
人の情報と希望休リクエストについて. 特筆すべきことはありませんが, シフト計算の都合上, リーダーフラグを持たせています. 店長->副店長->バイトリーダーなど, もっと階層化したかったら列挙体を考慮すべきでしょうか.
class Person(BaseModel):
idf: int = Field(
...,
description="ユーザID"
)
name: str = Field(
...,
description="ユーザ名"
)
is_leader: bool = Field(
...,
description="リーダーフラグ"
)
class ShiftRequest(BaseModel):
person_idf: int = Field(
...,
description="ユーザID, Person.idfと共通"
)
holiday_requests: list[int] = Field(
...,
description="希望休の日付リスト"
)
max_shift: int = Field(
...,
description="当該月に出勤出来る最大の日数"
)
続いて, 計算機(バックエンド側)とのやり取りです. CalcResultに関して, ここはかなりシンプルにしています. 集計結果を格納して表示しても良いかなーと思っています.
また, CalcResponseは計算機(バックエンド)とのやり取りに特化させ, CalcResultは画面表示に特化させるため, 両者に共通する部分が増えています.
class CalcRequest(BaseModel):
shift_requests: dict[int, ShiftRequest] = Field(
...,
description="ユーザID vs 希望休リスト"
)
person_info: dict[int, Person] = Field(
...,
description="ユーザID vs 人事情報"
)
resource_requests: dict[int, int] = Field(
...,
description="日付 vs 必要な人員数, 0なら定休日と解釈する"
)
class CalcResponse(BaseModel):
success: bool = Field(
...,
description="計算が成功したか否か"
)
message: str = Field(
...,
description="メッセージ"
)
calc_time: float = Field(
...,
description="計算時間[ms]"
)
result_dict: dict[int, list[int]] = Field(
...,
description="ユーザID vs 出勤日リスト"
)
class CalcResult(BaseModel):
success: bool = Field(
...,
description="計算が成功したか否か"
)
message: str = Field(
...,
description="メッセージ"
)
result_array: list[list[str]] = Field(
...,
description="結果の配列"
)
データフロー
データの流れはシンプルですね. 先述したデータモデルでやり取りします.
フォルダ構成
フォルダ構成は以下の通りです.
.
├── Makefile
├── README.md
├── main.py
├── pyproject.toml
├── data/
│ ├── base/
│ │ ├── 7月シフト表.xlsx
│ │ ├── 8月シフト表.xlsx
│ │ └── 名簿.xlsx
│ ├── configs/
│ │ └── configs.json
│ └── settings/
│ └── settings.json
└── src/
├── gui/
│ ├── drawer.py
│ ├── pages/
│ │ ├── calculator/
│ │ │ └── page_calc.py
│ │ └── data_uploader/
│ │ ├── page_up.py
│ │ └── parts/
│ │ ├── container_up.py
│ │ └── container_valid.py
│ └── sidebar/
│ └── sidebar_drawer.py
├── interface/
│ ├── if_factory.py
│ └── solver_if.py
├── prc/
│ ├── processor.py
│ └── tools/
│ ├── converter.py
│ └── validator.py
└── utils/
├── configs/
│ ├── default_paths.py
│ └── fixed_keys.py
├── enums/
│ └── pages.py
├── handlers/
│ ├── json_handler.py
│ └── path_handler.py
└── schemas/
├── configs_model.py
├── data_models.py
└── settings_model.py
srcディレクトリ直下のフォルダ同士には, 以下のような依存関係を作っています. これにより, 最初から循環参照が起こらないように気を付けています.
実装のポイント
ベーシックな事しかしていませんが, 個人的なTipsに近い事を書きます.
Excelデータのパース処理
コンバータクラス(src/prc/tools/converter.py)を省略して記載します. 計算用の生データを受け取り, 計算リクエストに変換して出力します. 中略していますが, 要するに, コンバータクラスでの呼び出し関数は1つにまとめて, コンバータクラスのプライベート関数でそれぞれを処理していく, という手段を取っています.
from pandas import DataFrame
from src.utils.schemas.data_models import CalcRequest, Person, ShiftRequest
class Converter:
def __init__(self):
pass
## 中略
def convert_raw_data(
self,
raw_data_dict: dict[str, DataFrame]
) -> CalcRequest:
# 希望休のテーブル
holiday_request_table = raw_data_dict["holiday_request_table"]
# 最低人数
least_resource_table = raw_data_dict["least_resource_table"]
# 各人員の最大日数
max_shift_table = raw_data_dict["max_shift_table"]
# 各メンバ情報
member_table = raw_data_dict["member_table"]
shift_requests: dict[int, ShiftRequest] = self.__create_shift_requests(
holiday_request_table=holiday_request_table,
max_shift_table=max_shift_table
)
person_info: dict[int, Person] = self.__create_person_info(
member_table=member_table
)
resource_requests: dict[int, int] = self.__create_resource_requests(
least_resource_table=least_resource_table
)
return CalcRequest(
shift_requests=shift_requests,
person_info=person_info,
resource_requests=resource_requests
)
画面描画周り
これもまた個人的なTipsですが, streamlitで実装する際, 幾つか気を付けている事を書きます.
※心がけに近いものもありますが, ご愛敬ということにしてください.
streamlitのハンドリングはguiディレクトリ直下のみ
掲題の通りですが, st.~は, guiディレクトリの直下に限ります. フォルダ構成の章でも説明しましたが, guiディレクトリは, 参照関係という観点で, (mainファイルを除いて)「最上流」に位置します. つまり, streamlitによるGUI変更を行った際, 「下流」(データアップロードや計算リクエストなど)の処理に対して影響を与えない, という意図があります.
.
└── src/
├── gui/
│ ├── drawer.py
│ ├── pages/
│ │ ├── calculator/
│ │ │ └── page_calc.py
│ │ └── data_uploader/
│ │ ├── page_up.py
│ │ └── parts/
│ │ ├── container_up.py
│ │ └── container_valid.py
│ └── sidebar/
│ └── sidebar_drawer.py
<後略>
GUIの各ページはenumで制御する
掲題の通りですが, GUIの各ページはenumで制御するように実装しています. まずは, GUI描画に関連するディレクトリ/ファイルの構造をご覧ください.
.
└── src/
├── gui/
│ ├── drawer.py
│ ├── pages/
│ │ ├── calculator/
│ │ │ └── page_calc.py
│ │ └── data_uploader/
│ │ └── page_up.py
│ └── sidebar/
│ └── sidebar_drawer.py
<中略>
└── utils/
├── enums/
│ └── pages.py
<後略>
上記のツリーで, 示したファイルの中で, src/gui/drawer.pyが以下のモジュールを呼び出します. 以下のモジュールがページとサイドバーの外観を描画します.
src/gui/pages/data_uploader/page_up.pysrc/gui/pages/calculator/page_calc.pysrc/gui/sidebar/sidebar_drawer.py
また, src/gui/drawer.pyは, 上記のモジュールを呼び出す際に, サイドバーにページ選択のラジオボタンを設置し, 選択されたラジオボタンに応じて, 画面を描画します. 各ラジオボタンがsrc/utils/enums/pages.pyのenumと対応しています. これにより, 人間の判断ではなく, 機械的な判断で, 画面がそれぞれ描画されるようにしています.
# src/gui/drawer.py
import streamlit as st
from streamlit.delta_generator import DeltaGenerator
from src.gui.pages.calculator.page_calc import draw_calc
from src.gui.pages.data_uploader.page_up import draw_data_uploader
from src.gui.sidebar.sidebar_drawer import draw_sidebar
from src.prc.processor import Processor
from src.utils.enums.pages import Pages, get_page
# 中略
def draw_gui(settings_path: str):
st.title("Shift table creation")
# processorを取得する
processor: Processor = __get_processor(settings_path)
# サイドバーの設定
sidebar: DeltaGenerator = st.sidebar
draw_sidebar(sidebar)
# 各ページの設定
sidebar.write("## Page selection")
selected_page = sidebar.radio(
label="pages",
options=[page.value for page in Pages]
)
# 各ページ描写
page: Pages = get_page(selected_page)
container: DeltaGenerator = st.container()
if page == Pages.DATA_UPLOADER:
draw_data_uploader(container, processor)
elif page == Pages.CALCULATOR:
draw_calc(container, processor)
else:
raise ValueError(f"Invalid page: {page.name}")
# src/utils/enums/pages.py
from enum import Enum, unique
from typing import Optional
@unique
class Pages(Enum):
DATA_UPLOADER = "Data uploader"
CALCULATOR = "Calculator"
def get_page(__val) -> Pages:
target: Optional[Pages] = None
for page in Pages:
if __val == page.value:
target = page
if target is None:
raise ValueError(f"Invalid value: {__val}")
return target
このように実装しておけば, 以下のステップを踏むことで簡単にページを追加でき, しかも他に影響を与えない変更が可能です.
-
src/utils/enums/pages.pyに新たなページの列挙体と, 名前を定義する -
src/gui/drawer.pyに新たな条件分岐を追加する -
src/gui/pages/以下に新たなフォルダ, ファイルを作成し, 新たなページの処理関数をsrc/gui/drawer.pyに記載する
st.session_stateのキーはそれ専用の.pyファイルを作成して管理する
前提として, streamlitのセッション管理について軽く触れておきます.
streamlitでは, 例えばボタンを押したりラジオボタンを選択したり, そのような「変更」を加える度に, セッションがリセットされます. この為, 「ボタンAを押した状態を保存して, ボタンBを押す」等の処理を行いたい場合には, 「ボタンAを押した状態」を保存する必要があります. そこで必要になってくるのが, st.session_stateです.
ところが, こちらのst.session_stateが中々の曲者でして. 辞書風のオブジェクトですので, キーを覚えておく必要が出てきます. しかし, 私は人間ですので, タイポを起こすことが良くあります. そのような理由で, src/utils/configs/fixed_keys.pyというファイルに, キーをべた書きして, それぞれのファイルから呼び出す, というハンドリングをしています.
# src/utils/configs/fixed_keys.py
# processorのキー
PROCESSOR_KEY = "processor"
# 入力ファイル周り
INPUT_FILES_KEY = "input_dict"
SHFIT_TABLE = "shift_table"
MEMBER_LIST = "member_list"
# 生データ周り
RAW_DATA_KEY = "raw_data_dict"
HOLIDAY_REQUEST = "holiday_request_table"
LEAST_RESOURCE = "least_resource_table"
MAX_SHIFT = "max_shift_table"
MEMBER = "member_table"
正直, このハンドリングで良いのか？という疑問は, 常に頭をよぎっています. ただし, 少なくとも, 別ファイルから同じsession_stateの値を参照する際には役立っています.
通信インターフェース
通信インターフェースに関して, 順番に説明します. まずはディレクトリ構造図をご覧ください.
.
├── data/
│ └── configs/
│ └── configs.json
└── src/
├── gui/
<中略>
├── interface/
│ ├── if_factory.py
│ └── solver_if.py
├── prc/
│ ├── processor.py
│ └── tools/
│ ├── converter.py
│ └── validator.py
└── utils/
<中略>
└── schemas/
├── configs_model.py
<後略>
src/utils/schemas/configs_model.pyにて, コンフィグ情報を定義します. シンプルですよね.
# src/utils/schemas/configs_model.py
from pydantic import BaseModel
class Configs(BaseModel):
url_base: str
solver_request_path: str
続いて, data/configs/configs.jsonをディクショナリとして取得し, 上記のConfigsモデルに変換します.
そして, Configsを, src/prc/processor.pyを通して, src/interface/if_factory.pyに流し込みます.
ここで, if_factory.pyとsolver_if.pyは, いわゆるファクトリーパターンを採用しています.
呼び出しがソルバだけなら同一ファイルにしてしまっても良いと思いますが, ソルバ以外にもAPI連携を行いたくなった時の為, 敢えて少し冗長にしています.
次がインターフェースの実装です.
# src/interface/solver_if.py
from time import time
import requests
from src.utils.handlers.json_handler import loads
from src.utils.schemas.data_models import CalcRequest, CalcResponse
class SolverInterface:
def __init__(
self,
url_base: str,
solver_request_path: str
) -> None:
self.__url_base: str = url_base
self.__solver_request_path: str = solver_request_path
self.__url = f"{self.__url_base}/{self.__solver_request_path}"
def throw_request(
self,
req: CalcRequest
) -> CalcResponse:
url = self.__url
time0 = time()
try:
response = requests.post(url, json=req.model_dump(), timeout=10)
# HTTPエラー（4xx, 5xx）を例外として検知
response.raise_for_status()
values = loads(response.text)
return CalcResponse(**values)
except requests.exceptions.RequestException as e:
time1 = time()
return CalcResponse(
success=False,
message=f"通信エラーが発生しました: {str(e)}",
calc_time=time1 - time0,
result_dict={}
)
次がインターフェースファクトリーの実装です.
# src/interface/if_factory.py
from src.interface.solver_if import SolverInterface
from src.utils.schemas.configs_model import Configs
def create_solver_if(configs: Configs) -> SolverInterface:
solver_if: SolverInterface = SolverInterface(
url_base=configs.url_base,
solver_request_path=configs.solver_request_path
)
return solver_if
画面スクショ
簡単に画面スクショを張っておきます. まずはデータローディング画面から.
デフォルトのデータをセットしております. データをアップロードすると,
以下の画面になります. アップロードしたデータの内容を確認できます.
計算機側です. まず, データを上げないと怒られます.
データを上げると, 計算ボタンが出現します.
計算が上手く行くと, 以下の画面になります.
最後に
最後まで読んでくださってありがとうございました. 計算機側の話は気が向いたら書くと思います.
Appendix
リンク集です