""" サービスロボット導入シミュレーション #1 配膳ロボット（試作 v2.4） v2.4 で直したこと（2026-10-22。動画の台本を codex で検証したときに見つかった2点。直し方はエイイチと決めた） 1. 入店待ちの終わりを「案内されて卓に着いたとき」にした。v2.3 までは、案内したスタッフが説明（guide_s）を 終えて受け渡し口まで戻るときまで数えていた。説明の時間と戻りの時間は、スタッフの手がふさがる時間には これまでどおり入れ、お客さんの流れ（注文を待つ）もこれまでどおりスタッフが戻ってから進める。 2. 「下げ膳にも使う」（bus）で、ロボットがスタッフより先に卓へ行き、皿を載せる前に帰っていたのを直した。 スタッフが片付けに向かうとき、空いているロボットがいれば一緒に卓へ向かう。ロボットはスタッフが皿を 載せ終えるまで卓で待ち、それから洗い場へ戻る（待っている間もロボットはふさがる）。空いていなければ人だけで下げる。 上げ膳だけ（serve）・食事中に集めて回る（patrol）は、入店待ち（と、その内訳の案内）のほかは v2.3 と同じ数字になる（図面のある店も同じ）。 飲食店のピーク時間を SimPy で再現する。 図面のある店（v2.3） 標準ファミレスの中に、店の形（図面）から作った店を6つ足した（store が fam_p0〜fam_p5）。 卓の数と、卓ごとの厨房（料理の受け渡し口）から卓までの距離は店の形から取る（layouts/ の *_analysis.json。 正は p2m-site の public/sim/layouts/）。来た組には、卓ごとの距離の中から1つを選んで使う。 来客のペースとスタッフの人数は、標準ファミレス（35卓）を卓の数に合わせて増減させる。それ以外は標準ファミレスと同じ。 今までの5つの店は、乱数の使い方も含めて何も変えていない（数字は v2.2 と同じ）。 入店待ちの内訳と「店が回らない」（v2.2） 入店待ちは、店に来てから案内されて卓に着くまで（entry_wait_mean。v2.4 で終わりを直した）。 そのうち、満席で待った時間（full_wait_mean）と、席は空いているのに片付け（下げ膳・卓拭き）や 案内が終わらず待った時間（staff_wait_mean）に分ける。staff_wait_mean が NG_MIN 分以上なら「店が回らない」。 片付け待ちは、客が帰ってから下げ膳が終わるまでの卓の数だけ、入口の列の前から数えた組が 「席はあるのに待っている」とみなして数える。乱数の使い方は変えていないので、ほかの数字は v2 と同じ。 時間の区切り方（v2.1） ページに出す数字（ピークの回し方）：空の店に peak_min 分だけ選んだペースで来客し、全員が帰るまで回す。 ランチ・ディナーのように、混む時間は限られている前提。v2 と同じで、数字も v2 と同じになる。 回るかの確かめ（--sustain。v2.2 のページでは使わない）：空の店から始め、来客は選んだペースのまま止めない。 最初の warmup_min 分は数えず、そのあとの peak_min 分に来た組を、卓が空くまで追いかける。 その時間のスタッフの手がふさがっている割合（staff_util）を出し、STAFF_FULL 以上なら「店が回らない」とする。 入店待ちの延び方（seat_wait_trend）は、卓の数を超える客が来ると人手に関係なく延びるので、判定には使わない （混雑は多くの店で卓の数を超えている。そのときの行列は、ピーク2時間の入店待ちの数字に出ている）。 店のタイプ（5種）、ロボットの使い方（3種）、1往復で回る卓の数（段の数）、 倒れやすい物（デザート・ドリンク）を誰が運ぶか、を切り替えて比べられる。 数字はすべて仮置き。公開事例・メーカー仕様・観察で順に置き換える。 SimPy が扱うのは「店の図面がなくても答えられる問い」（時間・順番・数）だけ。 通路での鉢合わせ・接触・揺れ・こぼれは Unity で扱い、その結果を 「一時停止」「立ち往生」の数字や、ロボットに載せる皿の決まりとしてここに入れる。 ロボットの使い方 serve : 上げ膳だけ。下げ膳は人がやる（ガスト型） bus : 下げ膳にも使う。皿はスタッフがロボットに載せる（サイゼリヤ・ブロンコビリー型） patrol : 食事中に席を回り、客が空いた皿を載せる（焼肉の和民・かみむら牧場型） 使い方: pip install simpy python haizen_sim.py # 総当たりを回して results.json を書く python haizen_sim.py --main # よく使う人数・台数（0〜3台）だけ先に回す 途中経過は results_partial.json に保存され、止まっても続きから回る python haizen_sim.py --one # 1条件だけ回して結果を表示 python haizen_sim.py --sustain # 回るかの確かめを総当たりで回して results_sustain.json を書く 依存: Python 3.10 以降 / simpy 4.1 """ import json import random import statistics import sys from itertools import product from multiprocessing import Pool import simpy # --------------------------------------------------------------- # 共通の前提（すべて仮） # --------------------------------------------------------------- COMMON = { "staff_speed": 1.0, # スタッフが歩く速さ（m/秒） "robot_speed": 0.5, # ロボットが店内を走る速さ（m/秒） "guide_s": 45, # 案内（歩く時間は別） "staff_place_s": 10, # スタッフが皿を置く "robot_handoff_s": 25, # 客がロボットから皿を取る "pay_s": 90, # 会計 "call_s": 30, # 呼び出し対応（歩く時間は別） "clear_s": 40, # テーブルの皿をまとめる "wipe_s": 40, # 卓を拭いてセットする "load_s": 15, # まとめた皿をロボットに載せる "guest_load_s": 20, # 客が空いた皿をロボットに載せる（巡回回収） "pause_rate": 0.3, # 一時停止：1往復で、人が前を横切って止まる確率（自然に解ける） "pause_s": (3, 10), # 一時停止の長さ（秒） "stuck_rate": 0.01, # 立ち往生：1往復で、スタッフが動かしに行かないと解けない確率 "rescue_s": 15, # スタッフがロボットを動かす作業 "hop_m": 3, # 1往復で複数卓を回るとき、卓から次の卓への寄り道（m） "fragile_cook": (2, 4, 7), # デザート・ドリンクができるまで（分） "cold_min": 3, # これを超えて届いた皿を「待たせた皿」とする "dish_extra_m": 0, # 洗い場までの追加距離（店のタイプで上書き） "sustain": False, # True：回るかの確かめ（来客を止めない回し方） "warmup_min": 60, # 回るかの確かめで、数えない慣らしの時間（分） "util_sample_s": 10, # 手がふさがっている人数・台数を数える間隔（秒） "max_extra_min": 600, # 数える時間のあと、追いかけるのを打ち切る上限（分）。店が回らない条件の保険 "robot_first_wait_s": 120, # v2.7 ロボット優先：料理はロボットを待つ。これだけ待っても来なければスタッフが運ぶ（秒） "model": "v2.7", # モデルの版（前の版の結果を使い回さないための印） } # --------------------------------------------------------------- # 店のタイプ（すべて仮） # dist_m : 厨房（洗い場）から卓までの距離の幅 # first_dishes : 最初の注文の皿。料理ができるまで（最短, 最頻, 最長 分） # extra_orders : 食事中の追加注文の回数（食べ放題） # bus_trips : 下げ膳を人だけでやるときの往復回数（皿の量） # dish_extra_m : 洗い場が厨房の受け渡し口よりさらに奥にあるときの追加距離（ホテル小倉の事例） # fragile : 1組あたりの倒れやすい物（デザート・パフェ・ドリンク）の数の平均 # staff : 最初に見せる人数 # staff_all : 「その他」で選べる人数（総当たりはこちら） # arrivals : 来客ペース（空いている・普通・混雑） # --------------------------------------------------------------- STORES = { "small": { "label": "小型店", "tables": 18, "dist_m": (4, 12), "peak_min": 120, "first_dishes": [(4, 7, 11), (6, 9, 14)], "extra_orders": 0, "extra_cook": (3, 5, 8), "eat_min": (15, 25, 40), "call_rate": 0.4, "bus_trips": 1, "fragile": 0.3, "staff": [1, 2, 3], "staff_all": [1, 2, 3, 4, 5], "arrivals": [15, 22, 30], }, "family": { "label": "標準ファミレス", "tables": 35, "dist_m": (6, 22), "peak_min": 120, "first_dishes": [(5, 8, 12), (8, 12, 18)], "extra_orders": 0, "extra_cook": (3, 5, 8), "eat_min": (20, 30, 45), "call_rate": 0.6, "bus_trips": 2, "fragile": 0.5, "staff": [2, 3, 4], "staff_all": [1, 2, 3, 4, 5, 6, 7], "arrivals": [25, 35, 45], }, "large": { "label": "大型店", "tables": 60, "dist_m": (8, 35), "peak_min": 120, "first_dishes": [(5, 8, 12), (8, 12, 18)], "extra_orders": 0, "extra_cook": (3, 5, 8), "eat_min": (20, 30, 45), "call_rate": 0.6, "bus_trips": 2, "fragile": 0.5, "staff": [4, 5, 6], "staff_all": [2, 3, 4, 5, 6, 7, 8, 9, 10], "arrivals": [45, 60, 75], }, "hotel": { "label": "洗い場が遠い店", "tables": 30, "dist_m": (6, 20), "dish_extra_m": 30, "peak_min": 120, "first_dishes": [(5, 8, 12), (8, 12, 18)], "extra_orders": 2, "extra_cook": (4, 6, 10), "eat_min": (30, 45, 60), "call_rate": 0.6, "bus_trips": 2, "fragile": 0.6, "staff": [2, 3, 4], "staff_all": [1, 2, 3, 4, 5, 6, 7], "arrivals": [16, 22, 28], }, "buffet": { "label": "焼肉食べ放題", "tables": 35, "dist_m": (6, 22), "peak_min": 150, "first_dishes": [(2, 4, 8), (3, 5, 9), (4, 7, 12)], "extra_orders": 5, "extra_cook": (2, 4, 7), "eat_min": (60, 80, 90), "call_rate": 0.8, "bus_trips": 3, "fragile": 1.5, "staff": [2, 3, 4], "staff_all": [1, 2, 3, 4, 5, 6, 7], "arrivals": [14, 20, 26], }, } # --------------------------------------------------------------- # 図面のある店（v2.3）：標準ファミレスの中の、店の形から作った店 # --------------------------------------------------------------- import os as _os LAYOUT_DIR = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "layouts") LAYOUT_LABELS = { "p0": "② 定番の郊外型", "p1": "③ 効率重視・入口が角", "p2": "④ 効率重視・小さめの店", "p3": "⑤ ゆったり・入口が正面", "p4": "⑥ ゆったり・入口が横", "p5": "⑦ ③の通路を80cmに", } def _scale(vals, t, base_t): return [max(1, round(v * t / base_t)) for v in vals] def layout_stores(): """layouts/ の店の形から、標準ファミレスをもとにした店を作る""" out = {} base = STORES["family"] names = sorted(f for f in _os.listdir(LAYOUT_DIR) if f.endswith("_analysis.json")) if _os.path.isdir(LAYOUT_DIR) else [] for f in names: a = json.load(open(_os.path.join(LAYOUT_DIR, f), encoding="utf-8")) nm = f.split("_")[0] dists = [t["dist"] for t in a["tables"]] if any(d is None for d in dists): raise ValueError(f"{f}: ロボットが届かない卓がある（人が運ぶ卓はまだ試算に入れていない）") t = len(dists) c = max(2, round(base["staff"][1] * t / base["tables"])) # 真ん中の人数 st = {**base} st.update({ "label": LAYOUT_LABELS.get(nm, nm), "base": "family", "layout": nm, "tables": t, "dist_m": (round(min(dists), 1), round(max(dists), 1)), "dist_list": dists, "arrivals": _scale(base["arrivals"], t, base["tables"]), "staff": [c - 1, c, c + 1], "staff_all": list(range(max(1, c - 2), c + 5)), }) out["fam_" + nm] = st return out STORES.update(layout_stores()) ROBOTS = [0, 1, 2, 3] # 最初に見せる台数 ROBOTS_ALL = list(range(7)) # 「その他」で選べる台数（総当たりはこちら） MODES = ["serve", "bus", "patrol"] TRAYS = [1, 3] # 1往復で回る卓の数（1卓ずつ／最大3卓） FRAGILE_BY = ["any", "staff"] # 倒れやすい物を ロボットも運ぶ／人だけが運ぶ REPS = 40 def tri(rng, lmh): lo, mode, hi = lmh return rng.triangular(lo, hi, mode) class Store: def __init__(self, env, rng, p): self.env, self.rng, self.p = env, rng, p self.tables = simpy.Resource(env, capacity=p["tables"]) self.staff = simpy.PriorityResource(env, capacity=p["staff"]) self.robots = simpy.PriorityResource(env, capacity=p["robots"]) if p["robots"] else None self.dish_waits, self.seat_waits, self.turns = [], [], [] self.staff_walk_m = self.staff_busy_s = self.robot_busy_s = 0.0 self.robot_trips = self.staff_trips = self.rescues = self.collects = self.done = 0 self.robot_bus = self.staff_bus = 0 self.fragile_robot = self.pauses = 0 self.pass_q = [] # 受け渡し口でロボットを待っている料理 # 回るかの確かめ（sustain）で数える時間。ピークの回し方では全部の組を数える self.t0 = p["warmup_min"] * 60 if p["sustain"] else 0 self.t1 = self.t0 + p["peak_min"] * 60 self.counted = self.counted_done = 0 self.window_closed = False self.finished = env.event() self.capped = False self.seat_wait_by_arrival = [] # (来た時刻, 入店待ち 分)。待ちが伸び続けるかを見る self.staff_busy_n = [] # 数える時間のあいだ、一定間隔で数えた手がふさがっている人数 self.robot_busy_n = [] self.snap0 = self.snap1 = None # 入店待ちの内訳（v2.2） # 片付けが終わらず使えない卓（客は帰ったが下げ膳・卓拭きが済んでいない）の数 # 入口の列の前から数えて、その数までの組は「席はあるのに待っている」とみなす self.dirty = 0 self._acc_t = 0.0 self.blocked = {} # 卓の順番待ちの request → 片付け待ちで待った秒数 self.blocked_waits = [] # 組ごと：席はあるのに片付けが終わらず待った時間（分） self.guide_waits = [] # 組ごと：卓が決まってから、案内のスタッフの手が空くまで（分） self.entry_waits = [] # 組ごと：店に来てから、案内されて席に着くまで（分） def _acc(self): """前回からの時間を、列の前から片付け待ちの卓の数までの組に足す（列や卓の数が変わる直前に呼ぶ）""" now = self.env.now dt = now - self._acc_t if dt > 0 and self.dirty > 0: for req in self.tables.queue[:self.dirty]: self.blocked[req] = self.blocked.get(req, 0.0) + dt self._acc_t = now SNAP = ("staff_walk_m", "robot_trips", "staff_trips", "rescues", "collects", "robot_bus", "staff_bus", "fragile_robot", "pauses") def _snap(self): return {k: getattr(self, k) for k in self.SNAP} def window(self): """数える時間の始めと終わりに数を控え、そのあいだは手がふさがっている人数・台数を数える""" env, p = self.env, self.p yield env.timeout(self.t0) self.snap0 = self._snap() while env.now < self.t1: self.staff_busy_n.append(self.staff.count) if self.robots is not None: self.robot_busy_n.append(self.robots.count) yield env.timeout(p["util_sample_s"]) self.snap1 = self._snap() self.window_closed = True self._check_finished() yield env.timeout(p["max_extra_min"] * 60) if not self.finished.triggered: # 店が回らず、数える組が出きらない self.capped = True self.finished.succeed() def _check_finished(self): if self.window_closed and self.counted_done >= self.counted and not self.finished.triggered: self.finished.succeed() # ---- スタッフの仕事（priority 0=ロボット救出, 1=配膳・案内・会計, 2=下げ膳） def staff_job(self, priority, walk_m, work_s, waited=None): t0 = self.env.now with self.staff.request(priority=priority) as req: yield req if waited is not None: waited.append(self.env.now - t0) # 手が空くのを待った時間（案内待ち） t = walk_m / self.p["staff_speed"] + work_s self.staff_walk_m += walk_m self.staff_busy_s += t yield self.env.timeout(t) # ---- ロボットが途中で止まる（1往復に1回判定） def maybe_stop(self, d, helper_waiting=False): p, rng = self.p, self.rng if rng.random() < p["pause_rate"]: # 一時停止：数秒で自然に解ける self.pauses += 1 yield self.env.timeout(rng.uniform(*p["pause_s"])) if rng.random() < p["stuck_rate"]: # 立ち往生：スタッフが動かしに行く self.rescues += 1 if helper_waiting: # 卓でロボットを待っているスタッフがそのまま動かす yield self.env.timeout(d / p["staff_speed"] + p["rescue_s"]) else: yield self.env.process(self.staff_job(0, d, p["rescue_s"])) # ---- 配膳 # v2.7 ロボット優先（2026-10-23 エイイチと決めた）：料理ができたら、まずロボットを待つ。スタッフは料理を持たず、 # ほかの仕事（案内・片付け・会計）を続ける。robot_first_wait_s（2分）待ってもロボットが空かなければ、 # そこからはロボットとスタッフのうち先に空いた方が運ぶ（料理が冷める前の逃げ道）。 # 割れ物を人が運ぶ設定（fragile_by=staff）と、ロボット0台のときは、今までどおりスタッフが運ぶ。 # v2.6 までは、はじめから「先に空いた方」だったため、受け渡し口の横で待つスタッフが多く運んでいた。 # ロボットが出るときは、受け渡し口で待っている料理を段の数までまとめて載せ、近い卓から回る。 def deliver(self, d, fragile=False, counted=True): env, p = self.env, self.p rec = {"d": d, "ready": env.now, "fragile": fragile, "counted": counted, "taken": env.event(), "arrived": env.event()} robot_ok = self.robots is not None and not (fragile and p["fragile_by"] == "staff") if not robot_ok: sreq = self.staff.request(priority=1) yield sreq yield env.process(self._staff_carry(rec, sreq)) return rreq = self.robots.request(priority=0) rec["sreq"], rec["rreq"] = None, rreq self.pass_q.append(rec) yield rreq | rec["taken"] | env.timeout(p["robot_first_wait_s"]) # まずロボットだけを待つ if not (rreq.triggered or rec["taken"].triggered): # 2分たった：ここからはスタッフも運べる sreq = self.staff.request(priority=1) rec["sreq"] = sreq yield rreq | sreq | rec["taken"] sreq = rec["sreq"] if rec["taken"].triggered: # ほかの料理と一緒にロボットに載った yield rec["arrived"] return self.pass_q.remove(rec) if rreq.triggered: # ロボットが運ぶ if sreq is not None: self._drop(sreq, self.staff) batch = [rec] for o in list(self.pass_q): if len(batch) >= p["trays"]: break self.pass_q.remove(o) self._drop(o["rreq"], self.robots) if o["sreq"] is not None: self._drop(o["sreq"], self.staff) o["taken"].succeed() batch.append(o) env.process(self._robot_run(batch, rreq)) yield rec["arrived"] return rreq.cancel() # スタッフが運ぶ yield env.process(self._staff_carry(rec, sreq)) @staticmethod def _drop(req, res): if req.triggered: res.release(req) else: req.cancel() def _robot_run(self, batch, rreq): env, p = self.env, self.p start = env.now batch.sort(key=lambda r: r["d"]) pos = 0.0 for i, rec in enumerate(batch): leg = rec["d"] - pos + (p["hop_m"] if i else 0) yield env.timeout(max(leg, 0) / p["robot_speed"]) if i == 0: yield env.process(self.maybe_stop(rec["d"])) if rec["counted"]: self.dish_waits.append((env.now - rec["ready"]) / 60) self.robot_trips += 1 self.fragile_robot += rec["fragile"] rec["arrived"].succeed() yield env.timeout(p["robot_handoff_s"]) pos = rec["d"] yield env.timeout(pos / p["robot_speed"]) self.robot_busy_s += env.now - start self.robots.release(rreq) def _staff_carry(self, rec, sreq): env, p = self.env, self.p self.staff_trips += 1 go = rec["d"] / p["staff_speed"] yield env.timeout(go) if rec["counted"]: self.dish_waits.append((env.now - rec["ready"]) / 60) rec["arrived"].succeed() yield env.timeout(p["staff_place_s"] + go) self.staff_walk_m += 2 * rec["d"] self.staff_busy_s += 2 * go + p["staff_place_s"] self.staff.release(sreq) # ---- 巡回回収（食事中にロボットが来て、客が空いた皿を載せる） def collect_visit(self, d, deadline, box): env, p = self.env, self.p req = self.robots.request(priority=2) timer = env.timeout(max(deadline - env.now, 0)) res = yield req | timer if req not in res: req.cancel() return start = env.now t = d / p["robot_speed"] back = (d + p["dish_extra_m"]) / p["robot_speed"] yield env.timeout(t) yield env.process(self.maybe_stop(d)) yield env.timeout(p["guest_load_s"] + back) self.robot_busy_s += env.now - start self.robots.release(req) self.collects += 1 box[0] += 1 # ---- 下げ膳（客が帰ってから卓が使えるようになるまで） def bus(self, d, collected, counted=True): env, p = self.env, self.p start = env.now trips = max(p["bus_trips"] - collected, 0) if p["mode"] == "bus" and self.robots is not None and trips > 0: # v2.4: スタッフの手が空いて片付けに向かうときに、空いているロボットがいれば一緒に卓へ向かわせる with self.staff.request(priority=2) as sreq: yield sreq s0 = env.now rreq = self.robots.request(priority=1) if rreq.triggered: self.robot_bus += 1 arrive, loaded = env.event(), env.event() env.process(self._robot_bus_trip(d, rreq, arrive, loaded)) # スタッフは卓まで行き、皿をまとめ、ロボットが着いたら載せ、卓を拭いて戻る yield env.timeout(d / p["staff_speed"] + p["clear_s"]) yield arrive yield env.timeout(p["load_s"]) loaded.succeed() # 載せ終えた。ロボットは洗い場へ yield env.timeout(p["wipe_s"]) done_at = env.now yield env.timeout(d / p["staff_speed"]) self.staff_walk_m += 2 * d self.staff_busy_s += env.now - s0 if counted: self.turns.append((done_at - start) / 60) return rreq.cancel() # 空いているロボットがいない：そのまま人だけで下げる（皿の量に応じて往復） self.staff_bus += 1 walk = 2 * (d + p["dish_extra_m"]) * trips tt = walk / p["staff_speed"] + p["clear_s"] + p["wipe_s"] self.staff_walk_m += walk self.staff_busy_s += tt yield env.timeout(tt) if counted: self.turns.append((env.now - start) / 60) return # 人だけで下げる（皿の量に応じて往復）。皿がもうなければ拭きに行くだけ self.staff_bus += 1 walk = 2 * (d + p["dish_extra_m"]) * trips if trips > 0 else 2 * d work = (p["clear_s"] if trips > 0 else 0) + p["wipe_s"] yield env.process(self.staff_job(2, walk, work)) if counted: self.turns.append((env.now - start) / 60) def _robot_bus_trip(self, d, rreq, arrive, loaded): env, p = self.env, self.p start = env.now t = d / p["robot_speed"] yield env.timeout(t) yield env.process(self.maybe_stop(d, helper_waiting=True)) arrive.succeed() yield loaded # v2.4: スタッフが載せ終えるまで卓で待つ back = (d + p["dish_extra_m"]) / p["robot_speed"] yield env.timeout(back + 15) # 洗い場で降ろすのは厨房側 self.robot_busy_s += env.now - start self.robots.release(rreq) # ---- 1組の流れ def group(self): env, p, rng = self.env, self.p, self.rng arrive = env.now counted = (self.t0 <= arrive < self.t1) if p["sustain"] else True # 数える時間に来た組だけを記録する if counted: self.counted += 1 # 図面のある店は卓ごとの距離から選ぶ。ほかの店は今までどおり（乱数の使い方も同じ） d = rng.choice(p["dist_list"]) if p.get("dist_list") else rng.uniform(*p["dist_m"]) self._acc() table = self.tables.request() yield table blocked = self.blocked.pop(table, 0.0) if counted: w = (env.now - arrive) / 60 self.seat_waits.append(w) self.seat_wait_by_arrival.append((arrive, w)) # 案内（v2.4）: スタッフの手が空いたら卓まで一緒に歩く。卓に着いた時点を「席に着いた」とする。 # 説明（guide_s）と受け渡し口への戻りは、スタッフの手がふさがる時間には入れ、お客さんの注文はそのあとに進む（v2.2 と同じ） g0 = env.now with self.staff.request(priority=1) as greq: yield greq gwait = env.now - g0 yield env.timeout(d / p["staff_speed"]) seated_at = env.now yield env.timeout(p["guide_s"] + d / p["staff_speed"]) self.staff_walk_m += 2 * d self.staff_busy_s += 2 * d / p["staff_speed"] + p["guide_s"] if counted: self.blocked_waits.append(blocked / 60) self.guide_waits.append(gwait / 60) self.entry_waits.append((seated_at - arrive) / 60) firsts = [env.process(self._cook_then_deliver(c, d, False, counted)) for c in p["first_dishes"]] yield simpy.AllOf(env, firsts) eat = tri(rng, p["eat_min"]) * 60 end = env.now + eat # 食事中の追加注文・呼び出し・巡回回収 extras = [] for _ in range(p["extra_orders"]): extras.append(env.process(self._order_later(rng.uniform(0, eat * 0.7), d, p["extra_cook"], False, counted))) n_fragile = int(p["fragile"]) + (rng.random() < p["fragile"] - int(p["fragile"])) for _ in range(n_fragile): extras.append(env.process(self._order_later(rng.uniform(0.3, 0.85) * eat, d, p["fragile_cook"], True, counted))) calls = sum(rng.random() < p["call_rate"] / 2 for _ in range(2)) for _ in range(calls): env.process(self._call_later(rng.uniform(0, eat), d)) collected = [0] if p["mode"] == "patrol" and self.robots is not None: for k in range(p["bus_trips"]): at = eat * (k + 1) / (p["bus_trips"] + 1) env.process(self._collect_later(at, d, end, collected)) yield env.timeout(eat) if extras: yield simpy.AllOf(env, extras) yield env.process(self.staff_job(1, 0, p["pay_s"])) self._acc() self.dirty += 1 # 客は帰った。片付けが済むまで卓は使えない yield env.process(self.bus(d, collected[0], counted)) self._acc() self.dirty -= 1 self.tables.release(table) self.done += 1 if counted: self.counted_done += 1 self._check_finished() def _cook_then_deliver(self, cook, d, fragile=False, counted=True): yield self.env.timeout(tri(self.rng, cook) * 60) yield self.env.process(self.deliver(d, fragile, counted)) def _order_later(self, delay, d, cook, fragile, counted=True): yield self.env.timeout(delay) yield self.env.process(self._cook_then_deliver(cook, d, fragile, counted)) def _call_later(self, delay, d): yield self.env.timeout(delay) yield self.env.process(self.staff_job(1, 2 * d, self.p["call_s"])) def _collect_later(self, delay, d, deadline, box): yield self.env.timeout(delay) yield self.env.process(self.collect_visit(d, deadline, box)) def arrivals(self): env, p, rng = self.env, self.p, self.rng rate = p["arrivals_per_h"] / 3600 while not self.finished.triggered: yield env.timeout(rng.expovariate(rate)) if not p["sustain"] and env.now > p["peak_min"] * 60: return # ピークの回し方：peak_min 分で来客が止まる env.process(self.group()) # 回るかの確かめ：来客は止めない（数える組が出きるまで） UNSTABLE_MIN = 5 # 参考：数える時間のうちに入店待ちがこれ以上延びていく条件（判定には使わない。卓が足りない店の見分けに使う） NG_MIN = 10 # 席は空いているのに待った時間（staff_wait_mean）がこれ以上なら「店が回らない」とする（分） STAFF_FULL = 0.95 # 回るかの確かめで、スタッフの手がこれ以上ふさがっていたら「店が回らない」とする def run_once(params, seed): store = STORES[params["store"]] p = {**COMMON, **store, **params} env = simpy.Environment() s = Store(env, random.Random(seed), p) env.process(s.arrivals()) if p["sustain"]: return _sustain_result(env, s, p) env.run() end = env.now w = s.dish_waits return { "dish_wait_mean": statistics.mean(w), "dish_wait_p90": sorted(w)[int(0.9 * (len(w) - 1))], "cold_share": sum(x > p["cold_min"] for x in w) / len(w), "seat_wait_mean": statistics.mean(s.seat_waits), "turn_mean": statistics.mean(s.turns) if s.turns else 0.0, "walk_km_per_staff": s.staff_walk_m / p["staff"] / 1000, "staff_util": s.staff_busy_s / (p["staff"] * end), "robot_util": (s.robot_busy_s / (p["robots"] * end)) if p["robots"] else 0.0, "robot_share": s.robot_trips / max(s.robot_trips + s.staff_trips, 1), "robot_bus_share": s.robot_bus / max(s.robot_bus + s.staff_bus, 1), "collects": s.collects, "rescues": s.rescues, "pauses": s.pauses, "fragile_robot": s.fragile_robot, "groups": s.done, # 入店待ちの内訳（v2.2） "entry_wait_mean": statistics.mean(s.entry_waits), # 来てから案内されて卓に着くまで（v2.4） "blocked_wait_mean": statistics.mean(s.blocked_waits), # 席はあるのに片付けが終わらず待った "guide_wait_mean": statistics.mean(s.guide_waits), # 卓は決まったが案内の手が空かず待った "staff_wait_mean": statistics.mean(b + g for b, g in zip(s.blocked_waits, s.guide_waits)), # 席はあるのに待った（合計） "full_wait_mean": statistics.mean(w - b for w, b in zip(s.seat_waits, s.blocked_waits)), # 席が埋まっていて待った } def _sustain_result(env, s, p): """回るかの確かめ：来客を止めずに回し、数える時間の結果を出す""" env.process(s.window()) env.run(until=s.finished) win_s = p["peak_min"] * 60 q = win_s / 4 first = [x for t, x in s.seat_wait_by_arrival if t < s.t0 + q] last = [x for t, x in s.seat_wait_by_arrival if t >= s.t1 - q] trend = (statistics.mean(last) - statistics.mean(first)) if first and last else 0.0 return { "seat_wait_mean": statistics.mean(s.seat_waits), "seat_wait_trend": trend, "staff_util": statistics.mean(s.staff_busy_n) / p["staff"], "capped": 1.0 if s.capped else 0.0, } def conditions(main_only=False): for key, st in STORES.items(): staffs = st["staff"] if main_only else st["staff_all"] robots_list = ROBOTS if main_only else ROBOTS_ALL for staff, robots, arr, mode, trays, fb in product(staffs, robots_list, st["arrivals"], MODES, TRAYS, FRAGILE_BY): if robots == 0 and (mode != "serve" or trays != 1 or fb != "any"): continue # ロボットなしなら、使い方・段の数・使い分けは関係ない yield {"store": key, "staff": staff, "robots": robots, "arrivals_per_h": arr, "mode": mode, "trays": trays, "fragile_by": fb} def run_condition(params): runs = [run_once(params, seed) for seed in range(REPS)] agg = {k: round(statistics.mean(r[k] for r in runs), 4) for k in runs[0]} return {**params, **agg} def key(c): return (c["store"], c["staff"], c["robots"], c["arrivals_per_h"], c["mode"], c["trays"], c["fragile_by"]) def model_sig(common, stores): """結果に効く前提だけを取り出す（人数・台数の選択肢や名前は含めない）""" skip = {"label", "staff", "staff_all", "arrivals"} return json.loads(json.dumps({"common": common, "reps": REPS, "stores": {k: {a: b for a, b in v.items() if a not in skip} for k, v in stores.items()}})) if __name__ == "__main__": if "--one" in sys.argv: for mode in MODES: for trays, fb in product(TRAYS, FRAGILE_BY): r = run_condition({"store": "family", "staff": 3, "robots": 2, "arrivals_per_h": 35, "mode": mode, "trays": trays, "fragile_by": fb}) print(mode, trays, fb, {k: round(v, 2) for k, v in r.items() if isinstance(v, float)}) elif "--sustain" in sys.argv: # 回るかの確かめ：全条件を、来客を止めない回し方で回す conds = [{**c, "sustain": True} for c in conditions()] done = {} try: with open("results_sustain_partial.json", encoding="utf-8") as f: old = json.load(f) if old.get("common") == COMMON and old.get("reps") == REPS and old.get("unstable_min") == UNSTABLE_MIN: done = {key(r): r for r in old["rows"]} except (FileNotFoundError, KeyError, TypeError, json.JSONDecodeError): pass def save_s(path, rows): out = {"common": COMMON, "reps": REPS, "unstable_min": UNSTABLE_MIN, "rows": rows} with open(path + ".tmp", "w", encoding="utf-8") as f: json.dump(out, f, ensure_ascii=False) import os os.replace(path + ".tmp", path) todo = [c for c in conds if key(c) not in done] print(f"{len(conds)} conditions, {len(conds) - len(todo)} reused, {len(todo)} to run", flush=True) with Pool() as pool: for i, r in enumerate(pool.imap_unordered(run_condition, todo, chunksize=2), 1): done[key(r)] = r if i % 100 == 0: save_s("results_sustain_partial.json", list(done.values())) print(f" {i}/{len(todo)}", flush=True) save_s("results_sustain_partial.json", list(done.values())) save_s("results_sustain.json", [done[key(c)] for c in conds]) print(len(conds), "conditions written", flush=True) else: conds = list(conditions(main_only="--main" in sys.argv)) sig = model_sig(COMMON, STORES) # 前提が同じなら、前回の結果と途中経過を使い回して足りない条件だけ回す cache = {} for path in ("results.json", "results_partial.json"): try: with open(path, encoding="utf-8") as f: old = json.load(f) if old.get("reps") == REPS: osig = model_sig(old["common"], old["stores"]) same = {k for k in sig["stores"] if osig["common"] == sig["common"] and osig["stores"].get(k) == sig["stores"][k]} cache.update({key(r): r for r in old["rows"] if r["store"] in same}) except (FileNotFoundError, KeyError, TypeError, json.JSONDecodeError): pass def save(path, rows): out = {"common": COMMON, "stores": STORES, "robots": ROBOTS, "robots_all": ROBOTS_ALL, "modes": MODES, "trays": TRAYS, "fragile_by": FRAGILE_BY, "reps": REPS, "rows": rows} with open(path + ".tmp", "w", encoding="utf-8") as f: json.dump(out, f, ensure_ascii=False) import os os.replace(path + ".tmp", path) todo = [c for c in conds if key(c) not in cache] print(f"{len(conds)} conditions, {len(conds) - len(todo)} reused, {len(todo)} to run", flush=True) with Pool() as pool: for i, r in enumerate(pool.imap_unordered(run_condition, todo, chunksize=2), 1): cache[key(r)] = r if i % 100 == 0: save("results_partial.json", list(cache.values())) print(f" {i}/{len(todo)}", flush=True) save("results_partial.json", list(cache.values())) save("results.json", [cache[key(c)] for c in conds]) print(len(conds), "conditions written", flush=True)