105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
from pigway_frame_flow.models import (
|
|
CurrentContent,
|
|
Device,
|
|
DisplayGeometry,
|
|
GalleryImage,
|
|
Power,
|
|
)
|
|
|
|
|
|
class KokonnaDriver:
|
|
def __init__(self, client, device_id, sanitize=lambda v: v):
|
|
self.client, self.device_id, self.sanitize = client, device_id, sanitize
|
|
self.cached: Device | None = None
|
|
self.cached_at = 0.0
|
|
|
|
def device(self, refresh=False):
|
|
import time
|
|
|
|
if not refresh and self.cached and time.monotonic() - self.cached_at < 60:
|
|
return self.cached
|
|
d = self.sanitize(self.client.get_device())
|
|
result = Device(
|
|
id=self.device_id,
|
|
name=d.get("nickname", "KoKonna"),
|
|
status="online" if d.get("online") else "offline",
|
|
model=d.get("model", "13.3-inch"),
|
|
model_code=d.get("modelCode", "B101"),
|
|
firmware=d.get("firmware"),
|
|
display=DisplayGeometry(
|
|
**{
|
|
"width": d.get("screenWidth", 1200),
|
|
"height": d.get("screenHeight", 1600),
|
|
"rotation": d.get("screenRotate", 270),
|
|
}
|
|
),
|
|
power=Power(
|
|
**{
|
|
"battery_percent": d.get("batteryLevel"),
|
|
"charging": bool(d.get("isCharging")),
|
|
}
|
|
),
|
|
current_content=CurrentContent(
|
|
**{
|
|
"id": d.get("imageId"),
|
|
"cover_id": d.get("coverId"),
|
|
"synced": bool(d.get("synced")),
|
|
}
|
|
),
|
|
slideshow={
|
|
"mode": d.get("switchType"),
|
|
"interval_minutes": d.get("switchMinute"),
|
|
"members_available": False,
|
|
"editable": False,
|
|
},
|
|
storage_used_bytes=d.get("sdUsedSize"),
|
|
model_verified=bool(d.get("model") and d.get("modelCode")),
|
|
last_seen_at=d.get("lastHeartbeat"),
|
|
timezone=d.get("timezone", "Asia/Shanghai"),
|
|
)
|
|
|
|
self.cached = result
|
|
self.cached_at = time.monotonic()
|
|
return result
|
|
|
|
def get_current_image(self):
|
|
return self.device().current_content.id
|
|
|
|
def health(self):
|
|
return bool(self.device())
|
|
|
|
def invalidate(self):
|
|
self.cached_at = 0.0
|
|
|
|
def display_by_name(self, name):
|
|
return self.client.display_by_name(name)
|
|
|
|
def images(self):
|
|
data = self.sanitize(self.client.list_images())
|
|
return [
|
|
GalleryImage(
|
|
image_id=d["id"],
|
|
width=d.get("width"),
|
|
height=d.get("height"),
|
|
size_bytes=d.get("size"),
|
|
media_type=d.get("type"),
|
|
name=d.get("name", "image"),
|
|
created_at=d.get("createdAt"),
|
|
current=bool(d.get("current")),
|
|
url=f"/api/v1/devices/{self.device_id}/images/{d['id']}/file",
|
|
)
|
|
for d in data.get("list", [])
|
|
]
|
|
|
|
def upload(self, path, name):
|
|
return self.client.upload(path, name)
|
|
|
|
def display(self, image_id):
|
|
return self.client.display(image_id)
|
|
|
|
def delete(self, image_id):
|
|
return self.client.delete(image_id)
|
|
|
|
def download(self, image_id):
|
|
return self.client.download(image_id)
|