Skip to content

workers / wine_inventory / src / pince_shelf / cli / stock_runtime.py

Path: workers/wine_inventory/src/pince_shelf/cli/stock_runtime.py
Lane: workers
Executable: yes

Purpose

Run the end-to-end stock detection pipeline and write a JSON run summary.

Execution order:

0) tapo_turn_on_and_settle (non-fatal): request light ON via pince_shelf.utils.tapo_manager.switch_light_on, then sleep for get_light_stabilize_seconds(cfg) when successful. 1) take_clean_snapshot.run(cfg, cam=0, init_templates=False) 2) crop_with_json.run(cfg) 3) onnx_batch.run(cfg) 4) DB_compare.run(cfg)

Finalization:

  • tapo_turn_off is always attempted in finally (non-fatal).
  • A summary is always written to <cfg.paths["det_out"]>/stock_runtime_summary.json, including per-step timing, success/failure flags, and top-level ok.

Behavior notes:

  • No CLI arguments; configuration is loaded from CONF_DIR/pince_shelf.ini.
  • run() never raises outward: it captures failures and returns a JSON-serializable dict with ok=False and error details.
  • main() maps run()["ok"] to process exit code (0 on success, 1 on failure).
  • stdout/stderr are mirrored into the runtime log so terminal-style output from downstream steps also lands in pipeline_runtime.log.

Entry points

  • __main__ guard detected

Key symbols

Classes

  • (none detected)

Functions

  • _now_ts()
  • _turn_light_on_and_settle(cfg)
  • main()
  • run()

Configuration & environment

  • (no env vars detected by regex)

Dependencies (imports)

  • __future__
  • json
  • logging
  • pathlib
  • pince_shelf.config.settings
  • pince_shelf.utils.logger
  • pince_shelf.utils.paths
  • pince_shelf.utils.tapo_manager
  • pince_shelf.vision
  • time
  • typing

Operational notes (fill in)

  • Headless/interactive:
  • Systemd unit (if any):
  • Schedules/triggers:
  • Debug artifacts:
  • Failure modes: