OPi: store-and-forward outbox для /api/capture (переживает обрыв сети)
- outbox.py: SQLite-очередь + JPEG на диске, at-least-once доставка - scales.py: capture_and_send кладёт в очередь вместо прямого POST, heartbeat_loop каждые 30с дренирует очередь, статус публикует outbox_pending - fw_version -> 2026-07-02-outbox
This commit is contained in:
Alexey Pavlov
@@ -0,0 +1,189 @@
|
||||
"""
|
||||
outbox.py — надёжная доставка взвешиваний на сервер (store-and-forward).
|
||||
|
||||
При обрыве сети OPi↔сервер события не теряются: кладём JPEG на диск + метаданные
|
||||
в SQLite, и досылаем при восстановлении связи. Сервер дедупит по event_id (upsert).
|
||||
Семантика: at-least-once (повторная отправка безопасна).
|
||||
|
||||
Зависимости: только stdlib (sqlite3, urllib, json).
|
||||
"""
|
||||
|
||||
import os
|
||||
import time
|
||||
import base64
|
||||
import sqlite3
|
||||
import threading
|
||||
import logging
|
||||
import json
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
|
||||
log = logging.getLogger("outbox")
|
||||
|
||||
|
||||
class CaptureOutbox:
|
||||
def __init__(self, server_url, device_id,
|
||||
db_path="/root/scales_outbox/outbox.db",
|
||||
photo_dir="/root/scales_outbox/photos",
|
||||
timeout=15, max_attempts=20):
|
||||
self.url = server_url.rstrip("/") + "/api/capture"
|
||||
self.device_id = device_id
|
||||
self.timeout = timeout
|
||||
self.max_attempts = max_attempts
|
||||
self.photo_dir = photo_dir
|
||||
|
||||
os.makedirs(os.path.dirname(db_path), exist_ok=True)
|
||||
os.makedirs(photo_dir, exist_ok=True)
|
||||
|
||||
self._lock = threading.Lock() # защита БД (heartbeat + main)
|
||||
self._flush_lock = threading.Lock() # одна flush за раз
|
||||
self._db = sqlite3.connect(db_path, check_same_thread=False)
|
||||
self._db.execute("PRAGMA journal_mode=WAL")
|
||||
self._db.execute("""
|
||||
CREATE TABLE IF NOT EXISTS outbox (
|
||||
event_id TEXT PRIMARY KEY,
|
||||
weight REAL,
|
||||
weight_raw TEXT,
|
||||
captured_at TEXT,
|
||||
photo_path TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
attempts INTEGER NOT NULL DEFAULT 0,
|
||||
status TEXT NOT NULL DEFAULT 'pending' -- pending | failed
|
||||
)""")
|
||||
self._db.commit()
|
||||
self._reconcile()
|
||||
|
||||
def _reconcile(self):
|
||||
"""После рестарта убрать осиротевшие JPEG."""
|
||||
with self._lock:
|
||||
known = {r[0] for r in self._db.execute("SELECT photo_path FROM outbox")}
|
||||
for name in os.listdir(self.photo_dir):
|
||||
p = os.path.join(self.photo_dir, name)
|
||||
if p not in known:
|
||||
try:
|
||||
os.remove(p)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
# ── публичный API ─────────────────────────────────────────────
|
||||
def enqueue(self, event_id, weight, weight_raw, captured_at, jpeg_bytes):
|
||||
"""Положить взвешивание в очередь. JPEG на диск, метаданные в БД."""
|
||||
photo_path = os.path.join(self.photo_dir, f"{event_id}.jpg")
|
||||
with open(photo_path, "wb") as f:
|
||||
f.write(jpeg_bytes)
|
||||
with self._lock:
|
||||
self._db.execute(
|
||||
"INSERT OR REPLACE INTO outbox "
|
||||
"(event_id, weight, weight_raw, captured_at, photo_path, created_at, attempts, status) "
|
||||
"VALUES (?,?,?,?,?,?,0,'pending')",
|
||||
(event_id, weight, weight_raw, captured_at, photo_path,
|
||||
time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())))
|
||||
self._db.commit()
|
||||
log.info("enqueued %s (%.1f kg)", event_id, weight)
|
||||
|
||||
def pending_count(self):
|
||||
"""Количество событий в очереди."""
|
||||
with self._lock:
|
||||
return self._db.execute(
|
||||
"SELECT COUNT(*) FROM outbox WHERE status='pending'").fetchone()[0]
|
||||
|
||||
def flush(self):
|
||||
"""Досылает pending-события (старые первыми). Вернёт (отправлено, осталось)."""
|
||||
if not self._flush_lock.acquire(blocking=False):
|
||||
return (0, self.pending_count())
|
||||
sent = 0
|
||||
try:
|
||||
while True:
|
||||
with self._lock:
|
||||
row = self._db.execute(
|
||||
"SELECT event_id, weight, weight_raw, captured_at, photo_path, attempts "
|
||||
"FROM outbox WHERE status='pending' ORDER BY rowid LIMIT 1").fetchone()
|
||||
if row is None:
|
||||
break
|
||||
event_id, weight, weight_raw, captured_at, photo_path, attempts = row
|
||||
try:
|
||||
with open(photo_path, "rb") as f:
|
||||
b64 = base64.b64encode(f.read()).decode()
|
||||
except OSError:
|
||||
self._drop(event_id)
|
||||
continue
|
||||
|
||||
payload = json.dumps({
|
||||
"event_id": event_id,
|
||||
"weight": weight,
|
||||
"weight_raw": weight_raw,
|
||||
"device_id": self.device_id,
|
||||
"captured_at": captured_at,
|
||||
"photo_b64": b64,
|
||||
"media_type": "image/jpeg",
|
||||
}).encode()
|
||||
|
||||
try:
|
||||
req = urllib.request.Request(
|
||||
self.url,
|
||||
data=payload,
|
||||
headers={"Content-Type": "application/json"})
|
||||
with urllib.request.urlopen(req, timeout=self.timeout) as r:
|
||||
resp = r.read().decode()
|
||||
if _ok_response(resp):
|
||||
self._delete(event_id)
|
||||
sent += 1
|
||||
log.info("delivered %s", event_id)
|
||||
else:
|
||||
log.warning("server rejected %s: %s", event_id, resp[:100])
|
||||
self._park(event_id)
|
||||
except urllib.error.HTTPError as e:
|
||||
if 400 <= e.code < 500:
|
||||
log.error("rejected %s: %d", event_id, e.code)
|
||||
self._park(event_id)
|
||||
else:
|
||||
self._bump(event_id, attempts)
|
||||
break
|
||||
except (urllib.error.URLError, OSError, TimeoutError) as e:
|
||||
log.warning("network unavailable (%s), pending=%d",
|
||||
type(e).__name__, self.pending_count())
|
||||
break
|
||||
finally:
|
||||
self._flush_lock.release()
|
||||
return (sent, self.pending_count())
|
||||
|
||||
# ── внутреннее ────────────────────────────────────────────────
|
||||
def _delete(self, event_id):
|
||||
with self._lock:
|
||||
row = self._db.execute("SELECT photo_path FROM outbox WHERE event_id=?",
|
||||
(event_id,)).fetchone()
|
||||
self._db.execute("DELETE FROM outbox WHERE event_id=?", (event_id,))
|
||||
self._db.commit()
|
||||
if row:
|
||||
try:
|
||||
os.remove(row[0])
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def _drop(self, event_id):
|
||||
"""Удалить событие (фото потеряно)."""
|
||||
with self._lock:
|
||||
self._db.execute("DELETE FROM outbox WHERE event_id=?", (event_id,))
|
||||
self._db.commit()
|
||||
|
||||
def _park(self, event_id):
|
||||
"""Паркировать событие (не отправлять больше)."""
|
||||
with self._lock:
|
||||
self._db.execute("UPDATE outbox SET status='failed' WHERE event_id=?", (event_id,))
|
||||
self._db.commit()
|
||||
|
||||
def _bump(self, event_id, attempts):
|
||||
"""Увеличить счётчик попыток, паркировать если превышен лимит."""
|
||||
status = "failed" if attempts + 1 >= self.max_attempts else "pending"
|
||||
with self._lock:
|
||||
self._db.execute("UPDATE outbox SET attempts=attempts+1, status=? WHERE event_id=?",
|
||||
(status, event_id))
|
||||
self._db.commit()
|
||||
|
||||
|
||||
def _ok_response(text):
|
||||
"""Проверить, что в ответе есть 'ok': true."""
|
||||
try:
|
||||
return bool(json.loads(text).get("ok"))
|
||||
except Exception:
|
||||
return False
|
||||
Reference in New Issue
Block a user