diff --git a/.site/gallery/index.html b/.site/gallery/index.html index fa9db9a0..97156d12 100644 --- a/.site/gallery/index.html +++ b/.site/gallery/index.html @@ -366,6 +366,16 @@

Fonts

Framebuf Simpletest

Simple test example to demonstrate pygraphics.FrameBuffer.

+ +
+ + + lvgl + +
+

Google Photos

+

Pick Google Photos on your phone; the board lists, views, and slideshows them.

+
diff --git a/.site/gallery/python-files.json b/.site/gallery/python-files.json index c3743319..71ca2130 100644 --- a/.site/gallery/python-files.json +++ b/.site/gallery/python-files.json @@ -52,6 +52,13 @@ "feathers.py", "fonts.py", "framebuf_simpletest.py", + "google_photos/__init__.py", + "google_photos/__main__.py", + "google_photos/assets/gen_sim_assets.py", + "google_photos/google_photos.py", + "google_photos/gphotos_engine.py", + "google_photos/gphotos_lvgl.py", + "google_photos/gphotos_sim.py", "graphics_simpletest.py", "hello.py", "joystick_list_select.py", diff --git a/.site/pyscript/index.html b/.site/pyscript/index.html index 49057a04..0fee8eed 100644 --- a/.site/pyscript/index.html +++ b/.site/pyscript/index.html @@ -366,6 +366,16 @@

Fonts

Framebuf Simpletest

Simple test example to demonstrate pygraphics.FrameBuffer.

+ +
+ + + lvgl + +
+

Google Photos

+

Pick Google Photos on your phone; the board lists, views, and slideshows them.

