109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
"""Persistent mock implements the documented upload-immediately-activates behavior."""
|
|
|
|
import json
|
|
|
|
from PIL import Image
|
|
|
|
from pigway_frame_flow.errors import ErrorCode as E
|
|
from pigway_frame_flow.errors import FlowError
|
|
from pigway_frame_flow.storage.database import now_iso
|
|
|
|
|
|
class MockClient:
|
|
def __init__(self, root):
|
|
self.root = root
|
|
self.path = root / "mock-device.json"
|
|
if self.path.exists():
|
|
self.state = json.loads(self.path.read_text())
|
|
else:
|
|
self.state = {
|
|
"current": 1,
|
|
"counter": 1,
|
|
"images": [
|
|
{"id": 1, "name": "Original Art.jpg", "createdAt": now_iso()}
|
|
],
|
|
}
|
|
img = Image.new("RGB", (1200, 1600), "#efe2ca")
|
|
img.save(root / "mock-1.jpg")
|
|
self.save()
|
|
self.online = True
|
|
self.synced = True
|
|
self.uploads = 0
|
|
self.displays = []
|
|
|
|
def save(self):
|
|
temporary = self.path.with_suffix(".tmp")
|
|
temporary.write_text(json.dumps(self.state))
|
|
temporary.replace(self.path)
|
|
|
|
def get_device(self):
|
|
return {
|
|
"nickname": "客厅相框",
|
|
"model": "13.3-inch",
|
|
"modelCode": "B101",
|
|
"firmware": "MOCK-B101",
|
|
"online": self.online,
|
|
"lastHeartbeat": now_iso(),
|
|
"batteryLevel": 40,
|
|
"isCharging": False,
|
|
"screenWidth": 1200,
|
|
"screenHeight": 1600,
|
|
"screenRotate": 270,
|
|
"timezone": "Asia/Shanghai",
|
|
"switchType": "queue",
|
|
"switchMinute": 60,
|
|
"sdUsedSize": 123456,
|
|
"imageId": self.state["current"],
|
|
"coverId": None,
|
|
"synced": self.synced,
|
|
}
|
|
|
|
def list_images(self):
|
|
return {
|
|
"total": len(self.state["images"]),
|
|
"list": [
|
|
dict(x, current=x["id"] == self.state["current"])
|
|
for x in reversed(self.state["images"])
|
|
],
|
|
}
|
|
|
|
def upload(self, path, name):
|
|
self.uploads += 1
|
|
self.state["counter"] += 1
|
|
image_id = self.state["counter"]
|
|
(self.root / f"mock-{image_id}.jpg").write_bytes(path.read_bytes())
|
|
self.state["images"].append(
|
|
{"id": image_id, "name": name, "createdAt": now_iso()}
|
|
)
|
|
self.state["current"] = image_id
|
|
self.save()
|
|
return image_id
|
|
|
|
def display(self, image_id):
|
|
if image_id not in [x["id"] for x in self.state["images"]]:
|
|
raise FlowError(E.NOT_FOUND, "Image not found", 404)
|
|
self.displays.append(image_id)
|
|
self.state["current"] = image_id
|
|
self.save()
|
|
|
|
def delete(self, image_id):
|
|
self.state["images"] = [x for x in self.state["images"] if x["id"] != image_id]
|
|
if self.state["current"] == image_id:
|
|
self.state["current"] = (
|
|
self.state["images"][-1]["id"] if self.state["images"] else None
|
|
)
|
|
(self.root / f"mock-{image_id}.jpg").unlink(missing_ok=True)
|
|
self.save()
|
|
|
|
def download(self, image_id):
|
|
path = self.root / f"mock-{image_id}.jpg"
|
|
if not path.exists():
|
|
raise FlowError(E.NOT_FOUND, "Image not found", 404)
|
|
return path.read_bytes()
|
|
|
|
def display_by_name(self, name):
|
|
matches = [x for x in self.state["images"] if x["name"] == name]
|
|
if not matches:
|
|
raise FlowError(E.NOT_FOUND, "Image not found", 404)
|
|
return self.display(matches[-1]["id"])
|