+
diff --git a/.site/pyscript/thumbnails/google_photos.png b/.site/pyscript/thumbnails/google_photos.png new file mode 100644 index 00000000..fcf9f6ce Binary files /dev/null and b/.site/pyscript/thumbnails/google_photos.png differ diff --git a/docs/screenshots/README.md b/docs/screenshots/README.md index 4f37478b..058cd534 100644 --- a/docs/screenshots/README.md +++ b/docs/screenshots/README.md @@ -4,4 +4,5 @@ | ![color_test.png](color_test.png) | ![console_advanced_demo.gif](console_advanced_demo.gif) | ![displaybuf_simpletest.png](displaybuf_simpletest.png) | | ![fonts.png](fonts.png) | ![framebuf_simpletest.png](framebuf_simpletest.png) | ![noto_fonts.png](noto_fonts.png) | | ![paint.png](paint.png) | ![proverbs.png](proverbs.png) | ![testris.png](testris.png) | -| ![tiny_toasters.gif](tiny_toasters.gif) | ![piano.png](piano.png) | | \ No newline at end of file +| ![tiny_toasters.gif](tiny_toasters.gif) | ![piano.png](piano.png) | ![google_photos.png](google_photos.png) | +| ![google_photos_view.png](google_photos_view.png) | ![google_photos_connect.png](google_photos_connect.png) | | \ No newline at end of file diff --git a/docs/screenshots/google_photos.png b/docs/screenshots/google_photos.png new file mode 100644 index 00000000..ac8e1d14 Binary files /dev/null and b/docs/screenshots/google_photos.png differ diff --git a/docs/screenshots/google_photos_connect.png b/docs/screenshots/google_photos_connect.png new file mode 100644 index 00000000..e56f58ab Binary files /dev/null and b/docs/screenshots/google_photos_connect.png differ diff --git a/docs/screenshots/google_photos_view.png b/docs/screenshots/google_photos_view.png new file mode 100644 index 00000000..bdb0f452 Binary files /dev/null and b/docs/screenshots/google_photos_view.png differ diff --git a/lib/examples/google_photos/README.md b/lib/examples/google_photos/README.md new file mode 100644 index 00000000..9c63c641 --- /dev/null +++ b/lib/examples/google_photos/README.md @@ -0,0 +1,170 @@ +# google_photos — Google Photos on a PyDevices display + +Pick photos from your phone, browse them as a thumbnail list on the board, +view them full screen, or run a slideshow. LVGL front end; runs on +MicroPython, CircuitPython, and CPython wherever a `board_config` and the +PyDevices LVGL binding are available. + +| List | View | Pairing | +|:--:|:--:|:--:| +| ![list](../../../docs/screenshots/google_photos.png) | ![view](../../../docs/screenshots/google_photos_view.png) | ![connect](../../../docs/screenshots/google_photos_connect.png) | + +```bash +cd lib +micropython -m examples.google_photos # or python / circuitpython +``` + +Without a tokens file (below) the launcher runs a **simulator** with sample +photos, so the UI can be explored anywhere — including the +[PyScript gallery](https://PyDevices.github.io/pydevices-examples/pyscript/). +Set `GPHOTOS_SIM=1` to force it, `GPHOTOS_SIM=0` to force the real client. + +## How it works + +Since March 31, 2025 the Google Photos *Library* API only returns media that +the calling app uploaded itself, so a third-party app can no longer browse a +whole library. The self-serve path is the **Picker API**: + +1. The device creates a *picker session* and shows its link as a QR code. +2. You scan it with your phone, choose photos in Google Photos, tap **Done**. +3. The device polls the session, lists the picked items, and downloads + sized thumbnails (list tiles are exact-size center crops, the viewer asks + for an image that fits the panel). + +Google returns only what you picked; nothing else in the library is visible +to the device. Picks persist on the device (see *Files*) so a reboot lands on +the list, and thumbnails are cached on flash so they download once. + +The Google login cannot run on the microcontroller: Google's TV +"enter this code on your phone" flow refuses the Picker scope, and a LAN +address cannot be registered as an OAuth redirect. So the browser login runs +**once on a PC** (`tools/gphotos_auth.py`) and its output — a refresh token — +is copied to the board. From then on the device only trades that refresh +token for hour-long access tokens. + +## Setup (one time) + +1. **Google Cloud project** — : create a + project (any name), then *APIs & Services → Library → enable + **Google Photos Picker API***. +2. **OAuth consent screen** — user type *External*. Fill in the app name and + your e-mail. Then either **publish to Production** (an unverified app is + fine for personal use; Google shows a warning you click through once) or + stay in *Testing* and add your account as a test user. Testing refresh + tokens expire after **7 days**, so Production is the one that keeps + working. A Google Workspace account can use *Internal* instead. +3. **Credentials** — *Create credentials → OAuth client ID → Desktop app*. + Download the JSON (`client_secret_….json`). +4. **Sign in on the PC** (CPython 3.8+, stdlib only): + + ```bash + python tools/gphotos_auth.py --client-secrets ~/Downloads/client_secret_XXX.json --verify + ``` + + A browser opens (or copy the printed URL). Sign in with the account whose + photos you want to show. The script writes `~/.gphotos_tokens.json` + (owner-only permissions) and, with `--verify`, refreshes the token and + creates + deletes a picker session to prove the API is enabled. +5. **Copy to the board** (join Wi-Fi first on MicroPython/CircuitPython): + + ```bash + mpremote cp ~/.gphotos_tokens.json :/gphotos_tokens.json + ``` + + Desktop runs read `~/.gphotos_tokens.json` directly. `GPHOTOS_TOKENS` + overrides the path. + +Then run the example, tap **PICK**, scan the code with your phone, pick +photos, tap **Done** — the list appears on the board. + +## Using it + +| Page | Buttons | +|---|---| +| **list** | **PICK** new session · **SLIDES** slideshow from the first photo · **MORE** next page (long picks are paged to keep RAM bounded) · tap a row to view it | +| **view** | **BACK** · **PREV** · **PLAY / PAUSE** slideshow · **NEXT** · tapping the photo also advances | +| **connect** | QR code of the session link · **OPEN** (desktop only: opens the link in the local browser) · **RETRY** new session · **BACK** to the list | + +The slideshow interval is `slideshow_s` in the prefs file (default 5). + +## Files + +| Desktop | Microcontroller | Contents | +|---|---|---| +| `~/.gphotos_tokens.json` | `/gphotos_tokens.json` | `client_id`, `client_secret`, `refresh_token` (written by `tools/gphotos_auth.py`; read-only on the device) | +| `~/.gphotos_prefs` | `/gphotos_prefs` | JSON: current session, picked items (without their short-lived URLs), `slideshow_s` | +| `~/.gphotos_cache/` | `/gphotos_cache/` | `_x[c].jpg\|png` thumbnails, at most 48 files | + +Override with `GPHOTOS_TOKENS`, `GPHOTOS_PREFS`, `GPHOTOS_CACHE`. + +## Images on each platform + +Google serves JPEG (PNG for PNG originals). LVGL decodes: + +| Runtime | JPEG | PNG | +|---|---|---| +| CPython (`pydevices-lvgl`) | LVGL's built-in TJPGD | LODEPNG | +| MicroPython LVGL firmware (lvgl-micropython + displayif) | displayif's `jpegio`, registered as an LVGL image decoder ([displayif#23](https://github.com/PyDevices/displayif/issues/23)) | LODEPNG | +| CircuitPython LVGL firmware | CircuitPython's `jpegio` via lvgl-circuitpython | LODEPNG | + +Without a JPEG decoder the list still works: tiles keep a placeholder and +the viewer explains why. Both decoders hold the decoded image +(`width × height × 2` bytes) while it is drawn, so the viewer requests an +image no larger than the panel; a 320×480 panel needs ~300 KB of heap for +the full-screen photo — comfortable with PSRAM, tight without. Thumbnails +are tiny. Only one page of tiles is resident at a time. + +## Limits and gotchas + +- **Thumbnail URLs expire after 60 minutes.** The engine re-lists the + session (one request) to refresh them; nothing is re-picked. +- **Picker sessions expire** (about a week). After that the cached list still + shows, but new downloads fail with *session expired* until you **PICK** + again. The prefs keep the picked list either way. +- **Refresh tokens die after 7 days while the consent screen is in + *Testing*.** Publish to Production (see Setup) and they last until revoked. +- **`invalid_grant` on start** means the refresh token was revoked or expired: + run `tools/gphotos_auth.py` again and copy the new file. +- **`HTTP 403 … SERVICE_DISABLED`** means the Picker API is not enabled in + the project (Setup step 1). **`access_denied` in the browser** means your + account is not a test user of a Testing app. +- **TLS on MicroPython** does not verify Google's certificate (the platform + default: boards ship no CA bundle). The refresh token is the secret to + protect — treat the tokens file like a password. +- **`-m` launches** (`python -m examples.google_photos`) block inside the + launcher's `app.run()` because the shared app host-loop declines `-m` + entry points; a script-path launch (`python examples/google_photos/google_photos.py`) + keeps itself alive like every other example. +- Videos appear in the list with a *video* tag; the viewer shows their poster + frame. + +## Module layout + +``` +google_photos/ + google_photos.py launcher: sim detection, restore prefs, start LVGL front end + gphotos_engine.py Picker API client, token refresh, prefs, thumbnail cache (no UI) + gphotos_sim.py simulator (sample photos) + make_engine() + gphotos_lvgl.py LVGL front end: connect (QR), list, view, slideshow + assets/ sim_NN_*.jpg sample photos + gen_sim_assets.py (Pillow) +tools/gphotos_auth.py PC-side one-time Google sign-in (stdlib, PKCE, loopback) +tests/test_gphotos_engine.py +``` + +`gphotos_engine` has no display, event, or UI imports, so a pdwidgets or +pygraphics front end can be added the way `roku_remote` has three. All +network calls are blocking and are queued on the LVGL pump from the main +thread (no `_thread`, like the Roku example). + +## Testing + +```bash +.venv/bin/python -m unittest discover -s tests -p "test_gphotos_engine.py" +SDL_VIDEODRIVER=dummy SDL_AUDIODRIVER=dummy .venv/bin/python tools/example_test_kit.py \ + --no-unit-tests --only-interpreter cpython-venv --only-example google_photos gphotos_lvgl +``` + +The unit tests fake every HTTP call (token refresh, sessions, paging, +downloads, the raw-socket fallback, the MicroPython `requests` path) and the +login tool's loopback redirect. The kit run exercises the real LVGL UI in +simulator mode. diff --git a/lib/examples/google_photos/__init__.py b/lib/examples/google_photos/__init__.py new file mode 100644 index 00000000..1163223b --- /dev/null +++ b/lib/examples/google_photos/__init__.py @@ -0,0 +1,7 @@ +import sys + +_wd = __file__.replace("\\", "/") +_wd = _wd.rsplit("/", 1)[0] if "/" in _wd else "." +if _wd not in sys.path: + sys.path.insert(0, _wd) +from . import google_photos # noqa: F401 — gallery/kit entry: import google_photos diff --git a/lib/examples/google_photos/__main__.py b/lib/examples/google_photos/__main__.py new file mode 100644 index 00000000..8d65a8b4 --- /dev/null +++ b/lib/examples/google_photos/__main__.py @@ -0,0 +1,13 @@ +"""Entry for ``-m examples.google_photos`` (MicroPython / CPython). + +MicroPython requires this file when using ``-m`` on a package. The launcher +side effect lives in :mod:`google_photos` (also imported by ``__init__`` for +plain ``import google_photos`` / gallery). Re-import here is a no-op if +``__init__`` already ran the app to completion. + +From ``pydevices-examples/lib`` (swap in ``micropython``, ``python``, …):: + + micropython -m examples.google_photos +""" + +from . import google_photos # noqa: F401 diff --git a/lib/examples/google_photos/assets/gen_sim_assets.py b/lib/examples/google_photos/assets/gen_sim_assets.py new file mode 100644 index 00000000..3dc40d11 --- /dev/null +++ b/lib/examples/google_photos/assets/gen_sim_assets.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Generate the simulator's sample photos (CPython + Pillow only). + +Six procedural "landscape" scenes, each written as baseline JPEGs at the +sizes ``gphotos_sim.GPhotosSimEngine.asset_path`` looks for: + +* ``sim_NN_64.jpg`` / ``sim_NN_96.jpg`` — square list tiles (Google ``-c`` crop) +* ``sim_NN_320x240.jpg`` / ``sim_NN_160x120.jpg`` — landscape scenes (odd NN) +* ``sim_NN_240x320.jpg`` / ``sim_NN_120x160.jpg`` — portrait scenes (even NN) + +Run from the repo root:: + + .venv/bin/python lib/examples/google_photos/assets/gen_sim_assets.py + +Baseline (non-progressive) JPEG is required: TJpgDec cannot decode +progressive files. +""" + +import math +import os +import sys + +try: + from PIL import Image, ImageDraw +except ImportError: # pragma: no cover + raise SystemExit("Pillow is required: .venv/bin/pip install pillow") + +HERE = os.path.dirname(os.path.abspath(__file__)) + +# (sky top, sky bottom, hills, sun, water) per scene +_PALETTES = [ + ((0x1E, 0x3A, 0x8A), (0xF9, 0x73, 0x16), (0x14, 0x53, 0x2D), (0xFD, 0xE0, 0x47), (0x0E, 0x74, 0x90)), + ((0x0F, 0x17, 0x2A), (0x7C, 0x3A, 0xED), (0x1F, 0x29, 0x37), (0xF4, 0xF4, 0xF5), (0x31, 0x2E, 0x81)), + ((0x38, 0xBD, 0xF8), (0xE0, 0xF2, 0xFE), (0x16, 0xA3, 0x4A), (0xFA, 0xCC, 0x15), (0x06, 0x92, 0xB4)), + ((0x7F, 0x1D, 0x1D), (0xFB, 0xBF, 0x24), (0x45, 0x1A, 0x03), (0xFE, 0xF3, 0xC7), (0x9A, 0x34, 0x12)), + ((0x0C, 0x4A, 0x6E), (0x67, 0xE8, 0xF9), (0x0F, 0x76, 0x6E), (0xFF, 0xFF, 0xFF), (0x15, 0x5E, 0x75)), + ((0x3B, 0x07, 0x64), (0xEC, 0x48, 0x99), (0x2E, 0x10, 0x65), (0xFB, 0x71, 0x85), (0x6B, 0x21, 0xA8)), +] + + +def _lerp(a, b, t): + return tuple(int(a[i] + (b[i] - a[i]) * t) for i in range(3)) + + +def scene(width, height, palette, seed): + """Sky gradient, sun, rolling hills, and a reflective water band.""" + sky_top, sky_bot, hills, sun, water = palette + img = Image.new("RGB", (width, height)) + draw = ImageDraw.Draw(img) + horizon = int(height * 0.62) + for y in range(horizon): + draw.line([(0, y), (width, y)], fill=_lerp(sky_top, sky_bot, y / max(1, horizon - 1))) + # water below the horizon (darker toward the bottom) + for y in range(horizon, height): + t = (y - horizon) / max(1, height - horizon - 1) + draw.line([(0, y), (width, y)], fill=_lerp(water, _lerp(water, (0, 0, 0), 0.55), t)) + # sun + r = max(6, min(width, height) // 7) + cx = int(width * (0.25 + 0.5 * ((seed * 0.37) % 1.0))) + cy = int(horizon * 0.55) + draw.ellipse([cx - r, cy - r, cx + r, cy + r], fill=sun) + # hills: two layered sine ridges + for layer, (amp, freq, shade) in enumerate(((0.10, 1.7, 0.85), (0.16, 0.9, 1.0))): + pts = [] + for x in range(width + 1): + y = horizon - int(height * amp * (0.5 + 0.5 * math.sin(freq * x / width * math.pi * 2 + seed + layer))) + pts.append((x, y)) + pts += [(width, horizon), (0, horizon)] + draw.polygon(pts, fill=_lerp((0, 0, 0), hills, shade)) + # sun reflection stripes + for i in range(0, height - horizon, max(3, height // 40)): + y = horizon + i + w = max(2, int(r * (1.0 - i / max(1, height - horizon)) * 1.5)) + draw.line([(cx - w, y), (cx + w, y)], fill=_lerp(sun, water, 0.35)) + return img + + +def main(): + out_dir = sys.argv[1] if len(sys.argv) > 1 else HERE + os.makedirs(out_dir, exist_ok=True) + written = [] + for n, palette in enumerate(_PALETTES, start=1): + portrait = n % 2 == 0 + big = (240, 320) if portrait else (320, 240) + small = (120, 160) if portrait else (160, 120) + base = scene(big[0], big[1], palette, seed=n * 1.3) + variants = { + "%dx%d" % big: base, + "%dx%d" % small: base.resize(small, Image.LANCZOS), + } + # Square tiles: center crop like Google's ``-c`` parameter. + side = min(big) + left = (big[0] - side) // 2 + top = (big[1] - side) // 2 + square = base.crop((left, top, left + side, top + side)) + variants["96"] = square.resize((96, 96), Image.LANCZOS) + variants["64"] = square.resize((64, 64), Image.LANCZOS) + for tag, im in variants.items(): + path = os.path.join(out_dir, "sim_%02d_%s.jpg" % (n, tag)) + im.save(path, "JPEG", quality=82, optimize=False, progressive=False, subsampling=0) + written.append(path) + total = sum(os.path.getsize(p) for p in written) + print("wrote %d files (%.1f KiB) to %s" % (len(written), total / 1024.0, out_dir)) + + +if __name__ == "__main__": + main() diff --git a/lib/examples/google_photos/assets/sim_01_160x120.jpg b/lib/examples/google_photos/assets/sim_01_160x120.jpg new file mode 100644 index 00000000..4c378738 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_01_160x120.jpg differ diff --git a/lib/examples/google_photos/assets/sim_01_320x240.jpg b/lib/examples/google_photos/assets/sim_01_320x240.jpg new file mode 100644 index 00000000..4a25efb8 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_01_320x240.jpg differ diff --git a/lib/examples/google_photos/assets/sim_01_64.jpg b/lib/examples/google_photos/assets/sim_01_64.jpg new file mode 100644 index 00000000..e8f9d4f5 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_01_64.jpg differ diff --git a/lib/examples/google_photos/assets/sim_01_96.jpg b/lib/examples/google_photos/assets/sim_01_96.jpg new file mode 100644 index 00000000..434ff49a Binary files /dev/null and b/lib/examples/google_photos/assets/sim_01_96.jpg differ diff --git a/lib/examples/google_photos/assets/sim_02_120x160.jpg b/lib/examples/google_photos/assets/sim_02_120x160.jpg new file mode 100644 index 00000000..994e363b Binary files /dev/null and b/lib/examples/google_photos/assets/sim_02_120x160.jpg differ diff --git a/lib/examples/google_photos/assets/sim_02_240x320.jpg b/lib/examples/google_photos/assets/sim_02_240x320.jpg new file mode 100644 index 00000000..eac87b6d Binary files /dev/null and b/lib/examples/google_photos/assets/sim_02_240x320.jpg differ diff --git a/lib/examples/google_photos/assets/sim_02_64.jpg b/lib/examples/google_photos/assets/sim_02_64.jpg new file mode 100644 index 00000000..ccd06591 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_02_64.jpg differ diff --git a/lib/examples/google_photos/assets/sim_02_96.jpg b/lib/examples/google_photos/assets/sim_02_96.jpg new file mode 100644 index 00000000..8c7a7202 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_02_96.jpg differ diff --git a/lib/examples/google_photos/assets/sim_03_160x120.jpg b/lib/examples/google_photos/assets/sim_03_160x120.jpg new file mode 100644 index 00000000..0f0c2fc9 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_03_160x120.jpg differ diff --git a/lib/examples/google_photos/assets/sim_03_320x240.jpg b/lib/examples/google_photos/assets/sim_03_320x240.jpg new file mode 100644 index 00000000..17402ef3 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_03_320x240.jpg differ diff --git a/lib/examples/google_photos/assets/sim_03_64.jpg b/lib/examples/google_photos/assets/sim_03_64.jpg new file mode 100644 index 00000000..d98e13fc Binary files /dev/null and b/lib/examples/google_photos/assets/sim_03_64.jpg differ diff --git a/lib/examples/google_photos/assets/sim_03_96.jpg b/lib/examples/google_photos/assets/sim_03_96.jpg new file mode 100644 index 00000000..42adfc92 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_03_96.jpg differ diff --git a/lib/examples/google_photos/assets/sim_04_120x160.jpg b/lib/examples/google_photos/assets/sim_04_120x160.jpg new file mode 100644 index 00000000..2490beb8 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_04_120x160.jpg differ diff --git a/lib/examples/google_photos/assets/sim_04_240x320.jpg b/lib/examples/google_photos/assets/sim_04_240x320.jpg new file mode 100644 index 00000000..5f47c8b3 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_04_240x320.jpg differ diff --git a/lib/examples/google_photos/assets/sim_04_64.jpg b/lib/examples/google_photos/assets/sim_04_64.jpg new file mode 100644 index 00000000..ee3895ec Binary files /dev/null and b/lib/examples/google_photos/assets/sim_04_64.jpg differ diff --git a/lib/examples/google_photos/assets/sim_04_96.jpg b/lib/examples/google_photos/assets/sim_04_96.jpg new file mode 100644 index 00000000..cb8f3a75 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_04_96.jpg differ diff --git a/lib/examples/google_photos/assets/sim_05_160x120.jpg b/lib/examples/google_photos/assets/sim_05_160x120.jpg new file mode 100644 index 00000000..4de14153 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_05_160x120.jpg differ diff --git a/lib/examples/google_photos/assets/sim_05_320x240.jpg b/lib/examples/google_photos/assets/sim_05_320x240.jpg new file mode 100644 index 00000000..8bbd486c Binary files /dev/null and b/lib/examples/google_photos/assets/sim_05_320x240.jpg differ diff --git a/lib/examples/google_photos/assets/sim_05_64.jpg b/lib/examples/google_photos/assets/sim_05_64.jpg new file mode 100644 index 00000000..311b7c54 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_05_64.jpg differ diff --git a/lib/examples/google_photos/assets/sim_05_96.jpg b/lib/examples/google_photos/assets/sim_05_96.jpg new file mode 100644 index 00000000..49b49491 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_05_96.jpg differ diff --git a/lib/examples/google_photos/assets/sim_06_120x160.jpg b/lib/examples/google_photos/assets/sim_06_120x160.jpg new file mode 100644 index 00000000..25b876ed Binary files /dev/null and b/lib/examples/google_photos/assets/sim_06_120x160.jpg differ diff --git a/lib/examples/google_photos/assets/sim_06_240x320.jpg b/lib/examples/google_photos/assets/sim_06_240x320.jpg new file mode 100644 index 00000000..10aed70d Binary files /dev/null and b/lib/examples/google_photos/assets/sim_06_240x320.jpg differ diff --git a/lib/examples/google_photos/assets/sim_06_64.jpg b/lib/examples/google_photos/assets/sim_06_64.jpg new file mode 100644 index 00000000..f3e2e02d Binary files /dev/null and b/lib/examples/google_photos/assets/sim_06_64.jpg differ diff --git a/lib/examples/google_photos/assets/sim_06_96.jpg b/lib/examples/google_photos/assets/sim_06_96.jpg new file mode 100644 index 00000000..05b45887 Binary files /dev/null and b/lib/examples/google_photos/assets/sim_06_96.jpg differ diff --git a/lib/examples/google_photos/google_photos.py b/lib/examples/google_photos/google_photos.py new file mode 100644 index 00000000..ae93630b --- /dev/null +++ b/lib/examples/google_photos/google_photos.py @@ -0,0 +1,133 @@ +# deps: lvgl +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +""" +google_photos +==================================================== +Pick Google Photos on your phone; the board lists, views, and slideshows them. + +Google Photos on a PyDevices display: pick photos from your phone, browse +the list with thumbnails, view them full screen, or run a slideshow. + +Launcher for the Google Photos stack: + +* :mod:`gphotos_engine` -- Picker API client, token refresh, prefs, thumbnail + cache (no UI) +* :mod:`gphotos_sim` -- offline / PyScript stand-in (``make_engine``) +* :mod:`gphotos_lvgl` -- LVGL front end (list, view, slideshow, QR pairing) + +How picking works: the device creates a Picker API *session* and shows its +link as a QR code. Scan it with your phone, choose photos in Google Photos, +tap Done, and the device lists what you picked. Google only hands out +photos the user explicitly picked -- since March 2025 third-party apps can +no longer browse a whole library -- so this is the official, self-serve path. + +Setup (one time, on a PC; see ``README.md`` beside this file): create a +Google Cloud OAuth *Desktop app* client with the Picker API enabled, run +``tools/gphotos_auth.py`` to sign in, then copy the resulting +``gphotos_tokens.json`` to the board (``mpremote cp … :/gphotos_tokens.json``). +Without that file, or without a network, the launcher runs the simulator +with sample photos so the UI can be explored anywhere (PyScript gallery too). + +Desktop launch from ``pydevices-examples/lib`` (same for ``micropython``, +``python``, ``python.exe``):: + + micropython -m examples.google_photos + +Join Wi-Fi before running on a microcontroller. Optional desktop panel size: +edit ``_WIDTH`` / ``_HEIGHT`` / ``_SCALE`` below (applied via +``displaydev.env_set`` before ``board_config`` is imported; never on MCUs). +""" + +import sys + +_PKG = __file__.replace("\\", "/").rsplit("/", 1)[0] +if _PKG not in sys.path: + sys.path.insert(0, _PKG) + +from displaydev import env_set + +# Local desktop test panel -- change these and re-run. Must stay above board_config. +_WIDTH = None +_HEIGHT = None +_SCALE = None # e.g. 1 or 2; None = AutoDisplay default + autoscale +_MCU_PLATFORMS = ( + "esp32", + "esp8266", + "rp2", + "samd", + "nrf", + "mimxrt", + "renesas-ra", + "stm32", + "zephyr", +) +if getattr(sys, "platform", "") not in _MCU_PLATFORMS: + env_set("PYDEVICES_WIDTH", _WIDTH) + env_set("PYDEVICES_HEIGHT", _HEIGHT) + if _SCALE is not None: + env_set("PYDEVICES_SCALE", _SCALE) + +import gphotos_engine # noqa: E402 +from gphotos_sim import make_engine # noqa: E402 + + +def main(): + # Suppress front-end auto-start on import; we call ``run()`` below. + gphotos_engine._LAUNCHER_OWNS_RUN = True + + engine = make_engine() + has_items = engine.restore() + start_page = "list" if has_items else "connect" + notice = getattr(engine, "sim_notice", "") or "" + print( + "google_photos: engine=%s start_page=%s items=%d%s" + % ( + "sim" if engine.sim else "picker", + start_page, + len(engine.items), + (" (" + notice + ")") if notice else "", + ) + ) + + import gphotos_lvgl + + gphotos_lvgl.run(engine=engine, start_page=start_page) + _block_if_batch() + + +def _block_if_batch(): + """Keep ``python -m examples.google_photos`` alive. + + The shared app host-loop decides its strategy once, when ``display_driver`` + builds the ``App`` -- here that happens while the package ``__init__`` is + still being imported by ``runpy``, before ``__main__.__file__`` exists, so + a ``-m`` launch looks like a REPL ("ambient") or a tool run ("none") and + the process would return to the prompt with the UI half-built. Decide from + the command line instead: under ``-m`` / ``-c`` without ``-i`` on a + desktop host, block here until the app quits. Every other host (script + path, MCU REPL, PyScript, the example kit) keeps the app alive by itself. + """ + try: + from appdev import _hostloop + from display_driver import app + + if not _hostloop.batch() or _hostloop.ambient(): + return + if getattr(getattr(sys, "flags", None), "interactive", 0): + return + except Exception: + return + if app.strategy != _hostloop.AMBIENT or app.timer_async: + app.run() # "none": blocking service loop (async timers: asyncio.run) + return + # "ambient" was chosen at import time; the timer already drives the app, + # so just hold the main thread until the window closes. + import time + + while not app.quit_requested: + time.sleep(0.05) + + +main() diff --git a/lib/examples/google_photos/gphotos_engine.py b/lib/examples/google_photos/gphotos_engine.py new file mode 100644 index 00000000..8d84cda4 --- /dev/null +++ b/lib/examples/google_photos/gphotos_engine.py @@ -0,0 +1,1127 @@ +# gallery: skip +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +""" +`gphotos_engine` +==================================================== + +Google Photos Picker API client — no display, event, or UI imports. + +Since March 31, 2025 the Google Photos Library API only returns media that +the calling app uploaded itself, so the only self-serve way for a device to +show photos from a user's own library is the **Picker API**: the app creates +a *picker session*, the user opens the session's ``pickerUri`` (shown here as +a QR code) in Google Photos on their phone, picks photos, and the app then +lists the picked items and fetches sized thumbnails. + +What this module does, with plain HTTPS (``urllib`` on CPython, ``requests`` +on MicroPython, ``adafruit_requests`` on CircuitPython, raw ``socket`` + TLS +as the last resort): + +* Mint access tokens from a stored **refresh token** (Google's TV + "device code" login refuses the Picker scope, and Google will not register + a LAN IP as an OAuth redirect, so the one-time browser login runs on a PC + via ``tools/gphotos_auth.py`` and its output JSON is copied to the device). +* Create / poll / delete picker sessions, list picked media items, and + download ``baseUrl`` thumbnails (which need the bearer token and expire + after 60 minutes — this module re-lists to refresh them). +* Persist the session and the picked list in a small prefs file so a reboot + lands on the list without re-picking, and keep a bounded thumbnail cache + on flash so images are downloaded once. + +Files (desktop ``~/…``, MCU ``/…``; override with ``GPHOTOS_TOKENS``, +``GPHOTOS_PREFS``, ``GPHOTOS_CACHE``): + +* ``.gphotos_tokens.json`` — ``client_id``, ``client_secret``, + ``refresh_token`` (written by ``tools/gphotos_auth.py``; read-only here). +* ``.gphotos_prefs`` — JSON: current session, picked items (without their + short-lived URLs), ``slideshow_s``. +* ``.gphotos_cache/`` — ``_x[c].jpg|png`` thumbnails. + +Usage:: + + from gphotos_engine import GPhotosEngine + + eng = GPhotosEngine() + eng.restore() # prefs: session + items + if not eng.items: + s = eng.create_session() # show s["pickerUri"] as a QR code + while not eng.poll_session(): + time.sleep(eng.poll_interval_s()) + eng.list_items() + path = eng.thumbnail_path(eng.items[0], 96, 96, crop=True) + +Blocking network calls only; the LVGL front end queues them on its main-tick +pump (no ``_thread``). Errors never raise out of the public methods: they +return ``None`` / ``False`` / ``""`` and set :attr:`GPhotosEngine.last_error`. +""" + +import sys + +try: + import json +except ImportError: # pragma: no cover + json = None + +try: + import os +except ImportError: # pragma: no cover + os = None + +try: + import time +except ImportError: # pragma: no cover + time = None + +try: + import socket +except ImportError: # pragma: no cover — e.g. CircuitPython unix, PyScript + socket = None + +TOKEN_URL = "https://oauth2.googleapis.com/token" +PICKER_BASE = "https://photospicker.googleapis.com/v1" +SCOPE = "https://www.googleapis.com/auth/photospicker.mediaitems.readonly" + +TOKENS_HOME_NAME = ".gphotos_tokens.json" +TOKENS_MCU_NAME = "/gphotos_tokens.json" +PREFS_HOME_NAME = ".gphotos_prefs" +PREFS_MCU_NAME = "/gphotos_prefs" +CACHE_HOME_NAME = ".gphotos_cache" +CACHE_MCU_NAME = "/gphotos_cache" + +# Picker base URLs stay valid for 60 minutes; re-list a little early. +URL_TTL_S = 50 * 60 +# Refresh the access token this long before Google says it expires. +TOKEN_MARGIN_S = 60 +PAGE_SIZE = 25 +DEFAULT_MAX_ITEMS = 100 +DEFAULT_CACHE_MAX = 48 +DEFAULT_SLIDESHOW_S = 5 +DOWNLOAD_CHUNK = 2048 + +# Set by the launcher so ``gphotos_lvgl`` does not auto-run on import. +_LAUNCHER_OWNS_RUN = False + + +class GPhotosError(Exception): + """Transport-level failure (no HTTP status available).""" + + +# ---------------------------------------------------------------------- +# Small portable helpers (no os.path / pathlib — MicroPython-safe) +# ---------------------------------------------------------------------- + + +def _env_get(name): + """Portable getenv (CPython / MicroPython / CircuitPython).""" + if os is None: + return None + environ = getattr(os, "environ", None) + if environ is not None: + try: + val = environ.get(name) + if val: + return val + except Exception: + pass + getenv = getattr(os, "getenv", None) + if getenv is not None: + try: + return getenv(name) + except Exception: + return None + return None + + +def _path_join(base, name): + """Join directory + filename without requiring ``os.path``.""" + if not base: + return name + if base.endswith("/") or base.endswith("\\"): + return base + name + if sys.platform == "win32" and len(base) >= 2 and base[1] == ":": + return base + "\\" + name + return base + "/" + name + + +def _user_home_dir(): + """Best-effort user home on desktop hosts; empty on typical MCU images.""" + for key in ("HOME", "USERPROFILE"): + val = _env_get(key) + if val: + return val + try: + expanduser = getattr(getattr(os, "path", None), "expanduser", None) + if expanduser is not None: + home = expanduser("~") + if home and home != "~": + return home + except Exception: + pass + return "" + + +def _default_path(env_name, home_name, mcu_name): + override = _env_get(env_name) + if override: + return override + home = _user_home_dir() + if home and home not in (".", "/"): + return _path_join(home, home_name) + return mcu_name + + +def tokens_path(): + """Desktop ``~/.gphotos_tokens.json``; MCU ``/gphotos_tokens.json``.""" + return _default_path("GPHOTOS_TOKENS", TOKENS_HOME_NAME, TOKENS_MCU_NAME) + + +def prefs_path(): + """Desktop ``~/.gphotos_prefs``; MCU ``/gphotos_prefs``.""" + return _default_path("GPHOTOS_PREFS", PREFS_HOME_NAME, PREFS_MCU_NAME) + + +def cache_dir(): + """Desktop ``~/.gphotos_cache``; MCU ``/gphotos_cache``.""" + return _default_path("GPHOTOS_CACHE", CACHE_HOME_NAME, CACHE_MCU_NAME) + + +def _exists(path): + if os is None or not path: + return False + try: + os.stat(path) + return True + except OSError: + return False + + +def _ensure_dir(path): + if os is None or not path: + return False + if _exists(path): + return True + try: + os.mkdir(path) + return True + except OSError: + return False + + +def _remove(path): + if os is None or not path: + return + try: + os.remove(path) + except OSError: + pass + + +def _read_text(path): + try: + with open(path, "r") as f: + return f.read() + except OSError: + return "" + + +def _write_text(path, text): + try: + with open(path, "w") as f: + f.write(text) + return True + except OSError: + return False + + +def _mono_ms(): + """Monotonic milliseconds; compare only via :func:`_elapsed_ms`.""" + if time is None: + return 0 + f = getattr(time, "ticks_ms", None) + if f is not None: + return f() + f = getattr(time, "monotonic", None) + if f is not None: + return int(f() * 1000) + return int(time.time() * 1000) + + +def _elapsed_ms(since): + now = _mono_ms() + f = getattr(time, "ticks_diff", None) if time is not None else None + if f is not None: + return f(now, since) + return now - since + + +_SAFE = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_.~" + + +def quote(value): + """RFC 3986 percent-encoding (everything but unreserved characters).""" + out = [] + for ch in str(value): + if ch in _SAFE: + out.append(ch) + else: + for b in ch.encode("utf-8"): + out.append("%%%02X" % b) + return "".join(out) + + +def form_encode(fields): + """``application/x-www-form-urlencoded`` body from ``[(key, value), …]``.""" + return "&".join(quote(k) + "=" + quote(v) for k, v in fields) + + +def json_loads(raw): + """Parse bytes/str JSON → dict (``{}`` on any failure).""" + if json is None or not raw: + return {} + try: + if isinstance(raw, (bytes, bytearray, memoryview)): + raw = bytes(raw).decode("utf-8") + data = json.loads(raw) + except Exception: + return {} + return data if isinstance(data, dict) else {} + + +def parse_duration_s(text, default=5.0): + """Protobuf JSON duration (``"5s"``, ``"1.5s"``) → seconds.""" + if text is None: + return default + try: + t = str(text).strip() + if t.endswith("s"): + t = t[:-1] + return float(t) + except (ValueError, TypeError): + return default + + +def date_label(create_time): + """``2024-05-03T14:22:11Z`` → ``2024-05-03``.""" + return (create_time or "")[:10] + + +def time_label(create_time): + """``2024-05-03T14:22:11Z`` → ``14:22`` (UTC as reported by Google).""" + t = create_time or "" + return t[11:16] if len(t) >= 16 and t[10] == "T" else "" + + +def item_hash(item_id): + """Short stable name for cache files (FNV-1a 32-bit of the media item id).""" + h = 2166136261 + for b in str(item_id).encode("utf-8"): + h ^= b + h = (h * 16777619) & 0xFFFFFFFF + return "%08x" % h + + +def thumb_url(base_url, w, h, crop=False): + """Sized image URL from a Picker ``baseUrl`` (``-c`` crops to exactly w×h).""" + if not base_url: + return "" + return "%s=w%d-h%d%s" % (base_url, int(w), int(h), "-c" if crop else "") + + +def _to_int(value): + try: + return int(value) + except (TypeError, ValueError): + return 0 + + +def normalize_item(raw): + """Flatten one Picker ``PickedMediaItem`` into the dict the UI consumes.""" + if not isinstance(raw, dict): + return {} + mf = raw.get("mediaFile") or {} + meta = mf.get("mediaFileMetadata") or {} + return { + "id": raw.get("id") or "", + "filename": mf.get("filename") or "", + "createTime": raw.get("createTime") or "", + "type": raw.get("type") or "", + "mime": mf.get("mimeType") or "", + "width": _to_int(meta.get("width")), + "height": _to_int(meta.get("height")), + "baseUrl": mf.get("baseUrl") or "", + } + + +def sniff_ext(head): + """``.jpg`` / ``.png`` / ``""`` from the first bytes of an image file.""" + if head[:3] == b"\xff\xd8\xff": + return ".jpg" + if head[:8] == b"\x89PNG\r\n\x1a\n": + return ".png" + return "" + + +# ---------------------------------------------------------------------- +# HTTP transport (urllib → requests / adafruit_requests → raw socket + TLS) +# ---------------------------------------------------------------------- + +_requests_backend = None + + +def _find_requests(): + """MicroPython ``requests`` module or a CircuitPython ``adafruit_requests`` session.""" + global _requests_backend + if _requests_backend is not None: + return _requests_backend or None + try: + import requests as mod + + if hasattr(mod, "request"): + _requests_backend = mod + return mod + except ImportError: + pass + try: + import ssl + import wifi + import socketpool + import adafruit_requests + + pool = socketpool.SocketPool(wifi.radio) + _requests_backend = adafruit_requests.Session(pool, ssl.create_default_context()) + return _requests_backend + except ImportError: + pass + _requests_backend = False + return None + + +def _wrap_tls(sock, host): + """Client TLS for the raw-socket fallback (no CA verification — the + MicroPython ``requests`` default, and most boards have no CA bundle).""" + try: + import tls as _tls + + ctx = _tls.SSLContext(_tls.PROTOCOL_TLS_CLIENT) + ctx.verify_mode = _tls.CERT_NONE + return ctx.wrap_socket(sock, server_hostname=host) + except ImportError: + pass + import ssl + + try: + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + try: + ctx.check_hostname = False + except AttributeError: + pass + ctx.verify_mode = ssl.CERT_NONE + return ctx.wrap_socket(sock, server_hostname=host) + except (AttributeError, TypeError): + return ssl.wrap_socket(sock, server_hostname=host) + + +class _SockIO: + """Tiny buffered stream over CPython / MicroPython sockets (plain or TLS).""" + + def __init__(self, sock): + self.s = sock + self._buf = b"" + + def send(self, data): + s = self.s + if hasattr(s, "sendall"): + s.sendall(data) + return + mv = memoryview(data) + off = 0 + while off < len(mv): + n = s.write(mv[off:]) if hasattr(s, "write") else s.send(mv[off:]) + if n is None: + n = len(mv) - off + off += n + + def _recv(self, n): + s = self.s + if hasattr(s, "recv"): + return s.recv(n) + return s.read(n) + + def readline(self): + while b"\n" not in self._buf: + chunk = self._recv(512) + if not chunk: + line, self._buf = self._buf, b"" + return line + self._buf += chunk + i = self._buf.index(b"\n") + 1 + line, self._buf = self._buf[:i], self._buf[i:] + return line + + def read(self, n): + if self._buf: + out, self._buf = self._buf[:n], self._buf[n:] + return out + return self._recv(n) or b"" + + +def _socket_request(method, url, headers, body, timeout, sink=None): + """HTTP/1.0 over ``socket`` (+TLS). Returns ``(status, body_bytes | n)``.""" + if socket is None: + raise GPhotosError("no socket module") + parts = url.split("/", 3) + proto = parts[0] + host = parts[2] if len(parts) > 2 else "" + path = "/" + (parts[3] if len(parts) > 3 else "") + https = proto == "https:" + port = 443 if https else 80 + if ":" in host: + host, p = host.rsplit(":", 1) + port = int(p) + ai = socket.getaddrinfo(host, port, 0, socket.SOCK_STREAM)[0] + s = socket.socket(ai[0], socket.SOCK_STREAM, ai[2]) + try: + s.settimeout(timeout) + except (AttributeError, OSError): + pass + s.connect(ai[-1]) + if https: + s = _wrap_tls(s, host) + io = _SockIO(s) + try: + lines = ["%s %s HTTP/1.0" % (method, path), "Host: " + host, "Connection: close"] + hdrs = dict(headers or {}) + if body is not None and "Content-Length" not in hdrs: + hdrs["Content-Length"] = str(len(body)) + for k in hdrs: + lines.append("%s: %s" % (k, hdrs[k])) + io.send(("\r\n".join(lines) + "\r\n\r\n").encode("utf-8")) + if body: + io.send(body) + status_line = io.readline().split(None, 2) + if len(status_line) < 2: + raise GPhotosError("bad status line") + status = int(status_line[1]) + length = -1 + while True: + line = io.readline() + if not line or line in (b"\r\n", b"\n"): + break + if line.lower().startswith(b"content-length:"): + length = int(line.split(b":", 1)[1].strip()) + total = 0 + chunks = [] + while length < 0 or total < length: + want = DOWNLOAD_CHUNK if length < 0 else min(DOWNLOAD_CHUNK, length - total) + chunk = io.read(want) + if not chunk: + break + total += len(chunk) + if sink is not None: + sink(chunk) + else: + chunks.append(chunk) + finally: + try: + s.close() + except Exception: + pass + return status, (total if sink is not None else b"".join(chunks)) + + +def _urllib(): + """``(Request, urlopen, HTTPError, URLError)`` on CPython, else ``None``.""" + try: + from urllib.request import Request, urlopen + from urllib.error import HTTPError, URLError + except ImportError: + return None + return Request, urlopen, HTTPError, URLError + + +def _status_of(resp): + for name in ("status_code", "status", "code"): + val = getattr(resp, name, None) + if val is not None: + try: + return int(val) + except (TypeError, ValueError): + pass + return 0 + + +def _transport(method, url, headers, body, timeout, sink=None): + """Run one request. ``sink(chunk)`` streams the body; else it is returned.""" + headers = dict(headers or {}) + if isinstance(body, str): + body = body.encode("utf-8") + + # 1) CPython + mods = _urllib() + if mods is not None: + Request, urlopen, HTTPError, URLError = mods + req = Request(url, data=body, headers=headers, method=method) + try: + resp = urlopen(req, timeout=timeout) + except HTTPError as err: + raw = b"" + try: + raw = err.read() + except Exception: + pass + return int(getattr(err, "code", 0) or 0), raw + except URLError as err: + raise GPhotosError(str(getattr(err, "reason", err))) + try: + status = int(getattr(resp, "status", 200) or 200) + if sink is None: + return status, resp.read() + total = 0 + while True: + chunk = resp.read(DOWNLOAD_CHUNK) + if not chunk: + break + total += len(chunk) + sink(chunk) + return status, total + finally: + resp.close() + + # 2) MicroPython requests / CircuitPython adafruit_requests + req = _find_requests() + if req is not None: + kwargs = {"data": body, "headers": headers, "stream": sink is not None} + try: + resp = req.request(method, url, timeout=timeout, **kwargs) + except TypeError: + resp = req.request(method, url, **kwargs) + except AttributeError: + # settimeout unsupported on this port + resp = req.request(method, url, **kwargs) + try: + status = _status_of(resp) + if sink is None: + raw = getattr(resp, "content", None) + if raw is None: + raw = resp.raw.read() + return status, bytes(raw) + total = 0 + iter_content = getattr(resp, "iter_content", None) + if iter_content is not None: + for chunk in iter_content(DOWNLOAD_CHUNK): + if chunk: + total += len(chunk) + sink(chunk) + else: + raw = resp.raw + while True: + chunk = raw.read(DOWNLOAD_CHUNK) + if not chunk: + break + total += len(chunk) + sink(chunk) + return status, total + finally: + try: + resp.close() + except Exception: + pass + + # 3) raw socket + return _socket_request(method, url, headers, body, timeout, sink) + + +def http_request(method, url, headers=None, data=None, timeout=20.0): + """Portable HTTPS request → ``(status, body_bytes)``. + + Raises :class:`GPhotosError` only when no response arrived at all. + """ + if isinstance(data, str): + data = data.encode("utf-8") + return _transport(method.upper(), url, headers, data, timeout) + + +def http_download(url, path, headers=None, timeout=30.0): + """Stream a GET to ``path`` → ``(status, bytes_written)``. + + The file holds whatever the server sent (an error body on non-2xx), so + callers delete it unless the status is 200. + """ + f = open(path, "wb") + try: + return _transport("GET", url, headers, None, timeout, sink=f.write) + finally: + f.close() + + +# ---------------------------------------------------------------------- +# Credentials +# ---------------------------------------------------------------------- + + +def load_tokens(path=None): + """Read the tokens JSON (``{}`` when missing/invalid).""" + data = json_loads(_read_text(path or tokens_path())) + return data if isinstance(data, dict) else {} + + +def credentials_ok(tokens): + """True when ``client_id`` and ``refresh_token`` are present.""" + return bool(tokens and tokens.get("client_id") and tokens.get("refresh_token")) + + +# ---------------------------------------------------------------------- +# Engine +# ---------------------------------------------------------------------- + + +class GPhotosEngine: + """Picker API client with prefs + thumbnail cache (blocking calls).""" + + sim = False + + def __init__( + self, + tokens=None, + tokens_file=None, + prefs_file=None, + cache_path=None, + timeout=20.0, + max_items=DEFAULT_MAX_ITEMS, + cache_max=DEFAULT_CACHE_MAX, + ): + self.tokens_file = tokens_file or tokens_path() + self.prefs_file = prefs_file or prefs_path() + self.cache_path = cache_path or cache_dir() + self.tokens = tokens if tokens is not None else load_tokens(self.tokens_file) + self.timeout = timeout + self.max_items = max_items + self.cache_max = cache_max + self.access_token = "" + self._token_at = None + self._token_ttl_ms = 0 + self.session = None + self.items = [] + self._urls_at = None + self.last_error = "" + self.prefs = None + + # ----- credentials / tokens --------------------------------------- + + def has_credentials(self): + return credentials_ok(self.tokens) + + def _token_expired(self): + if self._token_at is None or not self.access_token: + return True + return _elapsed_ms(self._token_at) >= self._token_ttl_ms + + def ensure_token(self, force=False): + """Mint an access token from the refresh token when needed.""" + if not force and not self._token_expired(): + return True + if not self.has_credentials(): + self.last_error = "no credentials - run tools/gphotos_auth.py" + return False + t = self.tokens + body = form_encode( + [ + ("client_id", t.get("client_id", "")), + ("client_secret", t.get("client_secret", "")), + ("refresh_token", t.get("refresh_token", "")), + ("grant_type", "refresh_token"), + ] + ) + status, raw = self._request( + "POST", + TOKEN_URL, + {"Content-Type": "application/x-www-form-urlencoded"}, + body, + ) + data = json_loads(raw) + token = data.get("access_token") if status == 200 else None + if not token: + err = data.get("error") or "" + if isinstance(err, dict): + err = err.get("status") or err.get("message") or "" + desc = data.get("error_description") or "" + text = ("%s: %s" % (err, desc)) if (err and desc) else (err or desc) + if status: + self.last_error = ("token refresh: HTTP %d %s" % (status, text)).rstrip() + elif not self.last_error: + self.last_error = "token refresh failed" + self.access_token = "" + self._token_at = None + return False + ttl = _to_int(data.get("expires_in")) or 3600 + self.access_token = token + self._token_at = _mono_ms() + self._token_ttl_ms = max(30, ttl - TOKEN_MARGIN_S) * 1000 + self.last_error = "" + return True + + def auth_headers(self): + return {"Authorization": "Bearer " + self.access_token} + + # ----- raw HTTP ---------------------------------------------------- + + def _request(self, method, url, headers=None, data=None): + try: + return http_request(method, url, headers=headers, data=data, timeout=self.timeout) + except Exception as err: + self.last_error = "network: %s" % err + return 0, b"" + + def _api(self, method, url, body=None, content_type=None, retry=True): + """Authorized JSON call → ``(status, dict)``; refreshes once on 401.""" + if not self.ensure_token(): + return 0, {} + headers = self.auth_headers() + if body is not None: + headers["Content-Type"] = content_type or "application/json" + status, raw = self._request(method, url, headers, body) + if status == 401 and retry and self.ensure_token(force=True): + return self._api(method, url, body, content_type, retry=False) + return status, json_loads(raw) + + def _api_error(self, what, status, data): + err = data.get("error") if isinstance(data, dict) else None + msg = "" + if isinstance(err, dict): + msg = err.get("message") or err.get("status") or "" + elif isinstance(err, str): + msg = err + if status == 0: + return self.last_error or ("%s: no response" % what) + return "%s: HTTP %d %s" % (what, status, msg) if msg else "%s: HTTP %d" % (what, status) + + @staticmethod + def _session_gone(status, data): + if status in (404, 410): + return True + err = data.get("error") if isinstance(data, dict) else None + text = "" + if isinstance(err, dict): + text = ("%s %s" % (err.get("status", ""), err.get("message", ""))).lower() + return status == 400 and ("expired" in text or "not found" in text) + + # ----- sessions ---------------------------------------------------- + + @staticmethod + def _session_from(data, previous=None): + pc = data.get("pollingConfig") or {} + prev = previous or {} + return { + "id": data.get("id") or prev.get("id") or "", + "pickerUri": data.get("pickerUri") or prev.get("pickerUri") or "", + "expireTime": data.get("expireTime") or prev.get("expireTime") or "", + "mediaItemsSet": bool(data.get("mediaItemsSet")), + "pollInterval": parse_duration_s( + pc.get("pollInterval"), prev.get("pollInterval", 5.0) + ), + "timeoutIn": parse_duration_s(pc.get("timeoutIn"), prev.get("timeoutIn", 1800.0)), + } + + def create_session(self): + """Start a picker session; returns the session dict (``pickerUri``…).""" + status, data = self._api("POST", PICKER_BASE + "/sessions", body=b"{}") + if status != 200 or not data.get("id") or not data.get("pickerUri"): + self.last_error = self._api_error("create session", status, data) + return None + self.session = self._session_from(data) + self.items = [] + self._urls_at = None + self.last_error = "" + self._save_prefs() + return self.session + + def poll_interval_s(self): + s = self.session or {} + return max(2.0, float(s.get("pollInterval") or 5.0)) + + def poll_session(self): + """Refresh session state; True once the user finished picking.""" + s = self.session + if not s or not s.get("id"): + self.last_error = "no session" + return False + status, data = self._api("GET", PICKER_BASE + "/sessions/" + quote(s["id"])) + if status == 200 and data.get("id"): + self.session = self._session_from(data, s) + self.last_error = "" + self._save_prefs() + return self.session["mediaItemsSet"] + if self._session_gone(status, data): + self.last_error = "session expired - pick again" + self.session = None + self._save_prefs() + return False + self.last_error = self._api_error("poll session", status, data) + return False + + def delete_session(self): + """Drop the picker session (Google forgets the picks); keeps the list.""" + s = self.session + self.session = None + self._urls_at = None + self._save_prefs() + if not s or not s.get("id"): + return True + status, data = self._api("DELETE", PICKER_BASE + "/sessions/" + quote(s["id"])) + if status not in (200, 204) and not self._session_gone(status, data): + self.last_error = self._api_error("delete session", status, data) + return False + return True + + # ----- media items ------------------------------------------------- + + def list_items(self): + """Fetch every picked item (fresh ``baseUrl``s) → list or ``None``.""" + s = self.session + if not s or not s.get("id"): + self.last_error = "no session" + return None + items = [] + token = "" + while True: + url = "%s/mediaItems?sessionId=%s&pageSize=%d" % (PICKER_BASE, quote(s["id"]), PAGE_SIZE) + if token: + url += "&pageToken=" + quote(token) + status, data = self._api("GET", url) + if status != 200: + if self._session_gone(status, data): + self.last_error = "session expired - pick again" + self.session = None + self._save_prefs() + else: + self.last_error = self._api_error("list media items", status, data) + return None + for raw in data.get("mediaItems") or []: + item = normalize_item(raw) + if item.get("id"): + items.append(item) + if len(items) >= self.max_items: + break + token = data.get("nextPageToken") or "" + if not token or len(items) >= self.max_items: + break + self.items = items + self._urls_at = _mono_ms() + self.last_error = "" + self._save_prefs() + return items + + def urls_valid(self): + """True while the cached ``baseUrl``s are still usable.""" + if self._urls_at is None or not self.items: + return False + if _elapsed_ms(self._urls_at) >= URL_TTL_S * 1000: + return False + for item in self.items: + if item.get("baseUrl"): + return True + return False + + def refresh_urls(self): + """Re-list to get fresh ``baseUrl``s (keeps the same items).""" + if self.urls_valid(): + return True + return self.list_items() is not None + + def item_by_id(self, item_id): + for item in self.items: + if item.get("id") == item_id: + return item + return None + + # ----- thumbnail cache --------------------------------------------- + + def cache_name(self, item, w, h, crop=False): + return "%s_%dx%d%s" % (item_hash(item.get("id", "")), int(w), int(h), "c" if crop else "") + + def cached_path(self, item, w, h, crop=False): + """Existing cache file for this item/size, or ``""``.""" + base = _path_join(self.cache_path, self.cache_name(item, w, h, crop)) + for ext in (".jpg", ".png"): + if _exists(base + ext): + return base + ext + return "" + + def _download(self, url, path, retry=True): + if not self.ensure_token(): + return 0 + try: + status, n = http_download(url, path, headers=self.auth_headers(), timeout=self.timeout) + except Exception as err: + self.last_error = "download: %s" % err + _remove(path) + return 0 + if status == 401 and retry and self.ensure_token(force=True): + return self._download(url, path, retry=False) + if status != 200 or not n: + self.last_error = "download: HTTP %d" % status + _remove(path) + return 0 + return n + + def thumbnail_path(self, item, w, h, crop=False): + """Path of a cached w×h image for ``item`` (downloading it if needed). + + ``crop=True`` asks Google for an exact w×h center crop (list tiles); + ``crop=False`` fits the photo inside w×h (full-screen view). Returns + ``""`` on failure (see :attr:`last_error`). + """ + if not item or not item.get("id"): + self.last_error = "no item" + return "" + path = self.cached_path(item, w, h, crop) + if path: + return path + if not _ensure_dir(self.cache_path): + self.last_error = "cache dir unavailable: " + self.cache_path + return "" + fresh = item + if not item.get("baseUrl") or not self.urls_valid(): + if not self.refresh_urls(): + return "" + fresh = self.item_by_id(item["id"]) or item + if not fresh.get("baseUrl"): + self.last_error = "item no longer in the picked set" + return "" + url = thumb_url(fresh["baseUrl"], w, h, crop) + base = _path_join(self.cache_path, self.cache_name(item, w, h, crop)) + tmp = base + ".part" + if not self._download(url, tmp): + return "" + head = b"" + try: + with open(tmp, "rb") as f: + head = f.read(8) + except OSError: + pass + ext = sniff_ext(head) + if not ext: + _remove(tmp) + self.last_error = "download: not an image" + return "" + final = base + ext + _remove(final) + try: + os.rename(tmp, final) + except OSError: + self.last_error = "cache write failed" + _remove(tmp) + return "" + self._evict_cache() + return final + + def _evict_cache(self): + """Keep at most ``cache_max`` files: drop orphans first, then oldest.""" + if os is None or self.cache_max <= 0: + return + try: + names = [n for n in os.listdir(self.cache_path) if not n.endswith(".part")] + except OSError: + return + if len(names) <= self.cache_max: + return + keep = set(item_hash(it.get("id", "")) for it in self.items) + orphans = [n for n in names if n[:8] not in keep] + for n in orphans: + _remove(_path_join(self.cache_path, n)) + names.remove(n) + if len(names) <= self.cache_max: + return + + def _mtime(name): + try: + return os.stat(_path_join(self.cache_path, name))[8] + except (OSError, IndexError): + return 0 + + names.sort(key=_mtime) + while len(names) > self.cache_max: + _remove(_path_join(self.cache_path, names.pop(0))) + + def clear_cache(self): + if os is None: + return + try: + names = os.listdir(self.cache_path) + except OSError: + return + for n in names: + _remove(_path_join(self.cache_path, n)) + + # ----- prefs ------------------------------------------------------- + + def _load_prefs(self): + if self.prefs is None: + data = json_loads(_read_text(self.prefs_file)) + self.prefs = data if isinstance(data, dict) else {} + return self.prefs + + def _save_prefs(self): + prefs = self._load_prefs() + prefs["session"] = dict(self.session) if self.session else None + prefs["items"] = [ + { + "id": it.get("id", ""), + "filename": it.get("filename", ""), + "createTime": it.get("createTime", ""), + "type": it.get("type", ""), + "mime": it.get("mime", ""), + "width": it.get("width", 0), + "height": it.get("height", 0), + } + for it in self.items + ] + if json is None: + return False + try: + body = json.dumps(prefs) + except Exception: + return False + return _write_text(self.prefs_file, body) + + def get_pref(self, key, default=None): + val = self._load_prefs().get(key) + return default if val is None else val + + def set_pref(self, key, value): + self._load_prefs()[key] = value + return self._save_prefs() + + def restore(self): + """Load the saved session + picked list; True when items exist.""" + prefs = self._load_prefs() + session = prefs.get("session") + self.session = dict(session) if isinstance(session, dict) and session.get("id") else None + items = [] + for raw in prefs.get("items") or []: + if isinstance(raw, dict) and raw.get("id"): + it = dict(raw) + it["baseUrl"] = "" + items.append(it) + self.items = items + self._urls_at = None + return bool(items) + + def forget(self, clear_cache=False): + """Drop session + list (and optionally the cached thumbnails).""" + self.session = None + self.items = [] + self._urls_at = None + self._save_prefs() + if clear_cache: + self.clear_cache() + + # ----- labels ------------------------------------------------------ + + def count_label(self): + n = len(self.items) + if n == 1: + return "1 photo" + return "%d photos" % n diff --git a/lib/examples/google_photos/gphotos_lvgl.py b/lib/examples/google_photos/gphotos_lvgl.py new file mode 100644 index 00000000..84b6f890 --- /dev/null +++ b/lib/examples/google_photos/gphotos_lvgl.py @@ -0,0 +1,1197 @@ +# deps: lvgl +# modules: gphotos_engine, gphotos_sim +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +""" +gphotos_lvgl +==================================================== +Google Photos picker, list, viewer, and slideshow built with LVGL. + +Pages: + +* **connect** -- QR code of the Picker session link. Scan it with your + phone, choose photos in Google Photos, tap Done; the device polls the + session and jumps to the list. Desktop hosts also get OPEN (opens the + link in the local browser). +* **list** -- paged rows of thumbnail + file name + date. PICK starts a new + session, SLIDES starts the slideshow, MORE pages through long picks. +* **view** -- one photo fitted to the panel with BACK / PREV / PLAY / NEXT; + tapping the photo advances. PLAY auto-advances every ``slideshow_s`` + seconds (prefs; default 5). + +Import order matters: ``display_driver`` must be imported after +``board_config`` so LVGL's display / input devices are wired before widgets +are created. All network work (token refresh, session polling, image +downloads) is queued and drained from an ``lv.timer`` pump on the main +thread -- no ``_thread`` (ESP32 thread stacks are too small for TLS). Jobs +only touch the engine and post results into mailboxes; the pump applies +them to LVGL. + +JPEG decoding: CPython's LVGL keeps TJPGD; MicroPython / CircuitPython LVGL +firmware decodes through displayif's ``jpegio``, registered as an LVGL image +decoder. PNG goes through LODEPNG on every build. Without a JPEG decoder the +list still works -- tiles keep a placeholder and the viewer says why. + +Launch via ``google_photos`` (prefs + sim detection). Direct +``gphotos_lvgl.run()`` also works. Join Wi-Fi before running on a +microcontroller. +""" + +import gc +import sys + +_EXAMPLES = __file__.replace("\\", "/").rsplit("/", 1)[0] +if _EXAMPLES not in sys.path: + sys.path.insert(0, _EXAMPLES) + +import board_config # noqa: E402,F401 — before display_driver so LVGL wiring sees it +import display_driver # noqa: E402 — wires LVGL display/input into the app +import lvgl as lv # noqa: E402 +from board_config import display_drv # noqa: E402 +from gphotos_engine import date_label, time_label # noqa: E402 +from gphotos_sim import make_engine # noqa: E402 +from multimer import ticks_diff, ticks_ms # noqa: E402 + +try: + import webbrowser as _webbrowser +except ImportError: # MicroPython / CircuitPython + _webbrowser = None + +FRONTEND = "lvgl" +PUMP_MS = 250 +DEFAULT_SLIDESHOW_S = 5 + +# Charcoal chassis, one teal accent (distinct from the Roku violet). +_COL = { + "bg": 0x10141A, + "plaque": 0x1B2129, + "plaque_edge": 0x2C3540, + "row": 0x222B35, + "row_alt": 0x1D252E, + "tile": 0x33404D, + "accent": 0x1E9E8A, + "accent2": 0x157A6A, + "ui": 0x2F3A47, + "ui2": 0x243039, + "text": 0xF3F5F7, + "muted": 0x9AA5B1, + "on_accent": 0xFFFFFF, + "black": 0x000000, + "white": 0xFFFFFF, +} + + +def _hex(rgb): + return lv.color_hex(rgb) + + +def _shade(rgb, factor): + r = max(0, min(255, int(((rgb >> 16) & 0xFF) * factor))) + g = max(0, min(255, int(((rgb >> 8) & 0xFF) * factor))) + b = max(0, min(255, int((rgb & 0xFF) * factor))) + return (r << 16) | (g << 8) | b + + +def _sym(name, fallback): + sym = getattr(lv, "SYMBOL", None) + if sym is not None: + val = getattr(sym, name, None) + if val: + return val + return fallback + + +def _pick_font_from(candidates, ref_obj=None): + for size in candidates: + font = getattr(lv, "font_montserrat_" + str(size), None) + if font is not None: + return font + if ref_obj is not None: + for getter in ("theme_get_font_normal", "theme_get_font_small"): + fn = getattr(lv, getter, None) + if fn is None: + continue + try: + font = fn(ref_obj) + if font is not None: + return font + except Exception: + pass + fn = getattr(lv, "font_get_default", None) + if fn is not None: + try: + return fn() + except Exception: + pass + return None + + +def _pick_font_at_most(max_size, ref_obj=None): + return _pick_font_from([s for s in (24, 22, 20, 18, 16, 14, 12) if s <= max_size], ref_obj) + + +def _apply_font(obj, font): + if font is None: + return + try: + obj.set_style_text_font(font, 0) + except Exception: + pass + + +def _flag(name): + obj_flag = getattr(getattr(lv, "obj", None), "FLAG", None) + if obj_flag is not None: + return getattr(obj_flag, name, None) + return getattr(getattr(lv, "OBJ_FLAG", None), name, None) + + +def _add_flag(obj, name): + flag = _flag(name) + if flag is None or obj is None: + return + try: + obj.add_flag(flag) + except Exception: + pass + + +def _remove_flag(obj, name): + flag = _flag(name) + if flag is None or obj is None: + return + try: + if hasattr(obj, "remove_flag"): + obj.remove_flag(flag) + else: + obj.clear_flag(flag) + except Exception: + pass + + +def _no_scroll(obj): + _remove_flag(obj, "SCROLLABLE") + + +def _set_hidden(obj, hidden): + if hidden: + _add_flag(obj, "HIDDEN") + else: + _remove_flag(obj, "HIDDEN") + + +# ----- image decoding --------------------------------------------------- + +_JPEG_OK = None +_PNG_OK = None + + +def jpeg_supported(): + """True when LVGL can decode JPEG on this build (TJPGD or jpegio).""" + global _JPEG_OK + if _JPEG_OK is not None: + return _JPEG_OK + ok = hasattr(lv, "tjpgd_init") # CPython: registered by lv.init() + if not ok: + try: + import jpegio # MicroPython (displayif) / CircuitPython + + ok = True + reg = getattr(jpegio, "register_lvgl_decoder", None) + if reg is not None: + try: + reg() + except RuntimeError: + ok = False + names = getattr(jpegio, "lvgl_decoders", None) + if ok and names is not None: + try: + ok = "jpegio" in tuple(names()) + except Exception: + pass + except ImportError: + ok = False + _JPEG_OK = ok + return ok + + +def png_supported(): + global _PNG_OK + if _PNG_OK is None: + _PNG_OK = hasattr(lv, "lodepng_init") + return _PNG_OK + + +def _jpeg_size(data): + """(width, height) from the SOF marker, or (0, 0).""" + i = 2 + n = len(data) + while i + 9 < n: + if data[i] != 0xFF: + i += 1 + continue + marker = data[i + 1] + i += 2 + if marker in (0xD8, 0xD9, 0x01) or 0xD0 <= marker <= 0xD7: + continue + seg = (data[i] << 8) | data[i + 1] + if marker in (0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB): + return (data[i + 5] << 8) | data[i + 6], (data[i + 3] << 8) | data[i + 4] + i += seg + return 0, 0 + + +def _png_size(data): + if len(data) < 24: + return 0, 0 + w = (data[16] << 24) | (data[17] << 16) | (data[18] << 8) | data[19] + h = (data[20] << 24) | (data[21] << 16) | (data[22] << 8) | data[23] + return w, h + + +def image_descriptor(path): + """``(dsc, data, w, h)`` for a cached JPEG/PNG, else ``(None, None, 0, 0)``. + + ``data`` must stay referenced as long as ``dsc`` is an image source. + """ + try: + with open(path, "rb") as f: + data = f.read() + except OSError: + return None, None, 0, 0 + if data[:3] == b"\xff\xd8\xff": + if not jpeg_supported(): + return None, None, 0, 0 + w, h = _jpeg_size(data) + elif data[:8] == b"\x89PNG\r\n\x1a\n": + if not png_supported(): + return None, None, 0, 0 + w, h = _png_size(data) + else: + return None, None, 0, 0 + if not w or not h: + return None, None, 0, 0 + try: + dsc = lv.image_dsc_t() + dsc.header.magic = lv.IMAGE_HEADER_MAGIC + dsc.header.cf = lv.COLOR_FORMAT.RAW + dsc.header.w = w + dsc.header.h = h + dsc.header.stride = 0 + dsc.data_size = len(data) + dsc.data = data + except Exception: + try: + dsc = lv.image_dsc_t( + { + "header": { + "magic": lv.IMAGE_HEADER_MAGIC, + "w": w, + "h": h, + "cf": lv.COLOR_FORMAT.RAW, + }, + "data_size": len(data), + "data": data, + } + ) + except Exception: + return None, None, 0, 0 + return dsc, data, w, h + + +# ----- front end ---------------------------------------------------------- + + +class _GPhotosLvgl: + PAGES = ("connect", "list", "view") + + def __init__(self, engine=None, start_page="list"): + self.engine = engine if engine is not None else make_engine() + self.page = start_page if start_page in self.PAGES else "list" + self.index = 0 + self.list_offset = 0 + self.page_size = 1 + self.slideshow = False + try: + self.slideshow_s = int(self.engine.get_pref("slideshow_s", DEFAULT_SLIDESHOW_S)) + except (TypeError, ValueError): + self.slideshow_s = DEFAULT_SLIDESHOW_S + if self.slideshow_s < 1: + self.slideshow_s = 1 + + # Cooperative bg queue (drained by ``_pump`` -- no ``_thread``). + self._bg_q = [] + self._bg_busy = False + # Mailboxes: jobs write, the pump applies on the LVGL thread. + self._pending_status = None + self._pending_thumbs = [] + self._pending_view = None + self._pending_list = False + self._pending_connect = False + self._poll_busy = False + self._poll_at = None + self._session_started = None + self._poll_timed_out = False + self._thumb_busy = {} + self._images = {} # item id -> (dsc, data): resident tiles (current page) + self._rows = {} # item id -> (tile image, placeholder label) + self._view_ref = None # (dsc, data) for the viewer + self._view_fetching = None + self._view_loaded_at = None + self._prefetched = None + self._page_root = None + self._last_status_text = None + self._last_title_text = None + self._last_count_text = None + + self.W = display_drv.width + self.H = display_drv.height + self.unit = min(self.W, self.H) + self.pad = max(4, self.unit // 64) + self.radius = max(6, self.unit // 26) + self.plaque_h = max(40 if self.H <= 360 else 60, self.H // 10) + self.font = None + self.font_sm = None + self.title_lbl = None + self.count_lbl = None + self.status_lbl = None + self.content = None + self.view_img = None + self.view_msg = None + self.play_lbl = None + self._timer = None + + self.build_ui() + + # ----- chrome --------------------------------------------------------- + + def _panel(self, obj, bg, radius=0, edge=None): + try: + obj.set_style_bg_color(_hex(bg), 0) + obj.set_style_bg_opa(lv.OPA.COVER, 0) + obj.set_style_radius(radius, 0) + obj.set_style_pad_all(0, 0) + if edge is None: + obj.set_style_border_width(0, 0) + else: + obj.set_style_border_color(_hex(edge), 0) + obj.set_style_border_width(1, 0) + obj.set_style_shadow_width(0, 0) + except Exception: + pass + + def build_ui(self): + loop = getattr(display_driver, "event_loop", None) + inst = loop.current_instance() if loop is not None else None + if inst is not None: + inst.disable() + try: + scr = lv.screen_active() + if self.H <= 360 or self.unit < 280: + self.font = _pick_font_at_most(14, scr) + self.font_sm = _pick_font_at_most(12, scr) or self.font + else: + self.font = _pick_font_at_most(16, scr) + self.font_sm = _pick_font_at_most(14, scr) or self.font + self._panel(scr, _COL["bg"]) + _no_scroll(scr) + + plaque = lv.obj(scr) + plaque_w = self.W - 2 * self.pad + plaque.set_size(plaque_w, self.plaque_h) + plaque.align(lv.ALIGN.TOP_MID, 0, self.pad) + self._panel(plaque, _COL["plaque"], self.radius, edge=_COL["plaque_edge"]) + _no_scroll(plaque) + inset = max(3, self.pad) + half = max(1, self.plaque_h // 2) + + self.count_lbl = lv.label(plaque) + self.count_lbl.set_text("") + self.count_lbl.set_style_text_color(_hex(_COL["muted"]), 0) + _apply_font(self.count_lbl, self.font) + self.count_lbl.align(lv.ALIGN.RIGHT_MID, -inset, -half // 2) + + self.title_lbl = lv.label(plaque) + self.title_lbl.set_text("Google Photos") + self.title_lbl.set_style_text_color(_hex(_COL["text"]), 0) + _apply_font(self.title_lbl, self.font) + self.title_lbl.set_width(plaque_w - 2 * inset - max(48, self.unit // 5)) + self._long_mode(self.title_lbl, "DOTS") + self.title_lbl.align(lv.ALIGN.LEFT_MID, inset, -half // 2) + + self.status_lbl = lv.label(plaque) + self.status_lbl.set_text("") + self.status_lbl.set_style_text_color(_hex(_COL["muted"]), 0) + _apply_font(self.status_lbl, self.font_sm) + self.status_lbl.set_width(plaque_w - 2 * inset) + self._long_mode(self.status_lbl, "DOTS") + self.status_lbl.align(lv.ALIGN.LEFT_MID, inset, half // 2) + + self.content = lv.obj(scr) + self.content.set_size(self.W, self.H - self.plaque_h - 2 * self.pad) + self.content.set_pos(0, self.plaque_h + 2 * self.pad) + self._panel(self.content, _COL["bg"]) + self.content.set_style_bg_opa(lv.OPA.TRANSP, 0) + _no_scroll(self.content) + + if self.page == "connect": + self._start_pick(rebuild=False) + if self.page == "view" and not self.engine.items: + self.page = "list" + self._show_page(self.page) + finally: + if inst is not None: + inst.enable() + self._install_pump() + + def _long_mode(self, lbl, name): + lm = getattr(lv.label, "LONG_MODE", None) + mode = getattr(lm, name, None) if lm is not None else None + if mode is None and name == "DOTS": + mode = getattr(lm, "DOT", None) if lm is not None else None + if mode is not None: + try: + lbl.set_long_mode(mode) + except Exception: + pass + + def _install_pump(self): + creator = getattr(lv, "timer_create", None) + if creator is None: + return + try: + self._timer = creator(self._pump, PUMP_MS, None) + except Exception: + self._timer = None + + def _set_label(self, lbl, text, attr): + if lbl is None: + return + text = text or "" + if getattr(self, attr) == text: + return + setattr(self, attr, text) + try: + lbl.set_text(text) + except Exception: + pass + + def _set_status(self, text): + self._set_label(self.status_lbl, text, "_last_status_text") + + def _set_title(self, text): + self._set_label(self.title_lbl, text, "_last_title_text") + + def _set_count(self, text): + self._set_label(self.count_lbl, text, "_last_count_text") + + def _content_metrics(self): + W = self.W + H = max(80, self.H - self.plaque_h - 2 * self.pad) + if self.content is not None: + try: + reported = int(self.content.get_height()) + if reported >= 80: + H = reported + except Exception: + pass + return W, H + + def _button(self, parent, text, x, y, w, h, role, on_click, font=None): + if role == "accent": + top, fg = _COL["accent"], _COL["on_accent"] + else: + top, fg = _COL["ui"], _COL["text"] + btn = lv.button(parent) + btn.set_size(int(w), int(h)) + btn.set_pos(int(x), int(y)) + _remove_flag(btn, "CLICK_FOCUSABLE") + self._panel(btn, top, self.radius, edge=_shade(top, 0.6)) + try: + btn.set_style_bg_color(_hex(_shade(top, 1.25)), lv.STATE.PRESSED) + except Exception: + pass + lbl = lv.label(btn) + lbl.set_text(text) + lbl.set_style_text_color(_hex(fg), 0) + _apply_font(lbl, font or self.font) + lbl.center() + + def _cb(_e, _fn=on_click): + _fn() + + btn.add_event_cb(_cb, lv.EVENT.CLICKED, None) + return btn, lbl + + # ----- pages ---------------------------------------------------------- + + def _clear_content(self): + root = self._page_root + self._page_root = None + self._rows = {} + self.view_img = None + self.view_msg = None + self.play_lbl = None + if root is not None: + try: + root.delete() + except Exception: + pass + + def _show_page(self, page): + self.page = page + if self.content is None: + return + self._clear_content() + if page != "view": + self._view_ref = None + self._view_fetching = None + self._view_loaded_at = None + if self.slideshow and page == "list": + pass # SLIDES re-arms it; PLAY on the viewer toggles it + root = lv.obj(self.content) + W, H = self._content_metrics() + root.set_size(W, H) + root.set_pos(0, 0) + self._panel(root, _COL["bg"]) + root.set_style_bg_opa(lv.OPA.TRANSP, 0) + _no_scroll(root) + self._page_root = root + if page == "connect": + self._build_connect(root) + elif page == "view": + self._build_view(root) + else: + self._build_list(root) + gc.collect() + + # connect ------------------------------------------------------------ + + def _build_connect(self, parent): + W, H = self._content_metrics() + pad = self.pad + gap = pad + btn_h = max(36, H // 11) + session = self.engine.session + uri = (session or {}).get("pickerUri") or "" + self._set_title("Google Photos") + self._set_count("") + + labels = [("RETRY", "ui", self._start_pick)] + if uri and _webbrowser is not None and not self.engine.sim: + labels.insert(0, ("OPEN", "accent", self._open_browser)) + if self.engine.items: + labels.append(("BACK", "ui", self._goto_list)) + n = len(labels) + w = W - 2 * pad + bw = (w - (n - 1) * gap) // n + y_btn = H - btn_h + for i, (text, role, cb) in enumerate(labels): + self._button(parent, text, pad + i * (bw + gap), y_btn, bw, btn_h, role, cb) + + caption = lv.label(parent) + caption.set_width(w) + self._long_mode(caption, "WRAP") + caption.set_style_text_color(_hex(_COL["text"]), 0) + _apply_font(caption, self.font_sm) + ta = getattr(lv, "TEXT_ALIGN", None) + center = getattr(ta, "CENTER", None) if ta is not None else None + if center is not None: + try: + caption.set_style_text_align(center, 0) + except Exception: + pass + caption_h = max(2 * 16, 2 * (self.unit // 16)) + caption.align(lv.ALIGN.BOTTOM_MID, 0, -(btn_h + gap)) + + qr_side = min(W - 4 * pad, y_btn - caption_h - 3 * gap, 320) + if uri: + if self.engine.sim: + caption.set_text("Simulator: a real session shows a Google Photos link here.\nPicks arrive by themselves in a moment.") + else: + caption.set_text("Scan with your phone, pick photos in Google Photos, tap Done.") + qr_cls = getattr(lv, "qrcode", None) + if qr_cls is not None and qr_side >= 48: + try: + qr = qr_cls(parent) + qr.set_size(int(qr_side)) + qr.set_dark_color(_hex(_COL["black"])) + qr.set_light_color(_hex(_COL["white"])) + if hasattr(qr, "set_quiet_zone"): + qr.set_quiet_zone(True) + raw = uri.encode("utf-8") # bytes: buffer protocol on every binding + qr.update(raw, len(raw)) + qr.align(lv.ALIGN.TOP_MID, 0, gap) + except Exception as err: + self._link_label(parent, uri, w, "QR unavailable (%s)" % err) + else: + self._link_label(parent, uri, w, "") + if not self.engine.sim: + print("google_photos: open this link in Google Photos:\n " + uri) + if self._poll_timed_out: + self._set_status("picker timed out - tap RETRY") + else: + self._set_status("waiting for your picks...") + else: + caption.set_text("Tap RETRY to start a picker session.") + if self.engine.last_error: + self._set_status(self.engine.last_error) + elif not self.engine.has_credentials(): + self._set_status("no tokens file - see README") + else: + self._set_status("starting picker session...") + + def _link_label(self, parent, uri, w, note): + lbl = lv.label(parent) + lbl.set_width(w) + self._long_mode(lbl, "WRAP") + lbl.set_style_text_color(_hex(_COL["muted"]), 0) + _apply_font(lbl, self.font_sm) + lbl.set_text((note + "\n" if note else "") + uri) + lbl.align(lv.ALIGN.TOP_MID, 0, self.pad) + + def _open_browser(self): + uri = (self.engine.session or {}).get("pickerUri") or "" + if not uri or _webbrowser is None: + return + try: + _webbrowser.open(uri) + self._set_status("opened in your browser - pick, then tap Done") + except Exception as err: + self._set_status("browser: %s" % err) + + def _start_pick(self, rebuild=True): + """Create a fresh picker session in the background, then show its QR.""" + self._poll_timed_out = False + self._session_started = None + self._poll_at = None + self._poll_busy = False + self.slideshow = False + self.list_offset = 0 + engine = self.engine + + def _work(): + s = engine.create_session() + if s: + self._session_started = ticks_ms() + else: + self._pending_status = engine.last_error or "could not start a session" + self._pending_connect = True + + self._run_bg(_work) + if rebuild: + self._show_page("connect") + + def _poll_pick(self): + engine = self.engine + self._poll_busy = True + + def _work(): + try: + done = engine.poll_session() + if done: + items = engine.list_items() + if items is None: + self._pending_status = engine.last_error + self._pending_connect = True + else: + self._pending_list = True + elif engine.session is None: + self._pending_status = engine.last_error or "session ended" + self._pending_connect = True + elif engine.last_error: + self._pending_status = engine.last_error + except Exception as err: + self._pending_status = "poll: %s" % err + self._poll_at = ticks_ms() + self._poll_busy = False + + self._run_bg(_work) + + # list ----------------------------------------------------------------- + + def _goto_list(self): + self.slideshow = False + self._show_page("list") + + def _build_list(self, parent): + W, H = self._content_metrics() + pad = self.pad + gap = pad + engine = self.engine + items = engine.items + n = len(items) + btn_h = max(36, H // 11) + x0 = pad + w = W - 2 * pad + row_h = max(48, min(104, H // 6)) + inner = max(3, pad // 2) + tile = row_h - 2 * inner + avail = H - btn_h - gap + page_size = max(1, (avail + gap) // (row_h + gap)) + self.page_size = page_size + if self.list_offset >= n or self.list_offset < 0: + self.list_offset = 0 + offset = self.list_offset + + labels = [("PICK", "accent", self._start_pick)] + if n: + labels.append(("SLIDES", "ui", self._start_slideshow)) + if n > page_size: + labels.append(("MORE", "ui", self._next_page)) + count = len(labels) + bw = (w - (count - 1) * gap) // count + for i, (text, role, cb) in enumerate(labels): + self._button(parent, text, x0 + i * (bw + gap), 0, bw, btn_h, role, cb) + y = btn_h + gap + + self._set_title("Google Photos") + self._set_count(engine.count_label() if n else "") + notice = getattr(engine, "sim_notice", "") or "" + if not n: + msg = lv.label(parent) + msg.set_width(w) + self._long_mode(msg, "WRAP") + msg.set_style_text_color(_hex(_COL["muted"]), 0) + _apply_font(msg, self.font) + msg.set_text("No photos picked yet.\nTap PICK, then scan the code with your phone.") + msg.align(lv.ALIGN.CENTER, 0, y // 2) + self._set_status(notice or engine.last_error or "") + return + + # Drop tiles that are not on this page (bounded RAM on MCUs). + page_items = items[offset : offset + page_size] + keep = set(it.get("id") for it in page_items) + for key in list(self._images.keys()): + if key not in keep: + self._images.pop(key, None) + name_x = tile + 2 * inner + name_w = w - name_x - inner + for i, item in enumerate(page_items): + key = item.get("id", "") + row = lv.obj(parent) + row.set_size(w, row_h) + row.set_pos(x0, y + i * (row_h + gap)) + self._panel(row, _COL["row"] if i % 2 == 0 else _COL["row_alt"], self.radius) + _no_scroll(row) + _add_flag(row, "CLICKABLE") + try: + row.set_style_bg_color(_hex(_shade(_COL["row"], 1.3)), lv.STATE.PRESSED) + except Exception: + pass + + def _open(_e, _idx=offset + i): + self._open_view(_idx) + + row.add_event_cb(_open, lv.EVENT.CLICKED, None) + + img = lv.image(row) + img.set_size(tile, tile) + img.set_pos(inner, inner) + self._panel(img, _COL["tile"], max(2, self.radius // 2)) + _remove_flag(img, "CLICKABLE") + try: + img.set_inner_align(lv.image.ALIGN.CENTER) + except Exception: + pass + ph = lv.label(row) + ph.set_text(_sym("IMAGE", "#")) + ph.set_style_text_color(_hex(_COL["muted"]), 0) + _apply_font(ph, self.font) + ph.align_to(img, lv.ALIGN.CENTER, 0, 0) + + name = lv.label(row) + name.set_width(name_w) + self._long_mode(name, "DOTS") + name.set_style_text_color(_hex(_COL["text"]), 0) + _apply_font(name, self.font) + name.set_text(item.get("filename") or key or "photo") + name.align(lv.ALIGN.TOP_LEFT, name_x, inner) + + meta = lv.label(row) + meta.set_width(name_w) + self._long_mode(meta, "DOTS") + meta.set_style_text_color(_hex(_COL["muted"]), 0) + _apply_font(meta, self.font_sm) + meta.set_text(self._meta_text(item)) + meta.align(lv.ALIGN.BOTTOM_LEFT, name_x, -inner) + + self._rows[key] = (img, ph) + cached = self._images.get(key) + if cached is not None: + self._apply_tile(key, cached[0]) + else: + self._queue_thumb(item, tile) + + last = min(n, offset + page_size) + if n > page_size: + self._set_status("photos %d-%d of %d%s" % (offset + 1, last, n, (" " + notice) if notice else "")) + else: + self._set_status(notice or "tap a photo to view it") + + @staticmethod + def _meta_text(item): + parts = [] + d = date_label(item.get("createTime")) + t = time_label(item.get("createTime")) + if d: + parts.append(d + (" " + t if t else "")) + if item.get("width") and item.get("height"): + parts.append("%dx%d" % (item["width"], item["height"])) + if (item.get("type") or "").upper() == "VIDEO": + parts.append("video") + return " ".join(parts) + + def _next_page(self): + n = len(self.engine.items) + if not n: + return + self.list_offset += self.page_size + if self.list_offset >= n: + self.list_offset = 0 + self._show_page("list") + + def _queue_thumb(self, item, size): + key = item.get("id", "") + if not key or key in self._thumb_busy: + return + self._thumb_busy[key] = True + engine = self.engine + + def _work(): + path = "" + try: + path = engine.thumbnail_path(item, size, size, crop=True) or "" + except Exception as err: + engine.last_error = "thumbnail: %s" % err + if not path: + self._pending_status = engine.last_error + self._pending_thumbs.append((key, path)) + + self._run_bg(_work) + + def _apply_tile(self, key, dsc): + row = self._rows.get(key) + if row is None: + return + img, ph = row + try: + img.set_src(dsc) + _set_hidden(ph, True) + except Exception: + pass + + # view ----------------------------------------------------------------- + + def _open_view(self, idx): + n = len(self.engine.items) + if not n: + return + self.index = idx % n + self._show_page("view") + + def _start_slideshow(self): + if not self.engine.items: + return + self.slideshow = True + self.index = 0 + self._show_page("view") + + def _build_view(self, parent): + W, H = self._content_metrics() + pad = self.pad + gap = pad + bar_h = max(36, H // 11) + img_h = H - bar_h - gap + self.view_w = W + self.view_h = img_h + + frame = lv.obj(parent) + frame.set_size(W, img_h) + frame.set_pos(0, 0) + self._panel(frame, _COL["black"]) + _no_scroll(frame) + _add_flag(frame, "CLICKABLE") + + def _tap(_e): + self._go(1) + + frame.add_event_cb(_tap, lv.EVENT.CLICKED, None) + + img = lv.image(frame) + img.set_size(W, img_h) + img.set_pos(0, 0) + _remove_flag(img, "CLICKABLE") + try: + img.set_inner_align(lv.image.ALIGN.CENTER) + except Exception: + pass + self.view_img = img + + msg = lv.label(frame) + msg.set_width(W - 2 * pad) + self._long_mode(msg, "WRAP") + msg.set_style_text_color(_hex(_COL["muted"]), 0) + _apply_font(msg, self.font) + ta = getattr(lv, "TEXT_ALIGN", None) + center = getattr(ta, "CENTER", None) if ta is not None else None + if center is not None: + try: + msg.set_style_text_align(center, 0) + except Exception: + pass + msg.set_text("loading...") + msg.center() + self.view_msg = msg + + w = W - 2 * pad + bw = (w - 3 * gap) // 4 + y = img_h + gap + specs = ( + ("BACK", "ui", self._goto_list), + (_sym("PREV", "<"), "ui", lambda: self._go(-1)), + ("", "accent", self._toggle_slideshow), + (_sym("NEXT", ">"), "ui", lambda: self._go(1)), + ) + for i, (text, role, cb) in enumerate(specs): + _btn, lbl = self._button(parent, text, pad + i * (bw + gap), y, bw, bar_h, role, cb) + if i == 2: + self.play_lbl = lbl + self._refresh_play_face() + self._show_view_meta() + self._load_view(self.index) + + def _refresh_play_face(self): + if self.play_lbl is None: + return + try: + self.play_lbl.set_text(_sym("PAUSE", "||") if self.slideshow else _sym("PLAY", ">")) + self.play_lbl.center() + except Exception: + pass + + def _toggle_slideshow(self): + self.slideshow = not self.slideshow + self._view_loaded_at = ticks_ms() if self.slideshow else None + self._refresh_play_face() + + def _show_view_meta(self): + items = self.engine.items + n = len(items) + if not n: + return + item = items[self.index] + self._set_title(item.get("filename") or "photo") + self._set_count("%d/%d" % (self.index + 1, n)) + self._set_status(self._meta_text(item) or " ") + + def _go(self, delta): + n = len(self.engine.items) + if not n or self.page != "view": + return + self.index = (self.index + delta) % n + self._show_view_meta() + self._load_view(self.index) + + def _load_view(self, idx): + items = self.engine.items + if idx >= len(items) or self.view_img is None: + return + item = items[idx] + self._view_ref = None + self._view_loaded_at = None + gc.collect() + path = self.engine.cached_path(item, self.view_w, self.view_h) + if path: + self._apply_view(idx, path) + return + try: + self.view_img.set_src(None) + except Exception: + pass + _set_hidden(self.view_msg, False) + try: + self.view_msg.set_text("loading...") + except Exception: + pass + self._view_fetching = idx + engine = self.engine + vw, vh = self.view_w, self.view_h + + def _work(): + got = "" + try: + got = engine.thumbnail_path(item, vw, vh) or "" + except Exception as err: + engine.last_error = "photo: %s" % err + self._pending_view = (idx, got) + + self._run_bg(_work) + + def _apply_view(self, idx, path): + if self.page != "view" or idx != self.index or self.view_img is None: + return + self._view_fetching = None + if not path: + self._say_view(self.engine.last_error or "photo unavailable") + self._view_loaded_at = ticks_ms() # keep the slideshow moving + return + dsc, data, w, h = image_descriptor(path) + if dsc is None: + if not jpeg_supported(): + self._say_view("no JPEG decoder in this firmware\n(see README: jpegio)") + else: + self._say_view("could not decode image") + self._view_loaded_at = ticks_ms() + return + self._view_ref = (dsc, data) + try: + self.view_img.set_src(dsc) + _set_hidden(self.view_msg, True) + except Exception as err: + self._say_view("display: %s" % err) + self._view_loaded_at = ticks_ms() + self._prefetch(idx + 1) + + def _say_view(self, text): + if self.view_msg is None: + return + try: + self.view_msg.set_text(text) + _set_hidden(self.view_msg, False) + except Exception: + pass + + def _prefetch(self, idx): + """Warm the cache for the next photo so PLAY / NEXT is instant.""" + items = self.engine.items + if len(items) < 2: + return + idx = idx % len(items) + item = items[idx] + if self._prefetched == item.get("id"): + return + self._prefetched = item.get("id") + if self.engine.cached_path(item, self.view_w, self.view_h): + return + engine = self.engine + vw, vh = self.view_w, self.view_h + + def _work(): + try: + engine.thumbnail_path(item, vw, vh) + except Exception: + pass + + self._run_bg(_work) + + # ----- background queue + pump ---------------------------------------- + + def _run_bg(self, fn): + """Queue ``fn`` for the LVGL soft-pump (no ``_thread``).""" + self._bg_q.append(fn) + return True + + def _drain_bg(self): + """Run at most one queued job per pump tick.""" + if self._bg_busy or not self._bg_q: + return + self._bg_busy = True + job = self._bg_q.pop(0) + try: + job() + except Exception as err: + self._pending_status = "error: %s" % err + finally: + self._bg_busy = False + + def _pump(self, _timer=None): + self._drain_bg() + + if self._pending_status is not None: + text = self._pending_status + self._pending_status = None + if text: + self._set_status(text) + + if self._pending_list: + self._pending_list = False + self.list_offset = 0 + self.index = 0 + self._show_page("list") + n = len(self.engine.items) + self._set_status("%d picked" % n if n else "nothing was picked") + return + + if self._pending_connect: + self._pending_connect = False + if self.page == "connect": + self._show_page("connect") + + while self._pending_thumbs: + key, path = self._pending_thumbs.pop(0) + self._thumb_busy.pop(key, None) + if not path: + continue + dsc, data, _w, _h = image_descriptor(path) + if dsc is None: + if not jpeg_supported() and path.endswith(".jpg"): + self._set_status("no JPEG decoder in this firmware (see README)") + continue + self._images[key] = (dsc, data) + self._apply_tile(key, dsc) + + if self._pending_view is not None: + idx, path = self._pending_view + self._pending_view = None + self._apply_view(idx, path) + + if self.page == "connect": + session = self.engine.session + if session and not self._poll_busy and not self._poll_timed_out: + started = self._session_started + timeout_ms = int(float(session.get("timeoutIn") or 1800.0) * 1000) + if started is not None and ticks_diff(ticks_ms(), started) > timeout_ms: + self._poll_timed_out = True + self._set_status("picker timed out - tap RETRY") + else: + interval = int(self.engine.poll_interval_s() * 1000) + if self._poll_at is None or ticks_diff(ticks_ms(), self._poll_at) >= interval: + self._poll_pick() + + if ( + self.page == "view" + and self.slideshow + and self._view_loaded_at is not None + and self._view_fetching is None + and not self._bg_q + and ticks_diff(ticks_ms(), self._view_loaded_at) >= self.slideshow_s * 1000 + ): + self._go(1) + + +# Most recent front end instance (harnesses / screenshots / REPL poking). +_LAST = None + + +def create(engine=None, start_page="list"): + """Build the LVGL front end (does not call ``App.run``).""" + global _LAST + _LAST = _GPhotosLvgl(engine=engine, start_page=start_page) + return _LAST + + +def run(engine=None, start_page="list"): + """Create the UI and let the app run itself.""" + create(engine=engine, start_page=start_page) + + +# Direct import / example kit: auto-start. ``google_photos`` sets +# ``gphotos_engine._LAUNCHER_OWNS_RUN`` and calls ``run()`` itself. +import gphotos_engine as _gphotos_engine # noqa: E402 + +if not getattr(_gphotos_engine, "_LAUNCHER_OWNS_RUN", False): + _engine = make_engine() + _engine.restore() + run(engine=_engine, start_page="list" if _engine.items else "connect") diff --git a/lib/examples/google_photos/gphotos_sim.py b/lib/examples/google_photos/gphotos_sim.py new file mode 100644 index 00000000..7a25182f --- /dev/null +++ b/lib/examples/google_photos/gphotos_sim.py @@ -0,0 +1,269 @@ +# gallery: skip +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +""" +gphotos_sim +==================================================== +Duck-typed Picker API simulator for gallery / offline demos. + +``make_engine()`` returns :class:`GPhotosSimEngine` when: + +* ``GPHOTOS_SIM`` is truthy (explicit), or +* PyScript (``emscripten`` / ``webassembly`` / ``import pyscript``), or +* no tokens file with ``client_id`` + ``refresh_token`` exists (see + ``tools/gphotos_auth.py``), or +* no usable network is detected. + +``GPHOTOS_SIM=0`` forces the real :class:`gphotos_engine.GPhotosEngine`. + +The sim pretends the user already picked a handful of photos, so the launcher +opens on the list. PICK shows a fake QR code and "picks" again by itself after +a few polls. Thumbnails come from ``assets/sim_NN_.jpg`` when those +files are present (git checkout); the front end paints a placeholder tile +when they are not (MIP / gallery installs carry Python files only). +""" + +import sys + +from gphotos_engine import GPhotosEngine, credentials_ok, load_tokens, socket as _engine_socket +from gphotos_engine import _env_get, _exists + +_PKG = __file__.replace("\\", "/").rsplit("/", 1)[0] if "/" in __file__.replace("\\", "/") else "." +ASSETS_DIR = _PKG + "/assets" + +SIM_PICKER_URI = "https://photospicker.google.com/simulator" + +# Canned "picked" photos. Even entries are landscape, odd entries portrait. +_SIM_ITEMS = ( + ("IMG_20240503_142211.jpg", "2024-05-03T14:22:11Z", 4032, 3024, "sim_01"), + ("PXL_20240612_091530.jpg", "2024-06-12T09:15:30Z", 3024, 4032, "sim_02"), + ("IMG_20240704_203045.jpg", "2024-07-04T20:30:45Z", 4000, 3000, "sim_03"), + ("PXL_20240815_171012.jpg", "2024-08-15T17:10:12Z", 3000, 4000, "sim_04"), + ("IMG_20240921_120000.jpg", "2024-09-21T12:00:00Z", 4032, 3024, "sim_05"), + ("PXL_20241031_183322.jpg", "2024-10-31T18:33:22Z", 3024, 4032, "sim_06"), + ("IMG_20241225_101500.jpg", "2024-12-25T10:15:00Z", 4032, 3024, "sim_01"), + ("PXL_20250101_000102.jpg", "2025-01-01T00:01:02Z", 3024, 4032, "sim_02"), +) + +# Plaque status when the sim runs because no credentials / network exist. +NO_CREDENTIALS_PLAQUE = "no tokens file - running simulator" +NO_NETWORK_PLAQUE = "no network detected - running simulator" + + +def _is_pyscript(): + if getattr(sys, "platform", "") in ("emscripten", "webassembly"): + return True + try: + import pyscript # noqa: F401 + + return True + except ImportError: + return False + + +def _network_available(): + """Best-effort: station Wi-Fi up on MCUs, any route on desktop hosts.""" + try: + import network + + try: + return bool(network.WLAN(network.STA_IF).isconnected()) + except Exception: + return True # a network module without WLAN: assume wired + except ImportError: + pass + try: + import wifi + + return bool(getattr(wifi.radio, "ipv4_address", None)) + except ImportError: + pass + if _engine_socket is None: + return False + try: + s = _engine_socket.socket(_engine_socket.AF_INET, _engine_socket.SOCK_DGRAM) + try: + s.connect(("8.8.8.8", 80)) # no packet is sent for UDP connect + return True + finally: + s.close() + except Exception: + return False + + +def sim_reason(tokens=None): + """Why the sim should run, or ``None`` for the real engine. + + Returns ``"env"``, ``"pyscript"``, ``"no_credentials"``, ``"no_network"`` + or ``None``. Explicit ``GPHOTOS_SIM`` wins. + """ + raw = _env_get("GPHOTOS_SIM") + if raw is not None and raw != "": + return "env" if str(raw).strip().lower() not in ("0", "false", "no", "off") else None + if _is_pyscript(): + return "pyscript" + if not credentials_ok(load_tokens() if tokens is None else tokens): + return "no_credentials" + if not _network_available(): + return "no_network" + return None + + +def want_sim(): + return sim_reason() is not None + + +def make_engine(**kwargs): + """Return :class:`GPhotosSimEngine` or :class:`GPhotosEngine` per :func:`sim_reason`.""" + reason = sim_reason(kwargs.get("tokens")) + if reason: + return GPhotosSimEngine(reason=reason, **kwargs) + return GPhotosEngine(**kwargs) + + +def _sim_items(rotate=0): + items = [] + n = len(_SIM_ITEMS) + for i in range(n): + name, created, w, h, asset = _SIM_ITEMS[(i + rotate) % n] + items.append( + { + "id": "sim-%s" % name, + "filename": name, + "createTime": created, + "type": "PHOTO", + "mime": "image/jpeg", + "width": w, + "height": h, + "baseUrl": "sim://" + asset, + "asset": asset, + "portrait": h > w, + } + ) + return items + + +class GPhotosSimEngine(GPhotosEngine): + """In-memory Picker stand-in — same surface as :class:`GPhotosEngine`.""" + + sim = True + + def __init__(self, reason="env", **kwargs): + super().__init__(**kwargs) + self.sim_reason = reason or "env" + if self.sim_reason == "no_credentials": + self.sim_notice = NO_CREDENTIALS_PLAQUE + elif self.sim_reason == "no_network": + self.sim_notice = NO_NETWORK_PLAQUE + else: + self.sim_notice = "simulator" + self._polls = 0 + self._picks = 0 + self.prefs = {} + + # No network, no files: everything below is in-memory. + + def has_credentials(self): + return True + + def ensure_token(self, force=False): + self.access_token = "sim" + return True + + def _save_prefs(self): + return True + + def _load_prefs(self): + if self.prefs is None: + self.prefs = {} + return self.prefs + + def create_session(self): + self._polls = 0 + self.session = { + "id": "sim-session-%d" % (self._picks + 1), + "pickerUri": SIM_PICKER_URI, + "expireTime": "", + "mediaItemsSet": False, + "pollInterval": 2.0, + "timeoutIn": 1800.0, + } + self.items = [] + self._urls_at = None + self.last_error = "" + return self.session + + def poll_session(self): + if not self.session: + self.last_error = "no session" + return False + self._polls += 1 + done = self._polls >= 3 + self.session["mediaItemsSet"] = done + return done + + def delete_session(self): + self.session = None + return True + + def list_items(self): + if not self.session: + self.last_error = "no session" + return None + self._picks += 1 + self.items = _sim_items(rotate=self._picks % len(_SIM_ITEMS)) + self._urls_at = 0 + self.last_error = "" + return self.items + + def urls_valid(self): + return bool(self.items) + + def refresh_urls(self): + return bool(self.items) + + def restore(self): + self.session = { + "id": "sim-session-0", + "pickerUri": SIM_PICKER_URI, + "expireTime": "", + "mediaItemsSet": True, + "pollInterval": 2.0, + "timeoutIn": 1800.0, + } + self.items = _sim_items() + self._urls_at = 0 + return True + + def forget(self, clear_cache=False): + self.session = None + self.items = [] + + def asset_path(self, item, w, h, crop=False): + """Best bundled JPEG for the requested box, or ``""`` when absent.""" + base = item.get("asset") or "sim_01" + if crop: + tag = "96" if min(int(w), int(h)) >= 96 else "64" + else: + portrait = bool(item.get("portrait")) + sizes = ((240, 320), (120, 160)) if portrait else ((320, 240), (160, 120)) + tag = None + for cw, ch in sizes: + if cw <= int(w) and ch <= int(h): + tag = "%dx%d" % (cw, ch) + break + if tag is None: + cw, ch = sizes[-1] + tag = "%dx%d" % (cw, ch) + path = "%s/%s_%s.jpg" % (ASSETS_DIR, base, tag) + return path if _exists(path) else "" + + def thumbnail_path(self, item, w, h, crop=False): + if not item or not item.get("id"): + self.last_error = "no item" + return "" + path = self.asset_path(item, w, h, crop) + if not path: + self.last_error = "simulator: sample images not installed" + return path diff --git a/packages/examples.json b/packages/examples.json index bf889196..f93fc0db 100644 --- a/packages/examples.json +++ b/packages/examples.json @@ -424,6 +424,34 @@ "drum_seq/panel.py", "github:PyDevices/pydevices-examples/lib/examples/drum_seq/panel.py" ], + [ + "google_photos/__init__.py", + "github:PyDevices/pydevices-examples/lib/examples/google_photos/__init__.py" + ], + [ + "google_photos/__main__.py", + "github:PyDevices/pydevices-examples/lib/examples/google_photos/__main__.py" + ], + [ + "google_photos/google_photos.py", + "github:PyDevices/pydevices-examples/lib/examples/google_photos/google_photos.py" + ], + [ + "google_photos/gphotos_engine.py", + "github:PyDevices/pydevices-examples/lib/examples/google_photos/gphotos_engine.py" + ], + [ + "google_photos/gphotos_lvgl.py", + "github:PyDevices/pydevices-examples/lib/examples/google_photos/gphotos_lvgl.py" + ], + [ + "google_photos/gphotos_sim.py", + "github:PyDevices/pydevices-examples/lib/examples/google_photos/gphotos_sim.py" + ], + [ + "google_photos/assets/gen_sim_assets.py", + "github:PyDevices/pydevices-examples/lib/examples/google_photos/assets/gen_sim_assets.py" + ], [ "noto_fonts/NotoSansMono_32.py", "github:PyDevices/pydevices-examples/lib/examples/noto_fonts/NotoSansMono_32.py" diff --git a/packages/google_photos.json b/packages/google_photos.json new file mode 100644 index 00000000..add4deb3 --- /dev/null +++ b/packages/google_photos.json @@ -0,0 +1,33 @@ +{ + "urls": [ + [ + "google_photos/__init__.py", + "../lib/examples/google_photos/__init__.py" + ], + [ + "google_photos/__main__.py", + "../lib/examples/google_photos/__main__.py" + ], + [ + "google_photos/google_photos.py", + "../lib/examples/google_photos/google_photos.py" + ], + [ + "google_photos/gphotos_engine.py", + "../lib/examples/google_photos/gphotos_engine.py" + ], + [ + "google_photos/gphotos_lvgl.py", + "../lib/examples/google_photos/gphotos_lvgl.py" + ], + [ + "google_photos/gphotos_sim.py", + "../lib/examples/google_photos/gphotos_sim.py" + ], + [ + "google_photos/assets/gen_sim_assets.py", + "../lib/examples/google_photos/assets/gen_sim_assets.py" + ] + ], + "version": "0.0.1" +} diff --git a/pydevices-examples.toml b/pydevices-examples.toml index 3269bdb4..4824d5ed 100644 --- a/pydevices-examples.toml +++ b/pydevices-examples.toml @@ -77,6 +77,13 @@ "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/font_simpletest.py" = "/examples/font_simpletest.py" "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/fonts.py" = "/examples/fonts.py" "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/framebuf_simpletest.py" = "/examples/framebuf_simpletest.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/google_photos/__init__.py" = "/examples/google_photos/__init__.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/google_photos/__main__.py" = "/examples/google_photos/__main__.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/google_photos/assets/gen_sim_assets.py" = "/examples/google_photos/assets/gen_sim_assets.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/google_photos/google_photos.py" = "/examples/google_photos/google_photos.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/google_photos/gphotos_engine.py" = "/examples/google_photos/gphotos_engine.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/google_photos/gphotos_lvgl.py" = "/examples/google_photos/gphotos_lvgl.py" +"https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/google_photos/gphotos_sim.py" = "/examples/google_photos/gphotos_sim.py" "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/graphics_simpletest.py" = "/examples/graphics_simpletest.py" "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/hello.py" = "/examples/hello.py" "https://raw.githubusercontent.com/PyDevices/pydevices-examples/main/lib/examples/joystick_list_select.py" = "/examples/joystick_list_select.py" diff --git a/tests/test_gphotos_engine.py b/tests/test_gphotos_engine.py new file mode 100644 index 00000000..d90c87ee --- /dev/null +++ b/tests/test_gphotos_engine.py @@ -0,0 +1,602 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Unit tests for ``lib/examples/google_photos`` (engine, simulator) and +``tools/gphotos_auth.py`` -- all network traffic is faked.""" + +import hashlib +import io +import json +import os +import sys +import tempfile +import threading +import unittest +from unittest import mock +import urllib.request + +import _env # noqa: F401 + +_REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) +_PKG = os.path.join(_REPO, "lib", "examples", "google_photos") +if _PKG not in sys.path: + sys.path.insert(0, _PKG) +_TOOLS = os.path.join(_REPO, "tools") +if _TOOLS not in sys.path: + sys.path.insert(0, _TOOLS) + +import gphotos_auth # noqa: E402 +import gphotos_engine as ge # noqa: E402 +import gphotos_sim as gs # noqa: E402 + +JPEG = b"\xff\xd8\xff\xe0" + b"\x00" * 60 +PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 60 +TOKENS = {"client_id": "cid", "client_secret": "sec", "refresh_token": "rt"} + + +def _picked(i, base="https://lh3.example/photo%d"): + return { + "id": "item-%d" % i, + "createTime": "2024-05-%02dT14:22:11Z" % (i + 1), + "type": "PHOTO", + "mediaFile": { + "baseUrl": base % i, + "mimeType": "image/jpeg", + "filename": "IMG_%d.jpg" % i, + "mediaFileMetadata": {"width": "4032", "height": "3024"}, + }, + } + + +class FakeTransport: + """Stands in for ``gphotos_engine._transport``; routes by (method, url prefix).""" + + def __init__(self): + self.calls = [] + self.routes = [] # (method, url_prefix, handler) + + def route(self, method, prefix, handler): + self.routes.append((method, prefix, handler)) + + def __call__(self, method, url, headers, body, timeout, sink=None): + self.calls.append((method, url, dict(headers or {}), body)) + for m, prefix, handler in self.routes: + if m == method and url.startswith(prefix): + status, payload = handler(url, headers or {}, body) + if isinstance(payload, (dict, list)): + payload = json.dumps(payload).encode("utf-8") + if sink is not None: + sink(payload) + return status, len(payload) + return status, payload + raise AssertionError("unexpected %s %s" % (method, url)) + + +def token_ok(url, headers, body): + return 200, {"access_token": "AT1", "expires_in": 3600, "token_type": "Bearer"} + + +class HelperTests(unittest.TestCase): + def test_quote_and_form(self): + self.assertEqual(ge.quote("a b/c~d"), "a%20b%2Fc~d") + self.assertEqual(ge.form_encode([("a", "1 2"), ("b", "x&y")]), "a=1%202&b=x%26y") + + def test_labels_and_durations(self): + self.assertEqual(ge.parse_duration_s("5s"), 5.0) + self.assertEqual(ge.parse_duration_s("1.5s"), 1.5) + self.assertEqual(ge.parse_duration_s(None, 7.0), 7.0) + self.assertEqual(ge.parse_duration_s("junk", 3.0), 3.0) + self.assertEqual(ge.date_label("2024-05-03T14:22:11Z"), "2024-05-03") + self.assertEqual(ge.time_label("2024-05-03T14:22:11Z"), "14:22") + self.assertEqual(ge.time_label("2024-05-03"), "") + self.assertEqual(len(ge.item_hash("anything")), 8) + self.assertNotEqual(ge.item_hash("a"), ge.item_hash("b")) + + def test_thumb_url_and_sniff(self): + self.assertEqual(ge.thumb_url("https://x/y", 64, 64, True), "https://x/y=w64-h64-c") + self.assertEqual(ge.thumb_url("https://x/y", 320, 400), "https://x/y=w320-h400") + self.assertEqual(ge.thumb_url("", 1, 1), "") + self.assertEqual(ge.sniff_ext(JPEG), ".jpg") + self.assertEqual(ge.sniff_ext(PNG), ".png") + self.assertEqual(ge.sniff_ext(b""), "") + + def test_normalize_item(self): + item = ge.normalize_item(_picked(3)) + self.assertEqual(item["id"], "item-3") + self.assertEqual(item["filename"], "IMG_3.jpg") + self.assertEqual(item["width"], 4032) + self.assertEqual(item["baseUrl"], "https://lh3.example/photo3") + self.assertEqual(ge.normalize_item("nope"), {}) + + def test_path_overrides(self): + with mock.patch.dict( + os.environ, {"GPHOTOS_TOKENS": "/tmp/t.json", "GPHOTOS_CACHE": "/tmp/c"} + ): + self.assertEqual(ge.tokens_path(), "/tmp/t.json") + self.assertEqual(ge.cache_dir(), "/tmp/c") + with mock.patch.dict(os.environ, {"HOME": "/home/x"}, clear=False): + os.environ.pop("GPHOTOS_PREFS", None) + self.assertEqual(ge.prefs_path(), "/home/x/.gphotos_prefs") + + +class EngineTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.transport = FakeTransport() + patcher = mock.patch.object(ge, "_transport", self.transport) + patcher.start() + self.addCleanup(patcher.stop) + + def make(self, tokens=TOKENS, **kw): + return ge.GPhotosEngine( + tokens=dict(tokens), + prefs_file=os.path.join(self.tmp.name, "prefs"), + cache_path=os.path.join(self.tmp.name, "cache"), + **kw, + ) + + +class TokenTests(EngineTestCase): + def test_missing_credentials(self): + eng = self.make(tokens={}) + self.assertFalse(eng.has_credentials()) + self.assertFalse(eng.ensure_token()) + self.assertIn("gphotos_auth", eng.last_error) + self.assertEqual(self.transport.calls, []) + + def test_refresh_then_cache_then_force(self): + self.transport.route("POST", ge.TOKEN_URL, token_ok) + eng = self.make() + self.assertTrue(eng.ensure_token()) + self.assertEqual(eng.access_token, "AT1") + _method, _url, headers, body = self.transport.calls[0] + self.assertEqual(headers["Content-Type"], "application/x-www-form-urlencoded") + self.assertIn(b"grant_type=refresh_token", body) + self.assertIn(b"refresh_token=rt", body) + self.assertIn(b"client_secret=sec", body) + self.assertTrue(eng.ensure_token()) + self.assertEqual(len(self.transport.calls), 1) + self.assertTrue(eng.ensure_token(force=True)) + self.assertEqual(len(self.transport.calls), 2) + + def test_refresh_failure_reports_google_error(self): + self.transport.route( + "POST", + ge.TOKEN_URL, + lambda u, h, b: ( + 400, + {"error": "invalid_grant", "error_description": "Token has been expired"}, + ), + ) + eng = self.make() + self.assertFalse(eng.ensure_token()) + self.assertEqual( + eng.last_error, "token refresh: HTTP 400 invalid_grant: Token has been expired" + ) + + def test_network_failure(self): + def boom(*a, **k): + raise ge.GPhotosError("dns") + + with mock.patch.object(ge, "_transport", boom): + eng = self.make() + self.assertFalse(eng.ensure_token()) + self.assertIn("dns", eng.last_error) + + +class SessionTests(EngineTestCase): + def setUp(self): + super().setUp() + self.transport.route("POST", ge.TOKEN_URL, token_ok) + self.session = { + "id": "sess/1", + "pickerUri": "https://photospicker.google.com/s/1", + "pollingConfig": {"pollInterval": "3s", "timeoutIn": "1800s"}, + "expireTime": "2030-01-01T00:00:00Z", + "mediaItemsSet": False, + } + self.polls = 0 + + def create(url, headers, body): + self.assertEqual(headers["Authorization"], "Bearer AT1") + self.assertEqual(body, b"{}") + return 200, self.session + + def poll(url, headers, body): + self.polls += 1 + data = dict(self.session) + data["mediaItemsSet"] = self.polls >= 2 + return 200, data + + def listing(url, headers, body): + if "pageToken=" in url: + return 200, {"mediaItems": [_picked(2)]} + return 200, {"mediaItems": [_picked(0), _picked(1)], "nextPageToken": "p2"} + + self.transport.route("POST", ge.PICKER_BASE + "/sessions", create) + self.transport.route("GET", ge.PICKER_BASE + "/sessions/", poll) + self.transport.route("GET", ge.PICKER_BASE + "/mediaItems?", listing) + self.transport.route("GET", "https://lh3.example/photo", lambda u, h, b: (200, JPEG)) + + def test_full_flow_and_prefs_roundtrip(self): + eng = self.make() + s = eng.create_session() + self.assertEqual(s["id"], "sess/1") + self.assertEqual(s["pollInterval"], 3.0) + self.assertEqual(eng.poll_interval_s(), 3.0) + self.assertFalse(eng.poll_session()) + self.assertTrue(eng.poll_session()) + items = eng.list_items() + self.assertEqual([i["id"] for i in items], ["item-0", "item-1", "item-2"]) + self.assertTrue(eng.urls_valid()) + list_calls = [c for c in self.transport.calls if "/mediaItems?" in c[1]] + self.assertEqual(len(list_calls), 2) + self.assertIn("sessionId=sess%2F1", list_calls[0][1]) + + # prefs: session + items persisted, urls not + with open(eng.prefs_file) as f: + prefs = json.load(f) + self.assertEqual(prefs["session"]["id"], "sess/1") + self.assertEqual(len(prefs["items"]), 3) + self.assertNotIn("baseUrl", prefs["items"][0]) + + other = self.make() + self.assertTrue(other.restore()) + self.assertEqual(other.session["pickerUri"], "https://photospicker.google.com/s/1") + self.assertEqual(len(other.items), 3) + self.assertFalse(other.urls_valid()) + self.assertEqual(other.count_label(), "3 photos") + + def test_thumbnail_download_cache_and_refresh(self): + eng = self.make() + eng.create_session() + eng.list_items() + item = eng.items[1] + path = eng.thumbnail_path(item, 64, 64, crop=True) + self.assertTrue(path.endswith("_64x64c.jpg"), path) + self.assertTrue(os.path.isfile(path)) + dl = [c for c in self.transport.calls if c[1].startswith("https://lh3.example")] + self.assertEqual(dl[-1][1], "https://lh3.example/photo1=w64-h64-c") + self.assertEqual(dl[-1][2]["Authorization"], "Bearer AT1") + n = len(self.transport.calls) + self.assertEqual(eng.thumbnail_path(item, 64, 64, crop=True), path) + self.assertEqual(len(self.transport.calls), n) # served from cache + + # restored engine has no urls: thumbnail fetch re-lists first + other = self.make() + other.restore() + before = len(self.transport.calls) + view = other.thumbnail_path(other.items[2], 320, 400) + self.assertTrue(view.endswith("_320x400.jpg")) + urls = [c[1] for c in self.transport.calls[before:]] + self.assertTrue(any("/mediaItems?" in u for u in urls)) + self.assertIn("https://lh3.example/photo2=w320-h400", urls) + + def test_png_and_bad_downloads(self): + self.transport.routes = [ + r for r in self.transport.routes if not r[1].startswith("https://lh3") + ] + self.transport.route("GET", "https://lh3.example/photo0", lambda u, h, b: (200, PNG)) + self.transport.route("GET", "https://lh3.example/photo1", lambda u, h, b: (403, b"denied")) + self.transport.route("GET", "https://lh3.example/photo2", lambda u, h, b: (200, b"")) + eng = self.make() + eng.create_session() + eng.list_items() + self.assertTrue(eng.thumbnail_path(eng.items[0], 96, 96, True).endswith(".png")) + self.assertEqual(eng.thumbnail_path(eng.items[1], 96, 96, True), "") + self.assertIn("403", eng.last_error) + self.assertEqual(eng.thumbnail_path(eng.items[2], 96, 96, True), "") + self.assertIn("not an image", eng.last_error) + leftovers = [n for n in os.listdir(eng.cache_path) if not n.endswith(".png")] + self.assertEqual(leftovers, []) + + def test_401_refreshes_token_once(self): + seen = {"n": 0} + + def flaky(url, headers, body): + seen["n"] += 1 + if seen["n"] == 1: + return 401, {"error": {"message": "expired"}} + return 200, self.session + + self.transport.routes = [ + r for r in self.transport.routes if not r[1].endswith("/sessions") + ] + self.transport.route("POST", ge.PICKER_BASE + "/sessions", flaky) + eng = self.make() + self.assertIsNotNone(eng.create_session()) + tokens = [c for c in self.transport.calls if c[1] == ge.TOKEN_URL] + self.assertEqual(len(tokens), 2) + + def test_session_gone(self): + eng = self.make() + eng.create_session() + self.transport.routes = [ + r for r in self.transport.routes if not r[1].endswith("/sessions/") + ] + self.transport.route( + "GET", + ge.PICKER_BASE + "/sessions/", + lambda u, h, b: (404, {"error": {"status": "NOT_FOUND", "message": "gone"}}), + ) + self.assertFalse(eng.poll_session()) + self.assertIsNone(eng.session) + self.assertIn("expired", eng.last_error) + self.assertIsNone(eng.list_items()) + self.assertEqual(eng.last_error, "no session") + + def test_api_error_message(self): + self.transport.routes = [ + r for r in self.transport.routes if not r[1].endswith("/sessions") + ] + self.transport.route( + "POST", + ge.PICKER_BASE + "/sessions", + lambda u, h, b: ( + 403, + {"error": {"status": "PERMISSION_DENIED", "message": "API disabled"}}, + ), + ) + eng = self.make() + self.assertIsNone(eng.create_session()) + self.assertEqual(eng.last_error, "create session: HTTP 403 API disabled") + + def test_max_items_and_eviction(self): + eng = self.make(max_items=2, cache_max=2) + eng.create_session() + self.assertEqual(len(eng.list_items()), 2) + for w in (32, 48, 64): + eng.thumbnail_path(eng.items[0], w, w, True) + self.assertLessEqual(len(os.listdir(eng.cache_path)), 2) + eng.forget(clear_cache=True) + self.assertEqual(os.listdir(eng.cache_path), []) + self.assertEqual(eng.items, []) + self.assertFalse(self.make().restore()) + + def test_delete_session(self): + self.transport.route("DELETE", ge.PICKER_BASE + "/sessions/", lambda u, h, b: (200, {})) + eng = self.make() + eng.create_session() + self.assertTrue(eng.delete_session()) + self.assertIsNone(eng.session) + self.assertEqual(self.transport.calls[-1][0], "DELETE") + + +class FakeSocketModule: + """Enough of ``socket`` for ``_socket_request`` (plain HTTP only).""" + + AF_INET = 2 + SOCK_STREAM = 1 + + def __init__(self, response): + self.response = response + self.sent = b"" + self.closed = False + + def getaddrinfo(self, host, port, family, socktype): + return [(self.AF_INET, socktype, 6, "", (host, port))] + + def socket(self, family, socktype, proto): + return self + + def settimeout(self, t): + pass + + def connect(self, addr): + self.addr = addr + + def sendall(self, data): + self.sent += bytes(data) + + def recv(self, n): + chunk, self.response = self.response[:n], self.response[n:] + return chunk + + def close(self): + self.closed = True + + +class SocketFallbackTests(unittest.TestCase): + def test_plain_http_roundtrip(self): + body = b'{"ok": true}' + raw = ( + b"HTTP/1.0 200 OK\r\nContent-Type: application/json\r\nContent-Length: %d\r\n\r\n" + % len(body) + + body + ) + fake = FakeSocketModule(raw) + with mock.patch.object(ge, "socket", fake): + status, out = ge._socket_request( + "POST", "http://api.test:8080/v1/x?y=1", {"Authorization": "Bearer t"}, b"{}", 5.0 + ) + self.assertEqual((status, out), (200, body)) + self.assertEqual(fake.addr, ("api.test", 8080)) + head = fake.sent.split(b"\r\n\r\n", 1)[0] + self.assertTrue(head.startswith(b"POST /v1/x?y=1 HTTP/1.0\r\nHost: api.test")) + self.assertIn(b"Authorization: Bearer t", head) + self.assertIn(b"Content-Length: 2", head) + self.assertTrue(fake.sent.endswith(b"{}")) + self.assertTrue(fake.closed) + + def test_streaming_sink_without_content_length(self): + raw = b"HTTP/1.0 200 OK\r\n\r\n" + JPEG + fake = FakeSocketModule(raw) + chunks = [] + with mock.patch.object(ge, "socket", fake): + status, n = ge._socket_request("GET", "http://h/p", {}, None, 5.0, sink=chunks.append) + self.assertEqual((status, n), (200, len(JPEG))) + self.assertEqual(b"".join(chunks), JPEG) + + +class RequestsBackendTests(unittest.TestCase): + """The MicroPython ``requests`` path (urllib disabled).""" + + class Resp: + def __init__(self, status, data): + self.status_code = status + self.content = data + self.raw = io.BytesIO(data) + self.closed = False + + def close(self): + self.closed = True + + def test_request_and_stream(self): + calls = [] + resp = [None] + + class Requests: + @staticmethod + def request(method, url, data=None, headers=None, stream=None, timeout=None): + calls.append((method, url, data, headers, stream, timeout)) + resp[0] = RequestsBackendTests.Resp(200, JPEG) + return resp[0] + + with mock.patch.object(ge, "_urllib", lambda: None), mock.patch.object( + ge, "_find_requests", lambda: Requests + ): + self.assertEqual( + ge.http_request("GET", "https://h/x", headers={"A": "b"}), (200, JPEG) + ) + self.assertEqual(calls[-1][3], {"A": "b"}) + self.assertEqual(calls[-1][4], False) + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "f.bin") + self.assertEqual(ge.http_download("https://h/x", path), (200, len(JPEG))) + with open(path, "rb") as f: + self.assertEqual(f.read(), JPEG) + self.assertEqual(calls[-1][4], True) + self.assertTrue(resp[0].closed) + + +class SimTests(unittest.TestCase): + def test_sim_reason(self): + with mock.patch.dict(os.environ, {"GPHOTOS_SIM": "1"}): + self.assertEqual(gs.sim_reason(TOKENS), "env") + with mock.patch.dict(os.environ, {"GPHOTOS_SIM": "0"}): + self.assertIsNone(gs.sim_reason(TOKENS)) + with mock.patch.dict(os.environ, {"GPHOTOS_SIM": ""}): + os.environ.pop("GPHOTOS_SIM") + self.assertEqual(gs.sim_reason({}), "no_credentials") + with mock.patch.object(gs, "_network_available", lambda: False): + self.assertEqual(gs.sim_reason(TOKENS), "no_network") + with mock.patch.object(gs, "_network_available", lambda: True): + self.assertIsNone(gs.sim_reason(TOKENS)) + + def test_sim_flow(self): + with mock.patch.dict(os.environ, {"GPHOTOS_SIM": "1"}): + eng = gs.make_engine(tokens={}) + self.assertTrue(eng.sim) + self.assertTrue(eng.restore()) + first = eng.items[0]["id"] + self.assertTrue(eng.ensure_token()) + s = eng.create_session() + self.assertEqual(eng.items, []) + self.assertEqual(s["pickerUri"], gs.SIM_PICKER_URI) + self.assertFalse(eng.poll_session()) + self.assertFalse(eng.poll_session()) + self.assertTrue(eng.poll_session()) + items = eng.list_items() + self.assertEqual(len(items), len(gs._SIM_ITEMS)) + self.assertNotEqual(items[0]["id"], first) + self.assertTrue(eng.urls_valid()) + # bundled assets (git checkout) resolve by size class + tile = eng.thumbnail_path(items[0], 56, 56, crop=True) + self.assertTrue(tile.endswith("_64.jpg")) + self.assertTrue(eng.thumbnail_path(items[0], 120, 120, crop=True).endswith("_96.jpg")) + landscape = next(i for i in items if not i["portrait"]) + portrait = next(i for i in items if i["portrait"]) + self.assertTrue(eng.thumbnail_path(landscape, 320, 400).endswith("_320x240.jpg")) + self.assertTrue(eng.thumbnail_path(landscape, 200, 150).endswith("_160x120.jpg")) + self.assertTrue(eng.thumbnail_path(portrait, 240, 320).endswith("_240x320.jpg")) + self.assertTrue(eng.thumbnail_path(portrait, 100, 100).endswith("_120x160.jpg")) + for path in (tile,): + self.assertTrue(os.path.isfile(path)) + eng.delete_session() + self.assertIsNone(eng.session) + + +class AuthToolTests(unittest.TestCase): + def test_pkce(self): + verifier, challenge = gphotos_auth.pkce_pair() + self.assertGreaterEqual(len(verifier), 43) + expect = gphotos_auth.base64url(hashlib.sha256(verifier.encode()).digest()) + self.assertEqual(challenge, expect) + self.assertNotIn("=", challenge) + + def test_client_secrets_and_tokens_file(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "client.json") + with open(path, "w") as f: + json.dump({"installed": {"client_id": "cid.apps", "client_secret": "s"}}, f) + self.assertEqual(gphotos_auth.load_client_secrets(path), ("cid.apps", "s")) + with open(path, "w") as f: + json.dump({"client_id": "flat"}, f) + self.assertEqual(gphotos_auth.load_client_secrets(path), ("flat", "")) + with open(path, "w") as f: + json.dump({"web": {}}, f) + with self.assertRaises(gphotos_auth.AuthError): + gphotos_auth.load_client_secrets(path) + out = gphotos_auth.write_tokens(os.path.join(d, "sub", "t.json"), TOKENS) + self.assertEqual(gphotos_auth.read_tokens(out)["refresh_token"], "rt") + if os.name != "nt": + self.assertEqual(os.stat(out).st_mode & 0o777, 0o600) + # the engine reads what the tool writes + self.assertTrue(ge.credentials_ok(ge.load_tokens(out))) + + def test_auth_url(self): + url = gphotos_auth.build_auth_url("cid", "http://127.0.0.1:4321/", "chal", "st") + self.assertTrue(url.startswith(gphotos_auth.AUTH_URL + "?")) + query = dict(urllib.parse.parse_qsl(url.split("?", 1)[1])) + self.assertEqual(query["redirect_uri"], "http://127.0.0.1:4321/") + self.assertEqual(query["scope"], gphotos_auth.SCOPE) + self.assertEqual(query["code_challenge_method"], "S256") + self.assertEqual(query["access_type"], "offline") + self.assertEqual(query["prompt"], "consent") + self.assertEqual(query["state"], "st") + + def test_loopback_callback(self): + server = gphotos_auth.start_callback_server(0) + port = server.server_address[1] + result = {} + + def serve(): + result.update(gphotos_auth.wait_for_callback(server, timeout=10)) + + t = threading.Thread(target=serve) + t.start() + with urllib.request.urlopen( + "http://127.0.0.1:%d/?code=abc&state=xyz" % port, timeout=5 + ) as r: + self.assertIn(b"Signed in", r.read()) + t.join(5) + server.server_close() + self.assertEqual(result, {"code": "abc", "state": "xyz"}) + + def test_login_exchanges_code(self): + calls = [] + + def fake_post(url, fields, timeout=30.0): + calls.append((url, fields)) + return 200, {"access_token": "a", "refresh_token": "r", "scope": gphotos_auth.SCOPE} + + def fake_wait(server, timeout): + # base64url is patched below, so the tool's state is "fixedstate" + return {"code": "the-code", "state": "fixedstate"} + + with mock.patch.object(gphotos_auth, "_post_form", fake_post), mock.patch.object( + gphotos_auth, "wait_for_callback", fake_wait + ), mock.patch.object(gphotos_auth, "base64url", lambda b: "fixedstate"): + data = gphotos_auth.login("cid", "sec", open_browser=False, out=io.StringIO()) + self.assertEqual(data["refresh_token"], "r") + url, fields = calls[-1] + self.assertEqual(url, gphotos_auth.TOKEN_URL) + self.assertEqual(fields["grant_type"], "authorization_code") + self.assertEqual(fields["code"], "the-code") + self.assertEqual(fields["code_verifier"], "fixedstate") + self.assertTrue(fields["redirect_uri"].startswith("http://127.0.0.1:")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/example_test_manifest.toml b/tools/example_test_manifest.toml index 29c993df..2adaab1c 100644 --- a/tools/example_test_manifest.toml +++ b/tools/example_test_manifest.toml @@ -76,6 +76,22 @@ quit = "inject" deps = ["lvgl"] timeout_s = 45 +[examples.google_photos] +import = "google_photos.google_photos" +script = "examples/google_photos/google_photos.py" +kind = "lvgl" +quit = "inject" +deps = ["lvgl"] +timeout_s = 45 + +[examples.gphotos_lvgl] +import = "google_photos.gphotos_lvgl" +script = "examples/google_photos/gphotos_lvgl.py" +kind = "lvgl" +quit = "inject" +deps = ["lvgl"] +timeout_s = 45 + [examples.paint] import = "paint" script = "examples/paint.py" diff --git a/tools/gphotos_auth.py b/tools/gphotos_auth.py new file mode 100755 index 00000000..5c18af4d --- /dev/null +++ b/tools/gphotos_auth.py @@ -0,0 +1,397 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""One-time Google sign-in for ``lib/examples/google_photos`` (CPython, stdlib only). + +A microcontroller cannot complete Google's login by itself: the TV +"enter this code on your phone" flow refuses the Photos Picker scope, and a +LAN address cannot be registered as an OAuth redirect. So the browser login +runs here, once, and the resulting *refresh token* is copied to the device. +From then on the device only trades that refresh token for short-lived +access tokens. + +Usage:: + + python tools/gphotos_auth.py --client-secrets ~/Downloads/client_secret_XXX.json + python tools/gphotos_auth.py --client-id ID --client-secret SECRET + python tools/gphotos_auth.py --verify # re-check an existing tokens file + +Then copy the file to the board:: + + mpremote cp ~/.gphotos_tokens.json :/gphotos_tokens.json + +Google Cloud setup (details in ``lib/examples/google_photos/README.md``): + +1. APIs & Services -> Library -> enable **Google Photos Picker API**. +2. OAuth consent screen: user type External. Either publish the app to + *Production* (unverified is fine for personal use) or keep it in *Testing* + and add your account as a test user -- but Testing refresh tokens expire + after 7 days, so you would repeat this login weekly. +3. Credentials -> Create credentials -> OAuth client ID -> **Desktop app**; + download the JSON. + +The login uses PKCE and a loopback redirect (``http://127.0.0.1:/``), +the flow Google recommends for installed apps. ``--no-browser`` prints the +URL instead of opening a browser (copy it to any browser on this machine). +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import http.server +import json +import os +import sys +import threading +import urllib.error +import urllib.parse +import urllib.request +import webbrowser + +AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth" +TOKEN_URL = "https://oauth2.googleapis.com/token" +PICKER_SESSIONS_URL = "https://photospicker.googleapis.com/v1/sessions" +SCOPE = "https://www.googleapis.com/auth/photospicker.mediaitems.readonly" +DEFAULT_OUTPUT = "~/.gphotos_tokens.json" +LOGIN_TIMEOUT_S = 300 + +_DONE_HTML = ( + "" + "

{title}

{body}

" +) + + +class AuthError(Exception): + """Login could not be completed.""" + + +def base64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def pkce_pair() -> tuple[str, str]: + """PKCE ``(code_verifier, code_challenge)`` using S256.""" + verifier = base64url(os.urandom(48)) + challenge = base64url(hashlib.sha256(verifier.encode("ascii")).digest()) + return verifier, challenge + + +def load_client_secrets(path: str) -> tuple[str, str]: + """``client_id, client_secret`` from a Google Cloud client JSON download.""" + with open(os.path.expanduser(path), "r", encoding="utf-8") as f: + data = json.load(f) + for key in ("installed", "web"): + if isinstance(data.get(key), dict): + data = data[key] + break + client_id = data.get("client_id") or "" + client_secret = data.get("client_secret") or "" + if not client_id: + raise AuthError(f"{path}: no client_id found") + return client_id, client_secret + + +def build_auth_url(client_id: str, redirect_uri: str, challenge: str, state: str) -> str: + query = { + "client_id": client_id, + "redirect_uri": redirect_uri, + "response_type": "code", + "scope": SCOPE, + "code_challenge": challenge, + "code_challenge_method": "S256", + "access_type": "offline", + "prompt": "consent", + "state": state, + } + return AUTH_URL + "?" + urllib.parse.urlencode(query) + + +def _post_form(url: str, fields: dict, timeout: float = 30.0) -> tuple[int, dict]: + body = urllib.parse.urlencode(fields).encode("ascii") + req = urllib.request.Request( + url, + data=body, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return resp.status, _json(resp.read()) + except urllib.error.HTTPError as err: + return err.code, _json(err.read()) + except urllib.error.URLError as err: + raise AuthError(f"network error: {err.reason}") from err + + +def _json(raw: bytes) -> dict: + try: + data = json.loads(raw.decode("utf-8")) + except (ValueError, UnicodeDecodeError): + return {} + return data if isinstance(data, dict) else {} + + +def _describe(status: int, data: dict) -> str: + err = data.get("error") + if isinstance(err, dict): + return f"HTTP {status} {err.get('status', '')} {err.get('message', '')}".strip() + desc = data.get("error_description") or "" + return f"HTTP {status} {err or ''} {desc}".strip() + + +def exchange_code( + client_id: str, client_secret: str, code: str, redirect_uri: str, verifier: str +) -> dict: + status, data = _post_form( + TOKEN_URL, + { + "code": code, + "client_id": client_id, + "client_secret": client_secret, + "redirect_uri": redirect_uri, + "grant_type": "authorization_code", + "code_verifier": verifier, + }, + ) + if status != 200 or not data.get("access_token"): + raise AuthError("token exchange failed: " + _describe(status, data)) + return data + + +def refresh_access_token(client_id: str, client_secret: str, refresh_token: str) -> dict: + status, data = _post_form( + TOKEN_URL, + { + "client_id": client_id, + "client_secret": client_secret, + "refresh_token": refresh_token, + "grant_type": "refresh_token", + }, + ) + if status != 200 or not data.get("access_token"): + raise AuthError("token refresh failed: " + _describe(status, data)) + return data + + +def picker_probe(access_token: str, timeout: float = 30.0) -> str: + """Create and delete one picker session; returns a human-readable verdict.""" + headers = { + "Authorization": "Bearer " + access_token, + "Content-Type": "application/json", + } + req = urllib.request.Request(PICKER_SESSIONS_URL, data=b"{}", headers=headers, method="POST") + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + session = _json(resp.read()) + except urllib.error.HTTPError as err: + raise AuthError( + "Picker API check failed: " + _describe(err.code, _json(err.read())) + ) from err + except urllib.error.URLError as err: + raise AuthError(f"network error: {err.reason}") from err + sid = session.get("id") or "" + if sid: + req = urllib.request.Request( + PICKER_SESSIONS_URL + "/" + urllib.parse.quote(sid, safe=""), + headers={"Authorization": "Bearer " + access_token}, + method="DELETE", + ) + try: + urllib.request.urlopen(req, timeout=timeout).close() + except (urllib.error.HTTPError, urllib.error.URLError): + pass + return "Picker API OK (session created and deleted)" + + +def write_tokens(path: str, data: dict) -> str: + """Write the tokens JSON with owner-only permissions; returns the path.""" + path = os.path.expanduser(path) + directory = os.path.dirname(path) + if directory: + os.makedirs(directory, exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2) + f.write("\n") + try: + os.chmod(path, 0o600) + except OSError: + pass + return path + + +def read_tokens(path: str) -> dict: + with open(os.path.expanduser(path), "r", encoding="utf-8") as f: + data = json.load(f) + if not isinstance(data, dict) or not data.get("refresh_token"): + raise AuthError(f"{path}: not a tokens file (no refresh_token)") + return data + + +class _CallbackHandler(http.server.BaseHTTPRequestHandler): + """Captures ``?code=…&state=…`` from Google's loopback redirect.""" + + result: dict = {} + + def do_GET(self): # noqa: N802 - http.server API + query = urllib.parse.urlparse(self.path).query + params = {k: v[0] for k, v in urllib.parse.parse_qs(query).items()} + if "code" in params or "error" in params: + type(self).result = params + if "error" in params: + title, body = "Sign-in failed", params.get("error", "") + elif "code" in params: + title, body = "Signed in", "You can close this tab and return to the terminal." + else: + title, body = "Waiting", "This page only answers Google's redirect." + payload = _DONE_HTML.format(title=title, body=body).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def log_message(self, *args): # silence the default stderr access log + pass + + +def start_callback_server(port: int = 0) -> http.server.HTTPServer: + _CallbackHandler.result = {} + return http.server.HTTPServer(("127.0.0.1", port), _CallbackHandler) + + +def wait_for_callback(server: http.server.HTTPServer, timeout: float) -> dict: + """Serve until a code (or error) arrives, or ``timeout`` seconds pass.""" + server.timeout = 1.0 + deadline = _now() + timeout + while not _CallbackHandler.result and _now() < deadline: + server.handle_request() + return dict(_CallbackHandler.result) + + +def _now() -> float: + import time + + return time.monotonic() + + +def login( + client_id: str, + client_secret: str, + port: int = 0, + open_browser: bool = True, + timeout: float = LOGIN_TIMEOUT_S, + out=sys.stdout, +) -> dict: + """Run the browser login; returns Google's token response.""" + verifier, challenge = pkce_pair() + state = base64url(os.urandom(16)) + server = start_callback_server(port) + try: + redirect_uri = "http://127.0.0.1:%d/" % server.server_address[1] + url = build_auth_url(client_id, redirect_uri, challenge, state) + print("Sign in with the Google account whose photos you want to show.", file=out) + if open_browser: + print("Opening your browser (or copy this URL into one):", file=out) + else: + print("Open this URL in a browser on this machine:", file=out) + print("\n" + url + "\n", file=out) + if open_browser: + thread = threading.Thread(target=webbrowser.open, args=(url,), daemon=True) + thread.start() + params = wait_for_callback(server, timeout) + finally: + server.server_close() + if not params: + raise AuthError("timed out waiting for the browser redirect") + if params.get("error"): + raise AuthError("Google refused the sign-in: " + params["error"]) + if params.get("state") != state: + raise AuthError("state mismatch on the redirect (stale tab?) - try again") + return exchange_code(client_id, client_secret, params["code"], redirect_uri, verifier) + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument( + "--client-secrets", help="client_secret_*.json downloaded from Google Cloud" + ) + parser.add_argument("--client-id") + parser.add_argument("--client-secret", default="") + parser.add_argument( + "--output", default=DEFAULT_OUTPUT, help=f"tokens file (default {DEFAULT_OUTPUT})" + ) + parser.add_argument( + "--port", type=int, default=0, help="loopback port (default: any free port)" + ) + parser.add_argument( + "--no-browser", action="store_true", help="print the URL instead of opening a browser" + ) + parser.add_argument( + "--timeout", type=float, default=LOGIN_TIMEOUT_S, help="seconds to wait for the redirect" + ) + parser.add_argument( + "--verify", + action="store_true", + help="after login (or alone) refresh the token and check the Picker API", + ) + args = parser.parse_args(argv) + + try: + if args.client_secrets or args.client_id: + if args.client_secrets: + client_id, client_secret = load_client_secrets(args.client_secrets) + if args.client_secret: + client_secret = args.client_secret + else: + client_id, client_secret = args.client_id, args.client_secret + response = login( + client_id, + client_secret, + port=args.port, + open_browser=not args.no_browser, + timeout=args.timeout, + ) + refresh_token = response.get("refresh_token") + if not refresh_token: + raise AuthError( + "Google returned no refresh token. Remove the app at " + "https://myaccount.google.com/permissions and run this again." + ) + tokens = { + "client_id": client_id, + "client_secret": client_secret, + "refresh_token": refresh_token, + "scope": response.get("scope", SCOPE), + "token_type": response.get("token_type", "Bearer"), + } + path = write_tokens(args.output, tokens) + print(f"Wrote {path}") + print("Copy it to the board, e.g.:") + print(f" mpremote cp {path} :/gphotos_tokens.json") + print("Desktop runs read it from that same path (override: GPHOTOS_TOKENS=...).") + elif args.verify: + tokens = read_tokens(args.output) + else: + parser.error("give --client-secrets FILE or --client-id ID (or --verify)") + return 2 + + if args.verify: + data = refresh_access_token( + tokens["client_id"], tokens.get("client_secret", ""), tokens["refresh_token"] + ) + print("Token refresh OK (expires in %s s)" % data.get("expires_in", "?")) + print(picker_probe(data["access_token"])) + except AuthError as err: + print(f"error: {err}", file=sys.stderr) + return 1 + except KeyboardInterrupt: + print("\ninterrupted", file=sys.stderr) + return 130 + return 0 + + +if __name__ == "__main__": + sys.exit(main())