Embedded WebView2 engine for web links (Windows)

Web links previously launched a separate Chrome/Edge kiosk process, which
caused the whole class of bugs in the tracker: the browser opening behind the
player, being handed off to an already-running instance and exiting instantly,
fighting for foreground/z-order, and leaking msedge.exe/chrome.exe processes
that were never closed.

WebView2 renders as a CHILD HWND of Kivy's own SDL window instead, so there is
no separate top-level browser to open behind the player, nothing to hand the
URL off to, no z-order contest, and no leaked browser process.

Windows/webview2_browser.py
  - Environment -> controller -> navigate, driven through pythonnet.
  - Async .NET Tasks are polled from Kivy's Clock. Calling GetAwaiter()
    .GetResult() would deadlock: the continuation needs the same thread's
    message pump.
  - The controller is a .NET IntPtr, not a Python int (CreateAsync overloads
    do not match otherwise).
  - NavigationCompleted is tracked so a page that never loads can be told
    apart from one that did. This matters on a closed network: an unreachable
    host paints a Chromium error page, and without this the player would show
    a blank/error screen for the item's whole slot instead of skipping it.
  - is_alive() reports True while starting up. Start-up is async, so a
    controller that does not exist yet is not a dead browser; treating it as
    one made the first web link after a cold start be skipped instantly.

Windows/webview2_runtime.py
  - Detects the Runtime (registry pv value, SDK probe as fallback) and
    installs it silently when missing, unelevated, which produces a per-user
    install and therefore never raises a UAC prompt on the signage display.
  - Success is decided by RE-READING the installed version, not by the
    installer exit code: Edge Update returns a non-zero HRESULT
    (-2147219416) when the Runtime is already current, which is not a failure.
  - On a closed network the online bootstrapper can never succeed, so it fails
    fast with an actionable message instead of hanging for the full timeout.
  - Failed attempts are cooldown-gated so a broken machine does not re-run an
    installer on every start.

Offline hardening
  - Browser arguments disable component updates, field trials, safe-browsing
    list fetches, translate and other internet chatter. On an isolated LAN
    each of those would otherwise have to time out, costing start-up latency.
    Pages on the local server are unaffected.

Engine order (best first): WebView2 -> CEF -> Chrome/Edge subprocess. CEF has
no wheels past Python 3.9 so it is dormant on this build; the subprocess engine
remains only as a last resort.

Also fixes the reason the Windows adapters were never used at all:
SignagePlayer.__init__ assigned self.weblink_adapter_factory = None, which
shadowed the CLASS attribute that run_win.py injects. play_weblink() therefore
fell back to the generic adapter, whose find_browser() uses shutil.which() and
finds nothing on Windows because Chrome/Edge are not on PATH. The instance
attribute is now only set when the class attribute is absent.

Verified: windows/test_webview2_embed.py, test_webview2_navigation.py and
test_webview2_offline.py all pass (a locally served page renders with all
internet traffic disabled), and the packaged exe reports
"weblink_launch engine=webview2-embedded" -> "weblink_launched" on every cycle
with no leaked browser processes.
This commit is contained in:
ske087
2026-09-13 10:14:18 +03:00
parent eb8e66e427
commit 9f5409685d
15 changed files with 2418 additions and 66 deletions
+48 -9
View File
@@ -6,9 +6,10 @@ playlist item type (display a live web page / URL instead of an uploaded media
file).
> **Status: implemented.** The player supports `weblink` items on both
> Raspberry Pi (`chromium` subprocess) and Windows (embedded CEF with a
> Chrome/Edge subprocess fallback). Sections 14 describe the original design
> plan; section 6 documents the shipped architecture and the interaction model.
> Raspberry Pi (`chromium` subprocess) and Windows (embedded **WebView2**, with
> the CEF and Chrome/Edge subprocess engines as fallbacks). Sections 14
> describe the original design plan; section 6 documents the shipped
> architecture and the interaction model.
---
@@ -256,18 +257,56 @@ platform:
| Piece | Responsibility |
|--------------------------|----------------|
| `WeblinkSession` | Owns one weblink item: validate → launch → verify → watch → teardown. Generation-tokened so stale callbacks are ignored, and `atexit`-safe. |
| `WeblinkAdapter` | The only platform-specific part: launch / wait for the window / is it alive / tear it down / pre-warm. |
| `ChromiumSubprocessAdapter` | Default engine (Raspberry Pi `chromium`, Windows `chrome.exe`/`msedge.exe`). |
| `WeblinkAdapter` | The only platform-specific part: launch / wait for the window / is it alive / tear it down / pre-warm. `extra_launch_args()` lets a subclass add browser flags without copying `launch()`. |
| `ChromiumSubprocessAdapter` | Default engine (Raspberry Pi `chromium`; on Windows the Chrome/Edge fallback). |
| `InteractionWatcher` | Decides when the item is finished (see the interaction model below). |
| `WebInputSources` | Reads `/dev/input/event*` (Linux) and does a pointer-position tap (Windows, needed for embedded CEF). |
| `WebInputSources` | Reads `/dev/input/event*` (Linux) and does a pointer-position tap (Windows, needed for embedded engines). |
| `webview2_browser.py` | **Windows, preferred**: embeds WebView2 as a child HWND of the Kivy window. |
| `webview2_runtime.py` | **Windows**: detects the WebView2 Runtime and installs it silently when missing. |
Platform wrappers inject their engines through
`SignagePlayer.weblink_adapter_factory`:
* **Raspberry Pi / Linux** — built-in Chromium subprocess adapter.
* **Windows** (`windows/run_win.py`) — embedded CEF first (`cef_browser.py`,
renders inside the Kivy window: no z-order fights, no subprocess), then the
Chrome/Edge subprocess adapter as fallback.
* **Windows** (`windows/run_win.py`) — engines are tried in this order:
| Order | Engine | Renders | Notes |
|-------|--------|---------|-------|
| 1 | **WebView2** (`webview2_browser.py`) | child window **inside** Kivy | Preferred. No subprocess, so no background/z-order/hand-off/leak problems. |
| 2 | CEF (`cef_browser.py`) | child window inside Kivy | Dormant: `cefpython3` has no wheels past Python 3.9. |
| 3 | Chrome/Edge subprocess | separate window | Last resort only; retains the old drawbacks. |
`WeblinkAdapter.extra_launch_args()` is the hook subclasses use to add flags
without duplicating `launch()` — the Chrome adapter uses it for
`--user-data-dir` + `--kiosk`.
> **Do not give the factory a class-level `None` default combined with an
> unconditional instance assignment.** `SignagePlayer.__init__` originally set
> `self.weblink_adapter_factory = None`, which shadowed the class attribute the
> Windows wrapper installs — so the platform adapters were silently ignored and
> every weblink fell back to the generic adapter and failed. It now only sets
> the instance attribute when the class attribute is absent.
### 6.2 Windows: WebView2 Runtime
WebView2 is two separate things, and they ship differently:
* the **SDK** (`Microsoft.Web.WebView2.Core.dll`, `WebView2Loader.dll`) — the
API surface, bundled in the exe from `windows\webview2_sdk\` (~860 KB);
* the **Runtime** (`msedgewebview2.exe`) — the actual Chromium engine, shipped
by Microsoft and **verified/installed at start-up** by
`windows\webview2_runtime.py`.
If the Runtime is absent the player runs an installer silently
(`/silent /install`) and **unelevated**, which produces a *per-user* install and
therefore never raises a UAC prompt on the signage display. A ~1.7 MB online
bootstrapper is bundled by default; a ~203 MB offline standalone installer can
be bundled instead (see `windows\webview2_runtime\download_runtime_installers.ps1`)
for machines with no internet.
Install success is decided by **re-reading the installed version**, not by the
installer exit code — Edge Update returns a non-zero HRESULT (e.g.
`-2147219416`) when the Runtime is already current, which is not a failure.
### 6.1 Interaction model — web links are not passive media
+353 -51
View File
@@ -9,6 +9,8 @@ import os
import json
import platform
import signal
import subprocess
import sys
import threading
import time
import asyncio
@@ -91,6 +93,7 @@ from kivy.graphics import Color, Line, Ellipse
from kivy.uix.floatlayout import FloatLayout
from kivy.uix.slider import Slider
from playback_trace import trace # always-on playback transition logger
from video_safety import suppress_kivy_video_blocking_unload # bound Kivy's blocking video join
from weblink_session import (
WeblinkSession,
WeblinkSettings,
@@ -627,12 +630,19 @@ class ExitPasswordPopup(Popup):
Clock.schedule_once(lambda dt: self.dismiss(), 1)
class SettingsPopup(Popup):
def __init__(self, player_instance, was_paused=False, **kwargs):
def __init__(self, player_instance, was_paused=False, first_run=False, **kwargs):
super(SettingsPopup, self).__init__(**kwargs)
self.player = player_instance
self.was_paused = was_paused
self.first_run = bool(first_run)
self.keyboard_widget = None
# First-run setup is the ONLY thing on screen and there is no playlist
# yet, so the popup must not be dismissable into a blank player.
if self.first_run:
self.auto_dismiss = False
self.title = 'Player Setup - Not Configured'
# Cancel all scheduled cursor/control hide events
try:
if self.player.controls_timer:
@@ -649,15 +659,22 @@ class SettingsPopup(Popup):
except:
pass
# Populate current values
self.ids.server_input.text = self.player.config.get('server_ip', 'localhost')
self.ids.port_input.text = str(self.player.config.get('port', ''))
self.ids.screen_input.text = self.player.config.get('screen_name', 'kivy-player')
self.ids.quickconnect_input.text = self.player.config.get('quickconnect_key', '1234567')
# Populate current values. Fields are blank on a fresh install, so the
# hint text tells the operator what to enter.
self.ids.server_input.text = self.player.config.get('server_ip', '')
self.ids.port_input.text = str(self.player.config.get('port', '') or '')
self.ids.screen_input.text = self.player.config.get('screen_name', '')
self.ids.quickconnect_input.text = self.player.config.get('quickconnect_key', '')
self.ids.orientation_input.text = self.player.config.get('orientation', 'Landscape')
self.ids.touch_input.text = self.player.config.get('touch', 'True')
self.ids.resolution_input.text = self.player.config.get('max_resolution', 'auto')
self.ids.resolution_input.text = self.player.config.get('max_resolution', '1920x1080')
self.ids.edit_enabled_checkbox.active = self.player.config.get('edit_feature_enabled', True)
if self.first_run:
Logger.info(
"SettingsPopup: First-run setup opened (server_ip/screen_name/"
"quickconnect_key must be filled in)"
)
# Update status info
self.ids.playlist_info.text = f'Playlist: v{self.player.playlist_version}'
@@ -711,6 +728,11 @@ class SettingsPopup(Popup):
"""Handle popup dismissal - resume playback and restart cursor hide timer"""
# Hide and remove keyboard
self.hide_keyboard()
if self.first_run:
# Nothing was playing: there is no playlist to resume, and
# scheduling an advance here would race the first sync kicked off
# by on_first_run_config_saved().
return
# Resume playback and re-arm the media advance timer
self.player.resume_after_popup(self.was_paused)
@@ -919,27 +941,122 @@ class SettingsPopup(Popup):
Clock.schedule_once(lambda dt: popup.dismiss(), 2)
def save_and_close(self):
"""Save configuration and close popup"""
"""Save configuration and close popup.
During first-run setup the play/pause guard does not apply (there is no
playlist yet) and the player is told to start once the details are in.
"""
# Update config
self.player.config['server_ip'] = self.ids.server_input.text
self.player.config['server_ip'] = self.ids.server_input.text.strip()
self.player.config['port'] = self.ids.port_input.text.strip()
self.player.config['screen_name'] = self.ids.screen_input.text
self.player.config['quickconnect_key'] = self.ids.quickconnect_input.text
self.player.config['screen_name'] = self.ids.screen_input.text.strip()
self.player.config['quickconnect_key'] = self.ids.quickconnect_input.text.strip()
self.player.config['orientation'] = self.ids.orientation_input.text
self.player.config['touch'] = self.ids.touch_input.text
self.player.config['max_resolution'] = self.ids.resolution_input.text
self.player.config['edit_feature_enabled'] = self.ids.edit_enabled_checkbox.active
# First-run validation: refuse to "configure" the player with blanks,
# otherwise it would be marked configured and then fail on every sync.
if self.first_run:
missing = [
label for label, key in (
('Server IP', 'server_ip'),
('Player Name', 'screen_name'),
('Quick Connect Key', 'quickconnect_key'),
)
if not str(self.player.config.get(key, '') or '').strip()
]
if missing:
self._show_temp_message(
'Required: ' + ', '.join(missing), (1, 0.7, 0, 1)
)
return
# A brand-new install has no reason to keep HTTPS verification on.
self.player.config.setdefault('use_https', False)
self.player.config.setdefault('verify_ssl', False)
# Save to file
self.player.save_config()
# Notify user that resolution change requires restart
if self.ids.resolution_input.text != self.player.config.get('max_resolution', 'auto'):
Logger.info("SettingsPopup: Resolution changed - restart required")
was_first_run = self.first_run
# Close popup
self.dismiss()
if was_first_run:
self.player.on_first_run_config_saved()
# ── First-run configuration ──────────────────────────────────────────
# The player is shipped WITHOUT any server credentials: `app_config.json` is
# not bundled into the exe (see windows/build.spec). On first start there is
# nothing to connect to, so the player shows a notice after the splash video
# and then opens Settings automatically.
#
# These are the values the player cannot work without, plus the placeholder
# values a fresh install used to be given. A config holding only placeholders
# counts as UNCONFIGURED, so a machine that has never been set up runs the
# first-run flow instead of silently trying to reach "localhost".
CONFIG_REQUIRED_KEYS = ('server_ip', 'screen_name', 'quickconnect_key')
CONFIG_PLACEHOLDER_VALUES = {
'server_ip': {'', 'localhost', '127.0.0.1'},
'screen_name': {'', 'kivy-player'},
'quickconnect_key': {'', '1234567'},
}
#: In-memory starting point when no usable config file exists. Deliberately
#: has EMPTY credentials so `config_is_configured()` reports False.
DEFAULT_CONFIG = {
'server_ip': '',
'port': '8080',
'screen_name': '',
'quickconnect_key': '',
'orientation': 'Landscape',
'touch': 'True',
'max_resolution': '1920x1080',
'edit_feature_enabled': True,
'use_https': False,
'verify_ssl': False,
'production_mode': False,
'weblink': {
'engine': 'auto',
'interaction_postpone': 10,
'interaction_debounce': 0.5,
'interaction_grace': 5.0,
'max_dwell_factor': 6.0,
'min_max_dwell': 300,
'launch_timeout': 15,
'prewarm': True,
},
}
#: Seconds the "not configured" notice stays up before Settings opens.
SETUP_NOTICE_SECONDS = 5
def config_is_configured(config):
"""True when ``config`` has enough real values to talk to a server.
Missing file, empty file, unparseable JSON, missing keys and leftover
placeholder values all mean "not configured" — that is what triggers the
first-run setup screen instead of a playlist attempt.
"""
if not isinstance(config, dict) or not config:
return False
for key in CONFIG_REQUIRED_KEYS:
value = str(config.get(key, '') or '').strip()
if not value:
return False
if value.lower() in CONFIG_PLACEHOLDER_VALUES.get(key, set()):
return False
return True
class SignagePlayer(Widget):
from kivy.properties import StringProperty
@@ -968,8 +1085,16 @@ class SignagePlayer(Widget):
# watching and teardown for one weblink item at a time. Platform wrappers
# (e.g. windows/run_win.py) inject their adapters via
# `weblink_adapter_factory` before playback starts.
#
# CRITICAL: the factory is injected as a *class* attribute, so do NOT
# assign None unconditionally here. An instance attribute would shadow
# it, play_weblink() would silently fall back to the generic adapter,
# and on Windows that adapter's find_browser() (shutil.which) finds no
# browser because Chrome/Edge are not on PATH — so every weblink failed
# with "launch() returned False" and the item was skipped.
self._weblink_session = None
self.weblink_adapter_factory = None
if not callable(getattr(type(self), 'weblink_adapter_factory', None)):
self.weblink_adapter_factory = None
self.is_playing = False
self.is_paused = False
self.auto_resume_event = None # Track scheduled auto-resume
@@ -1238,31 +1363,141 @@ class SignagePlayer(Widget):
# Start media playback
Clock.schedule_interval(self.check_playlist_and_play, 30) # Check every 30 seconds
def load_config(self):
"""Load configuration from file"""
Logger.debug("SignagePlayer: load_config() starting...")
def requires_setup(self):
"""True when the player has no usable server configuration."""
return not getattr(self, '_configured', config_is_configured(self.config))
def on_intro_finished(self):
"""Splash video has ended — continue with either setup or playback.
Single decision point for "intro done", so the first-run branch and the
normal branch cannot drift apart.
"""
self.intro_played = True
if self.requires_setup():
Logger.warning(
"SignagePlayer: No server configuration - showing setup notice"
)
self.show_setup_required_notice()
return
# Normal start: load whatever playlist is cached and begin playing.
self.check_playlist_and_play(None)
def show_setup_required_notice(self):
"""Show the "not configured" notice, then open Settings automatically.
The player is shipped without credentials, so this is the expected
first-run experience on a brand-new .exe: tell the operator what is
missing, wait SETUP_NOTICE_SECONDS, then open the settings screen so
they can enter the server details.
"""
trace('setup_required_shown')
try:
if os.path.exists(self.config_file):
self.ids.status_label.text = (
'Player is not configured\n\n'
'No server settings found. Opening setup...'
)
self.ids.status_label.opacity = 1
except Exception:
pass
self._setup_notice_event = Clock.schedule_once(
lambda dt: self._open_first_run_settings(), SETUP_NOTICE_SECONDS
)
def _open_first_run_settings(self):
"""Open Settings for first-run configuration."""
trace('setup_opening_settings')
try:
self.ids.status_label.opacity = 0
except Exception:
pass
popup = SettingsPopup(player_instance=self, first_run=True)
popup.open()
def on_first_run_config_saved(self):
"""Called when Settings was used to configure the player.
Applies the new values immediately and starts playback, so the operator
does not have to restart the .exe after entering the server details.
"""
self._configured = config_is_configured(self.config)
if not self._configured:
Logger.warning(
"SignagePlayer: Settings saved but the player is still not "
"configured - setup will be offered again"
)
self.show_setup_required_notice()
return
trace('setup_completed')
Logger.info("SignagePlayer: First-run configuration saved - starting playback")
try:
self.ids.status_label.text = 'Configuration saved - connecting...'
self.ids.status_label.opacity = 1
except Exception:
pass
# Re-arm the pieces that depend on the server details.
self.start_network_monitoring()
# Pull the first playlist in the background, then play it.
def _fetch_and_play(dt):
try:
updated = update_playlist_if_needed(
self.config, self.playlists_dir, self.media_dir
)
if updated:
Logger.info("SignagePlayer: Playlist fetched after setup")
except Exception as exc:
Logger.error(f"SignagePlayer: Playlist fetch after setup failed: {exc}")
self.load_playlist()
self.is_playing = False
self.is_paused = False
self.start_playback()
threading.Thread(target=_fetch_and_play, args=(None,), daemon=True).start()
def load_config(self):
"""Load configuration from file.
A missing/empty/unreadable file is NOT an error and is NOT written
back: the player starts with empty credentials and the first-run setup
flow asks the operator to fill them in. Writing a placeholder file here
would defeat that (and used to plant `localhost` as a fake server).
"""
Logger.debug("SignagePlayer: load_config() starting...")
self.config = dict(DEFAULT_CONFIG)
self._config_file_existed = os.path.exists(self.config_file)
try:
if self._config_file_existed:
with open(self.config_file, 'r') as f:
self.config = json.load(f)
Logger.info(f"SignagePlayer: Configuration loaded from {self.config_file}")
loaded = json.load(f)
if isinstance(loaded, dict) and loaded:
# Keep defaults for anything the file omits.
self.config.update(loaded)
Logger.info(
f"SignagePlayer: Configuration loaded from {self.config_file}"
)
else:
Logger.warning(
"SignagePlayer: Configuration file is empty or not a JSON "
"object - treating the player as unconfigured"
)
else:
# Create default configuration with HTTPS support
self.config = {
"server_ip": "localhost",
"port": "443",
"screen_name": "kivy-player",
"quickconnect_key": "1234567",
"max_resolution": "auto",
"use_https": True,
"verify_ssl": True,
"production_mode": False
}
self.save_config()
Logger.info("SignagePlayer: Created default configuration with HTTPS enabled")
Logger.warning(
f"SignagePlayer: No configuration file at {self.config_file} "
"- first-run setup will be shown"
)
except Exception as e:
Logger.error(f"SignagePlayer: Error loading config: {e}")
self.show_error(f"Failed to load configuration: {e}")
self.config = dict(DEFAULT_CONFIG)
self._configured = config_is_configured(self.config)
Logger.info(
"SignagePlayer: Configuration status: "
+ ("configured" if self._configured else "NOT configured (setup required)")
)
def save_config(self):
"""Save configuration to file"""
@@ -1380,14 +1615,18 @@ class SignagePlayer(Widget):
if not os.path.exists(intro_path):
Logger.warning(f"SignagePlayer: Intro video not found at {intro_path}")
# Skip intro and load playlist
self.intro_played = True
Clock.schedule_once(self.check_playlist_and_play, 0.1)
# No splash to show: go straight to setup or playback.
self.on_intro_finished()
return
try:
Logger.info("SignagePlayer: Playing intro video...")
self.ids.status_label.opacity = 0 # Hide status label
# Same blocking-join hazard as playlist videos: Kivy's on_state
# teardown joins the ffpyplayer decode thread on the calling
# thread. Install the bound join before the widget exists.
suppress_kivy_video_blocking_unload()
# Create video widget for intro
intro_video = Video(
@@ -1407,23 +1646,29 @@ class SignagePlayer(Widget):
# Mark intro as played before removing video
self.intro_played = True
# Stop and unload the video properly
try:
instance.state = 'stop'
instance.unload()
except Exception as e:
Logger.debug(f"SignagePlayer: Could not unload intro video: {e}")
# Remove intro video
# Do NOT call instance.state = 'stop' / instance.unload()
# here: this runs during the Kivy MainThread dispatch of
# state, and Kivy's Video.unload() joins the ffpyplayer
# decode thread on the CALLING thread — a blocking call
# that froze the UI for 0.4s..100s+ and, when the thread
# never exited, hung the whole player. Detach the widget
# and let the normal video-teardown worker stop it.
try:
if intro_video in self.ids.content_area.children:
self.ids.content_area.remove_widget(intro_video)
except Exception as e:
Logger.warning(f"SignagePlayer: Error removing intro video widget: {e}")
threading.Thread(
target=self._teardown_video_async,
args=(intro_video,),
daemon=True,
name='intro-video-teardown',
).start()
# Start normal playlist immediately to reduce white screen
Logger.debug("SignagePlayer: Triggering playlist check after intro")
self.check_playlist_and_play(None)
self.on_intro_finished()
except Exception as e:
Logger.error(f"SignagePlayer: Error in intro end callback: {e}")
import traceback
@@ -1438,9 +1683,8 @@ class SignagePlayer(Widget):
Logger.error(f"SignagePlayer: Error playing intro video: {e}")
import traceback
Logger.error(f"SignagePlayer: Traceback: {traceback.format_exc()}")
# Skip intro and load playlist
self.intro_played = True
Clock.schedule_once(self.check_playlist_and_play, 0.1)
# Skip intro and continue with setup or playback
self.on_intro_finished()
def check_playlist_and_play(self, dt):
"""Check for playlist updates and ensure playback is running"""
@@ -1448,6 +1692,11 @@ class SignagePlayer(Widget):
if not self.intro_played:
return
# An unconfigured player has no server to sync from; leave the setup
# flow in charge (otherwise the 30s timer would fight the notice).
if self.requires_setup():
return
if not self.playlist:
self.load_playlist()
@@ -1663,6 +1912,39 @@ class SignagePlayer(Widget):
self.show_error(f"Error playing media: {e}")
self._skip_to_next_media()
def _video_has_audio(self, video_path):
"""True when the file actually contains an audio stream.
Needed because ffpyplayer initialises SDL2_mixer from the FIRST audio
file it opens and reuses those parameters. Handing it a video with NO
audio track (integer division of rate/channels by zero internally)
crashes the process with an access violation in SDL2_mixer.dll
(0xc0000005) — observed when a silent 4K clip entered the playlist
after an AAC stereo clip had already been played.
Uses ffprobe (bundled next to the app) and fails SAFE: if we cannot
determine the streams, we report True and let the normal path proceed.
"""
try:
ffprobe = os.path.join(os.path.dirname(sys.executable), '_internal', 'ffprobe.exe')
if not os.path.exists(ffprobe):
# Development run: fall back to PATH / the ffpyplayer bundle.
import shutil as _shutil
ffprobe = _shutil.which('ffprobe')
if not ffprobe:
return True # cannot tell -> assume it has audio
result = subprocess.run(
[ffprobe, '-v', 'error', '-select_streams', 'a',
'-show_entries', 'stream=index', '-of', 'csv=p=0', video_path],
capture_output=True, text=True, timeout=10,
creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0),
)
return bool((result.stdout or '').strip())
except Exception as exc:
Logger.debug(f"SignagePlayer: audio probe failed for {video_path}: {exc}")
return True # fail safe: let Kivy try
def play_video(self, video_path, duration, muted=False):
"""Play a video file using Kivy's Video widget with optimizations.
@@ -1682,9 +1964,29 @@ class SignagePlayer(Widget):
self.consecutive_errors += 1
self._skip_to_next_media()
return
# A video with no audio stream must never be handed to ffpyplayer
# with sound enabled: SDL2_mixer access-violates on a silent track.
# Signage has no audio output anyway, so mute it.
if not muted and not self._video_has_audio(video_path):
Logger.info(
"SignagePlayer: Video has no audio track - forcing mute "
"(avoids an SDL2_mixer crash)"
)
muted = True
Logger.debug(f"SignagePlayer: Loading video {os.path.basename(video_path)} for {duration}s")
# Bound the blocking join() Kivy's VideoFFPy performs on the
# CALLING thread during unload. Kivy's own on_eos handler sets
# state='stop' (which joins the ffpyplayer decode thread) while the
# event is being dispatched — i.e. on the Kivy main thread. When
# that thread is slow to exit, the join parks the UI thread and
# Windows declares the app hung (AppHangB1, observed after ~30-45
# minutes of looping). Must run BEFORE the widget is constructed:
# the decode thread is created during play(). See src/video_safety.py.
suppress_kivy_video_blocking_unload()
# Create Video widget with optimized settings for smooth playback.
# Apply the server's audio:off/muted flag via 'volume' (0.0=mute).
self._video_source = video_path
+27 -1
View File
@@ -223,6 +223,13 @@ class WeblinkAdapter:
#: The session then skips the "browser window appeared" requirement.
embedded = False
#: Set by adapters that render in-window but can still *prove* the page
#: appeared (WebView2). Without this, an embedded engine is trusted
#: blindly, so a page that fails to load (unreachable host on a closed
#: network, DNS failure, 404) would sit on screen for the whole slot
#: instead of being skipped.
can_verify_visibility = False
#: The launched process, when the engine is subprocess based. The session
#: mirrors this onto the player's historic ``_weblink_proc`` attribute.
process = None
@@ -705,6 +712,17 @@ class ChromiumSubprocessAdapter(WeblinkAdapter):
# Nothing generic to do; the platform layer hooks in here (overlay).
pass
def extra_launch_args(self):
"""Extra flags appended to the browser command line.
Subclasses override this instead of duplicating ``launch()``. The
Windows adapter uses it to inject a dedicated ``--user-data-dir``,
which is mandatory there: without it Chrome/Edge hands the URL to an
already-running instance, the process we launched exits immediately
and the weblink never becomes visible.
"""
return ()
def launch(self, url, width, height):
browser = self._browser or self.find_browser()
if not browser:
@@ -735,6 +753,7 @@ class ChromiumSubprocessAdapter(WeblinkAdapter):
'--force-device-scale-factor=1',
]
args += self._extra_flags
args += [str(arg) for arg in self.extra_launch_args()]
self._proc = subprocess.Popen(args)
return True
@@ -958,6 +977,13 @@ class WeblinkSession:
with self._lock:
if self._watcher is not None:
self._watcher.stop()
# An embedded engine normally cannot be verified (CEF), but some can
# (WebView2) — for those the visibility wait must still run,
# otherwise a page that never loads is shown as a blank screen for
# the whole duration instead of being skipped.
verify = adapter.wait_visible if (
not adapter.embedded or adapter.can_verify_visibility
) else None
self._watcher = InteractionWatcher(
duration=duration,
alive_check=adapter.is_alive,
@@ -972,7 +998,7 @@ class WeblinkSession:
interaction_debounce=self.settings.interaction_debounce,
interaction_grace=self.settings.interaction_grace,
embedded=adapter.embedded,
wait_visible=None if adapter.embedded else adapter.wait_visible,
wait_visible=verify,
visible_timeout=self.settings.launch_timeout,
launched_at=launched_at,
)
+67 -5
View File
@@ -103,6 +103,8 @@ hidden_imports = [
'tempfile',
# Windows-specific
'cef_browser',
'webview2_browser',
'webview2_runtime',
'win32gui',
'win32con',
# Unified web-link controller (launch / verified visibility / interaction
@@ -139,11 +141,60 @@ for item in RESOURCES_DIR.iterdir():
target_dir = 'config/resources'
resources_data.append((str(item), target_dir))
# Config directory (app_config.json)
# Config directory.
#
# app_config.json is deliberately NOT bundled. Including it shipped the
# developer's own server_ip / screen_name inside the exe, so a fresh install
# silently connected to the wrong server (or to a placeholder) instead of
# asking the operator. The player now starts unconfigured, shows a notice after
# the splash video and opens Settings to collect the real values, which are
# then saved next to the .exe.
config_data = []
config_file = CONFIG_DIR / 'app_config.json'
if config_file.exists():
config_data.append((str(config_file), 'config'))
print("[spec] app_config.json is NOT bundled (first-run setup collects it)")
# --- Bundled web engines ---------------------------------------------
# Embedded WebView2 (Edge) SDK: the managed assembly plus the native loader
# DLL. The *runtime* itself is a Microsoft-shipped evergreen component and is
# deliberately NOT bundled (that is the point of using WebView2 — no ~150 MB
# Chromium payload inside our exe).
webview2_data = []
webview2_sdk = BUILD_DIR / 'webview2_sdk'
if webview2_sdk.is_dir():
for item in webview2_sdk.iterdir():
if item.is_file():
webview2_data.append((str(item), 'webview2_sdk'))
print(f"[spec] Bundling {len(webview2_data)} WebView2 SDK file(s) from {webview2_sdk}")
# WebView2 Runtime installer, so a machine that ships WITHOUT the Runtime can
# install it on first start (see windows/webview2_runtime.py).
#
# Only the small bootstrapper (~1.7 MB) is bundled by default; it downloads the
# Runtime from Microsoft. Dropping the ~203 MB offline standalone installer
# into windows/webview2_runtime/ bundles it too, which is what you want for
# machines with no internet — but it triples the exe size, so it is opt-in.
webview2_runtime = BUILD_DIR / 'webview2_runtime'
_bootstrap = webview2_runtime / 'MicrosoftEdgeWebview2Setup.exe'
_standalone = webview2_runtime / 'MicrosoftEdgeWebView2RuntimeInstallerX64.exe'
if _bootstrap.is_file():
webview2_data.append((str(_bootstrap), 'webview2_runtime'))
print(f"[spec] Bundling WebView2 Runtime bootstrapper ({_bootstrap.stat().st_size / 1024 / 1024:.1f} MB)")
if _standalone.is_file():
webview2_data.append((str(_standalone), 'webview2_runtime'))
print(f"[spec] Bundling WebView2 offline standalone installer "
f"({_standalone.stat().st_size / 1024 / 1024:.0f} MB) exe will be much larger")
if not _bootstrap.is_file() and not _standalone.is_file():
print("=" * 70)
print("WARNING: no WebView2 Runtime installer in windows/webview2_runtime/.")
print("Machines without the Runtime cannot show web links (they fall back")
print("to the Chrome/Edge subprocess engine).")
print("=" * 70)
else:
print("=" * 70)
print("WARNING: windows/webview2_sdk/ not found.")
print("Web links will fall back to the Chrome/Edge subprocess engine.")
print("=" * 70)
# Source files - .kv file
kv_file = SRC_DIR / 'signage_player.kv'
@@ -151,8 +202,19 @@ kv_data = []
if kv_file.exists():
kv_data.append((str(kv_file), '.'))
# Bundle the entire src directory as a tree
source_tree = Tree(str(SRC_DIR), prefix='', excludes=['*.pyc', '__pycache__', '*.ini'])
# Bundle the entire src directory as a tree.
#
# EXCLUDE player_auth.json: it holds LIVE credentials (auth_code, player_id,
# server_url). Bundling it means the frozen app starts up in _internal/ and
# loads that snapshot as its auth state — so a freshly built exe boots
# "already authenticated" against whatever server the file happened to name,
# and plays a stale playlist. Auth must be created at runtime in the data dir
# next to the .exe (see run_win.py `_patch_auth_paths`).
source_tree = Tree(
str(SRC_DIR),
prefix='',
excludes=['*.pyc', '__pycache__', '*.ini', 'player_auth.json'],
)
# --- Collect binary DLLs from kivy_deps and ffpyplayer ----------------
import importlib.util
@@ -250,7 +312,7 @@ a = Analysis(
['run_win.py'], # Entry point (relative to this spec)
pathex=[str(BUILD_DIR), str(SRC_DIR)], # Where to find modules
binaries=_all_binaries,
datas=resources_data + config_data + kv_data,
datas=resources_data + config_data + kv_data + webview2_data,
hiddenimports=hidden_imports,
hookspath=[],
hooksconfig={},
+12
View File
@@ -23,6 +23,18 @@ bcrypt>=4.2.0,<5.0.0
# PyInstaller for building the .exe
pyinstaller>=6.0
# --- Embedded web engine (web links) ---
# pythonnet lets Python drive the WebView2 .NET SDK. WebView2 renders INSIDE
# the Kivy window as a child window, which is what removed the old subprocess
# browser bugs (window opening behind the player, instant hand-off exit,
# z-order/focus fights, leaked chrome.exe/msedge.exe processes).
# The WebView2 *runtime* is a free, Microsoft-shipped evergreen component and
# is intentionally NOT bundled; the small SDK DLLs live in windows/webview2_sdk/
# and are added to the exe by build.spec.
# Without pythonnet the player silently falls back to the Chrome/Edge
# subprocess engine, so weblinks still work but with the old drawbacks.
pythonnet>=3.0.3
# --- Windows-specific Libraries ---
# cefpython3: Embedded Chromium browser (replaces subprocess Chrome/Edge)
# Installed separately because it's a large package (69 MB):
+169
View File
@@ -0,0 +1,169 @@
"""Standalone harness for windows/webview2_browser.py — no Kivy, no player.
Creates a plain Win32 window, embeds WebView2 in it via the same
WebView2Browser class the player uses, navigates to a page, checks that the
page actually becomes visible, then resizes and tears down.
Run: windows\\venv\\Scripts\\python.exe windows\\test_webview2_embed.py
Exit code 0 = embedded engine works.
"""
import ctypes
import os
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
URL = os.environ.get('KIWY_TEST_URL', 'https://example.com/')
user32 = ctypes.windll.user32
kernel32 = ctypes.windll.kernel32
WNDPROC = ctypes.WINFUNCTYPE(
ctypes.c_int64,
ctypes.c_void_p, # HWND
ctypes.c_uint, # UINT msg
ctypes.c_void_p, # WPARAM
ctypes.c_void_p, # LPARAM
)
_messages = []
@WNDPROC
def _wnd_proc(hwnd, msg, wparam, lparam):
_messages.append(msg)
if msg == 0x0002: # WM_DESTROY
user32.PostQuitMessage(0)
return 0
user32.DefWindowProcW.restype = ctypes.c_int64
user32.DefWindowProcW.argtypes = [
ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p, ctypes.c_void_p,
]
return user32.DefWindowProcW(hwnd, msg, wparam, lparam)
def _make_window(width=1280, height=720):
"""Register a class and create a visible top-level window."""
hinstance = kernel32.GetModuleHandleW(None)
class_name = 'KiwyWebView2Test'
class WNDCLASSEX(ctypes.Structure):
_fields_ = [
('cbSize', ctypes.c_uint),
('style', ctypes.c_uint),
('lpfnWndProc', WNDPROC),
('cbClsExtra', ctypes.c_int),
('cbWndExtra', ctypes.c_int),
('hInstance', ctypes.c_void_p),
('hIcon', ctypes.c_void_p),
('hCursor', ctypes.c_void_p),
('hbrBackground', ctypes.c_void_p),
('lpszMenuName', ctypes.c_wchar_p),
('lpszClassName', ctypes.c_wchar_p),
('hIconSm', ctypes.c_void_p),
]
wc = WNDCLASSEX()
wc.cbSize = ctypes.sizeof(WNDCLASSEX)
wc.style = 0x0002 | 0x0001 # CS_HREDRAW | CS_VREDRAW
wc.lpfnWndProc = _wnd_proc
wc.hInstance = hinstance
wc.hbrBackground = ctypes.c_void_p(6) # COLOR_WINDOW+1
wc.lpszClassName = class_name
user32.RegisterClassExW(ctypes.byref(wc))
hwnd = user32.CreateWindowExW(
0,
class_name,
'Kiwy WebView2 Embed Test',
0x00CF0000 | 0x10000000, # WS_OVERLAPPEDWINDOW | WS_VISIBLE
100, 100, width, height,
0, 0, hinstance, 0,
)
if not hwnd:
raise RuntimeError(f'CreateWindowExW failed (err={kernel32.GetLastError()})')
user32.UpdateWindow(hwnd)
return hwnd
def _pump(seconds):
"""Pump Win32 messages — WebView2 needs this to deliver its callbacks."""
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
msg = ctypes.wintypes.MSG() if hasattr(ctypes, 'wintypes') else None
import ctypes.wintypes as wt
msg = wt.MSG()
while user32.PeekMessageW(ctypes.byref(msg), None, 0, 0, 1):
user32.TranslateMessage(ctypes.byref(msg))
user32.DispatchMessageW(ctypes.byref(msg))
time.sleep(0.02)
def main():
print('=' * 68)
print(' WebView2 embedded-engine test')
print('=' * 68)
from webview2_browser import WebView2Browser
print(f'SDK dir : {WebView2Browser.__module__}')
available = WebView2Browser.is_available()
print(f'available: {available}')
if not available:
print(f'REASON: {WebView2Browser._import_error}')
return 1
hwnd = _make_window()
print(f'window : hwnd=0x{hwnd:x}')
browser = WebView2Browser(hwnd_provider=lambda: hwnd)
started = time.monotonic()
ok = browser.show(URL)
print(f'show() : {ok}')
if not ok:
print(f'FAILED : {browser.failed_reason}')
return 1
# Drive the Kivy-style Clock poll manually while pumping messages.
visible = False
while time.monotonic() - started < 25:
browser._tick(0) # consume the async task
_pump(0.1)
if browser.failed_reason:
print(f'FAILED : {browser.failed_reason}')
return 1
if browser.is_showing():
visible = True
break
print(f'visible : {visible} after {time.monotonic() - started:.1f}s')
if not visible:
print('FAILED : page never became visible')
return 1
# Resize to the signage resolution and confirm it is applied.
browser.resize(1920, 1080)
_pump(1.0)
print(f'resized : 1920x1080 (bounds={browser._size})')
# Hide, then re-show to prove the controller survives a transition.
browser.hide()
_pump(0.5)
print(f'after hide -> is_showing={browser.is_showing()}')
browser.show(URL)
_pump(2.0)
print(f'after re-show -> is_showing={browser.is_showing()}')
browser.shutdown()
print('shutdown: ok')
print('=' * 68)
print(' RESULT: PASS')
print('=' * 68)
return 0
if __name__ == '__main__':
sys.exit(main())
+186
View File
@@ -0,0 +1,186 @@
"""Does pythonnet fire NavigationCompleted for a real page load?
This validates the mechanism the player relies on to tell "page loaded" apart
from "page failed" (e.g. unreachable host on a closed network). If the event
does not fire, the player cannot detect a failed weblink and would show
Chromium's error page for the full slot.
Checks three things:
1. the delegate can be constructed and subscribed,
2. it fires for a GOOD page -> IsSuccess True,
3. it fires for a BAD page -> IsSuccess False.
Run: windows\\venv\\Scripts\\python.exe windows\\test_webview2_navigation.py
Exit code 0 = PASS.
"""
import ctypes
import http.server
import socketserver
import sys
import threading
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
PORT = 18766
PAGE = '<!doctype html><html><body><h1 id="h">KIWY-NAV-OK</h1></body></html>'
class _Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
body = PAGE.encode()
self.send_response(200)
self.send_header('Content-Type', 'text/html')
self.send_header('Content-Length', str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
def _start_server():
socketserver.TCPServer.allow_reuse_address = True
httpd = socketserver.TCPServer(('127.0.0.1', PORT), _Handler)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
return httpd
user32 = ctypes.windll.user32
kernel32 = ctypes.windll.kernel32
WNDPROC = ctypes.WINFUNCTYPE(
ctypes.c_int64, ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p, ctypes.c_void_p
)
@WNDPROC
def _wnd_proc(hwnd, msg, wparam, lparam):
if msg == 0x0002:
user32.PostQuitMessage(0)
return 0
user32.DefWindowProcW.restype = ctypes.c_int64
user32.DefWindowProcW.argtypes = [
ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p, ctypes.c_void_p,
]
return user32.DefWindowProcW(hwnd, msg, wparam, lparam)
def _make_window(width=1024, height=768):
hinstance = kernel32.GetModuleHandleW(None)
name = 'KiwyNavTest'
class WNDCLASSEX(ctypes.Structure):
_fields_ = [
('cbSize', ctypes.c_uint), ('style', ctypes.c_uint),
('lpfnWndProc', WNDPROC), ('cbClsExtra', ctypes.c_int),
('cbWndExtra', ctypes.c_int), ('hInstance', ctypes.c_void_p),
('hIcon', ctypes.c_void_p), ('hCursor', ctypes.c_void_p),
('hbrBackground', ctypes.c_void_p), ('lpszMenuName', ctypes.c_wchar_p),
('lpszClassName', ctypes.c_wchar_p), ('hIconSm', ctypes.c_void_p),
]
wc = WNDCLASSEX()
wc.cbSize = ctypes.sizeof(WNDCLASSEX)
wc.style = 0x0002 | 0x0001
wc.lpfnWndProc = _wnd_proc
wc.hInstance = hinstance
wc.hbrBackground = ctypes.c_void_p(6)
wc.lpszClassName = name
user32.RegisterClassExW(ctypes.byref(wc))
hwnd = user32.CreateWindowExW(
0, name, 'Kiwy Nav Test', 0x00CF0000 | 0x10000000,
40, 40, width, height, 0, 0, hinstance, 0,
)
if not hwnd:
raise RuntimeError('CreateWindowExW failed')
user32.UpdateWindow(hwnd)
return hwnd
def _pump(seconds):
import ctypes.wintypes as wt
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
msg = wt.MSG()
while user32.PeekMessageW(ctypes.byref(msg), None, 0, 0, 1):
user32.TranslateMessage(ctypes.byref(msg))
user32.DispatchMessageW(ctypes.byref(msg))
time.sleep(0.02)
def _drive(browser, seconds):
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
browser._tick(0)
_pump(0.05)
def _wait_nav(browser, timeout):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
browser._tick(0)
_pump(0.05)
if browser.navigation_succeeded() is not None:
return browser.navigation_succeeded()
return None
def main():
print('=' * 68)
print(' WebView2 NavigationCompleted test')
print('=' * 68)
from webview2_browser import WebView2Browser
if not WebView2Browser.is_available():
print('FAIL: WebView2 unavailable:', WebView2Browser._import_error)
return 1
httpd = _start_server()
good = f'http://127.0.0.1:{PORT}/index.html'
# A port nothing listens on: guarantees a real navigation failure.
bad = 'http://127.0.0.1:1/missing'
hwnd = _make_window()
browser = WebView2Browser(hwnd_provider=lambda: hwnd)
ok = True
print(f'\n[1] good page: {good}')
browser.show(good)
_drive(browser, 1.0)
result = _wait_nav(browser, 20)
print(f' navigation_succeeded = {result}')
print(f' status = {browser.navigation_status()!r}')
if result is not True:
print(' FAIL: good page did not report success')
ok = False
print(f'\n[2] bad page: {bad}')
browser.show(bad)
_drive(browser, 1.0)
result = _wait_nav(browser, 25)
print(f' navigation_succeeded = {result}')
print(f' status = {browser.navigation_status()!r}')
if result is not False:
print(' FAIL: bad page did not report failure')
ok = False
browser.shutdown()
httpd.shutdown()
httpd.server_close()
print('=' * 68)
print(' RESULT:', 'PASS' if ok else 'FAIL')
if ok:
print(' The player can tell a loaded page from a failed one,')
print(' so unreachable weblinks are skipped instead of shown blank.')
print('=' * 68)
return 0 if ok else 1
if __name__ == '__main__':
sys.exit(main())
+221
View File
@@ -0,0 +1,221 @@
"""Closed-network test: does a WebView2 page still load with no internet?
The signage player lives on an isolated LAN, so the important question is not
"does example.com load" but "does a page on a reachable *local* host still
render when there is no internet at all".
This test simulates that properly:
1. Start a tiny HTTP server on 127.0.0.1 serving a known marker page.
2. Create the WebView2 environment **with the same offline browser arguments
the player uses** (webview2_browser._build_environment_options).
3. Navigate to the local page and confirm the page's actual content arrives —
not merely that the controller came up.
It also blocks real internet resolution for the browser by pointing it at the
local server only, so a pass here means offline playback genuinely works.
Run: windows\\venv\\Scripts\\python.exe windows\\test_webview2_offline.py
Exit code 0 = PASS.
"""
import ctypes
import http.server
import os
import socketserver
import sys
import threading
import time
import urllib.request
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
MARKER = 'KIWY-OFFLINE-LAN-OK'
PORT = 18765
PAGE = f"""<!doctype html>
<html><head><meta charset="utf-8"><title>pc</title></head>
<body style="background:#123;color:#fff;font:48px sans-serif">
<div id="m">{MARKER}</div>
</body></html>"""
class _Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
body = PAGE.encode('utf-8')
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.send_header('Content-Length', str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass # keep the test output clean
def _start_server():
socketserver.TCPServer.allow_reuse_address = True
httpd = socketserver.TCPServer(('127.0.0.1', PORT), _Handler)
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
thread.start()
return httpd
# ── Win32 window (same approach as test_webview2_embed.py) ──────────
user32 = ctypes.windll.user32
kernel32 = ctypes.windll.kernel32
WNDPROC = ctypes.WINFUNCTYPE(
ctypes.c_int64, ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p, ctypes.c_void_p
)
@WNDPROC
def _wnd_proc(hwnd, msg, wparam, lparam):
if msg == 0x0002: # WM_DESTROY
user32.PostQuitMessage(0)
return 0
user32.DefWindowProcW.restype = ctypes.c_int64
user32.DefWindowProcW.argtypes = [
ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p, ctypes.c_void_p,
]
return user32.DefWindowProcW(hwnd, msg, wparam, lparam)
def _make_window(width=1280, height=720):
hinstance = kernel32.GetModuleHandleW(None)
class_name = 'KiwyWebView2OfflineTest'
class WNDCLASSEX(ctypes.Structure):
_fields_ = [
('cbSize', ctypes.c_uint), ('style', ctypes.c_uint),
('lpfnWndProc', WNDPROC), ('cbClsExtra', ctypes.c_int),
('cbWndExtra', ctypes.c_int), ('hInstance', ctypes.c_void_p),
('hIcon', ctypes.c_void_p), ('hCursor', ctypes.c_void_p),
('hbrBackground', ctypes.c_void_p), ('lpszMenuName', ctypes.c_wchar_p),
('lpszClassName', ctypes.c_wchar_p), ('hIconSm', ctypes.c_void_p),
]
wc = WNDCLASSEX()
wc.cbSize = ctypes.sizeof(WNDCLASSEX)
wc.style = 0x0002 | 0x0001
wc.lpfnWndProc = _wnd_proc
wc.hInstance = hinstance
wc.hbrBackground = ctypes.c_void_p(6)
wc.lpszClassName = class_name
user32.RegisterClassExW(ctypes.byref(wc))
hwnd = user32.CreateWindowExW(
0, class_name, 'Kiwy Offline LAN Test',
0x00CF0000 | 0x10000000, 60, 60, width, height, 0, 0, hinstance, 0,
)
if not hwnd:
raise RuntimeError('CreateWindowExW failed')
user32.UpdateWindow(hwnd)
return hwnd
def _pump(seconds):
import ctypes.wintypes as wt
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
msg = wt.MSG()
while user32.PeekMessageW(ctypes.byref(msg), None, 0, 0, 1):
user32.TranslateMessage(ctypes.byref(msg))
user32.DispatchMessageW(ctypes.byref(msg))
time.sleep(0.02)
def main():
print('=' * 68)
print(' WebView2 closed-network (LAN-only) test')
print('=' * 68)
from webview2_browser import (
WebView2Browser, _build_environment_options, _offline_browser_arguments,
)
if not WebView2Browser.is_available():
print('FAIL: WebView2 unavailable:', WebView2Browser._import_error)
return 1
print('offline browser args:')
for flag in _offline_browser_arguments().split():
print(f' {flag}')
options = _build_environment_options()
if options is None:
print('\nFAIL: could not build offline environment options')
return 1
print(f'\nAdditionalBrowserArguments set: '
f'{bool(options.AdditionalBrowserArguments)}')
httpd = _start_server()
url = f'http://127.0.0.1:{PORT}/dashboard'
print(f'\nlocal server: {url}')
hwnd = _make_window()
browser = WebView2Browser(hwnd_provider=lambda: hwnd)
started = time.monotonic()
if not browser.show(url):
print('FAIL: show() returned False:', browser.failed_reason)
httpd.shutdown()
return 1
visible = False
deadline = time.monotonic() + 25
while time.monotonic() < deadline:
browser._tick(0)
_pump(0.1)
if browser.failed_reason:
print('FAIL:', browser.failed_reason)
httpd.shutdown()
return 1
if browser.is_showing():
visible = True
break
print(f'page visible : {visible} after {time.monotonic() - started:.1f}s')
# Confirm the page's REAL CONTENT arrived, not just the controller.
#
# NOTE: ExecuteScriptAsync also returns a .NET Task. Calling .Result here
# would deadlock — the continuation needs this thread's message pump, which
# is exactly the mistake this file otherwise exists to catch. Poll it while
# pumping messages instead.
body = ''
if visible:
try:
task = browser._webview.ExecuteScriptAsync(
"document.getElementById('m').innerText"
)
deadline = time.monotonic() + 10
while time.monotonic() < deadline:
_pump(0.05)
if task.IsCompleted:
body = task.Result
break
except Exception as exc:
print(f'note: script eval failed ({exc})')
got_marker = MARKER in (body or '')
print(f'page content : {"marker found" if got_marker else "MARKER MISSING"} '
f'({(body or "")[:60]})')
browser.shutdown()
httpd.shutdown()
httpd.server_close()
ok = visible and got_marker
print('=' * 68)
print(' RESULT:', 'PASS' if ok else 'FAIL')
if ok:
print(' A local page renders with all internet traffic disabled —')
print(' web links work on a closed network.')
print('=' * 68)
return 0 if ok else 1
if __name__ == '__main__':
sys.exit(main())
+106
View File
@@ -0,0 +1,106 @@
"""Standalone test for windows/webview2_runtime.py — no Kivy, no player.
Checks the runtime-detection logic and (optionally) a real silent install.
Run: windows\\venv\\Scripts\\python.exe windows\\test_webview2_runtime.py
windows\\venv\\Scripts\\python.exe windows\\test_webview2_runtime.py --install
Without --install this is read-only: it reports the detected version and which
installer would be used. With --install it forces the installer path to run
(useful on a machine that genuinely lacks the Runtime).
Exit code 0 = checks passed.
"""
import os
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import webview2_runtime as w # noqa: E402
def main():
force_install = '--install' in sys.argv
print('=' * 68)
print(' WebView2 Runtime detection test')
print('=' * 68)
version = w.get_runtime_version()
installed = w.is_runtime_installed()
print(f'registry/SDK version : {version or "(none)"}')
print(f'is_runtime_installed : {installed}')
installer, kind = w.find_installer()
print(f'installer : {installer}')
print(f'installer kind : {kind}')
print(f'describe() : {w.describe()}')
ok = True
# Version parsing must be comparable and tolerant of junk.
cases = {
'152.0.4191.66': (152, 0, 4191, 66),
'1.2': (1, 2, 0, 0),
'': (0, 0, 0, 0),
None: (0, 0, 0, 0),
}
for raw, expected in cases.items():
got = w._parse_version(raw)
flag = 'ok' if got == expected else 'FAIL'
if got != expected:
ok = False
print(f' parse({raw!r:16}) -> {got} [{flag}]')
# An installer must be discoverable: without one, a Runtime-less machine
# has no way to recover.
if installer is None:
print('\nWARNING: no installer found — a machine without the Runtime '
'cannot self-heal.')
print('Run: .\\webview2_runtime\\download_runtime_installers.ps1')
else:
sig_status = 'n/a'
try:
import subprocess
out = subprocess.run(
['powershell', '-NoProfile', '-Command',
f'(Get-AuthenticodeSignature -LiteralPath "{installer}").Status'],
capture_output=True, text=True, timeout=60,
)
sig_status = (out.stdout or '').strip() or 'unknown'
except Exception as exc:
sig_status = f'check failed: {exc}'
print(f'signature : {sig_status}')
if force_install:
print('\n--install given: running the silent installer path...')
result = w.ensure_runtime(timeout=600)
print(f'ensure_runtime() -> {result}')
if not result.get('installed'):
ok = False
else:
# Read-only path: ensure_runtime must be a no-op that reports presence.
result = w.ensure_runtime(timeout=60)
print(f'\nensure_runtime() (read-only) -> {result}')
if installed and not result.get('installed'):
print('FAIL: Runtime present but ensure_runtime() disagreed')
ok = False
if result.get('action') not in ('already-present', 'installer-missing',
'skipped-recent-failure',
'installed-standalone',
'installed-bootstrapper',
'attempted-standalone',
'attempted-bootstrapper'):
print(f'FAIL: unexpected action {result.get("action")!r}')
ok = False
print('=' * 68)
print(' RESULT:', 'PASS' if ok else 'FAIL')
print('=' * 68)
return 0 if ok else 1
if __name__ == '__main__':
sys.exit(main())
+718
View File
@@ -0,0 +1,718 @@
"""webview2_browser.py — Embedded WebView2 (Edge/Chromium) INSIDE Kivy's window.
Why this exists
---------------
The old weblink engines launched a *separate* browser process (Chrome/Edge
kiosk subprocess, or the dormant `cef_browser.py`). That model caused every
weblink bug in the tracker: the browser opening behind the Kivy window, being
handed off to an existing instance and exiting instantly, fighting for
foreground/z-order, and leaking `msedge.exe`/`chrome.exe` processes that were
never closed.
WebView2 renders as a **child HWND of Kivy's own SDL window**, so:
* no separate top-level window nothing can open "in the background",
* nothing to hand the URL off to no instant-exit hand-off,
* no z-order/foreground fight it is literally a child of our window,
* teardown is ours no leaked browser processes,
* the page renders at exactly the rectangle we give it (1920x1080 or
whatever the Kivy window currently is).
Licensing / distribution: the WebView2 **runtime** is a free, evergreen,
Microsoft-shipped component (already present on this host as
``152.0.4191.66``). We only ship the small managed SDK + native loader DLLs.
Implementation notes
--------------------
* We talk to the .NET SDK through **pythonnet** (``clr``).
* Every WebView2 API is async (returns a .NET ``Task``). We must NOT call
``.GetAwaiter().GetResult()``: the continuation needs the *same* thread's
message pump, so blocking would deadlock. Instead each task is **polled from
Kivy's Clock** (the SDL thread, which pumps Win32 messages) and consumed when
``IsCompleted``. This mirrors how the old CEF code pumped via the Clock.
* All public methods are safe to call from the Kivy main thread.
"""
from __future__ import annotations
import ctypes
import os
import sys
import threading
from pathlib import Path
# ── SDK discovery ────────────────────────────────────────────────────
# The managed Microsoft.Web.WebView2.Core.dll and the native
# WebView2Loader.dll must sit in a folder we can find both in development and
# inside the PyInstaller bundle.
_SDK_ENV_VAR = 'KIWY_WEBVIEW2_SDK'
def _sdk_candidates():
here = Path(__file__).resolve().parent
yield here / 'webview2_sdk'
# PyInstaller one-folder layout: bundled data lands next to the exe
# (sys._MEIPASS points at the temporary _MEIxxx dir).
meipass = getattr(sys, '_MEIPASS', None)
if meipass:
yield Path(meipass) / 'webview2_sdk'
yield here
def _find_sdk_dir():
env = os.environ.get(_SDK_ENV_VAR)
if env and (Path(env) / 'Microsoft.Web.WebView2.Core.dll').is_file():
return Path(env)
for candidate in _sdk_candidates():
try:
if (candidate / 'Microsoft.Web.WebView2.Core.dll').is_file():
return candidate
except OSError:
continue
return None
# ── Win32 helpers ────────────────────────────────────────────────────
_SW_HIDE = 0
_SW_SHOWNORMAL = 1
class WebView2Browser:
"""One embedded WebView2 instance parented to the Kivy (SDL) window.
Lifecycle::
show(url) -> is_showing() (True once painted) -> resize(w,h) -> hide()
-> shutdown()
``hide()`` only hides the controller (it stays alive), so switching back to
a weblink later is instant. ``shutdown()`` disposes it for good.
"""
#: Set True by the integration layer when the SDK + runtime are usable.
_import_error = None
def __init__(self, hwnd_provider=None, user_data_dir=None):
self._hwnd_provider = hwnd_provider
self._user_data_dir = user_data_dir or os.path.join(
os.environ.get('KIWY_DATA_DIR', os.getcwd()), '.webview2-profile'
)
self._env = None
self._controller = None
self._webview = None
self._hwnd = None
self._showing = False
self._stage = 'idle' # idle | env | controller | ready
self._pending_url = None
self._failed_reason = ''
self._poll_event = None
self._lock = threading.RLock()
self._task = None
self._task_kind = None
self._size = (0, 0)
# Navigation outcome. `_showing` only means "the controller was told to
# be visible", which happens the instant Navigate() is called — it says
# nothing about whether the page actually loaded. On a closed network
# that distinction is the whole point: an unreachable host paints a
# Chromium error page, so without this the player would show a blank
# error for the full slot instead of skipping the item.
self._navigation_ok = None # None = pending/unknown
self._navigation_status = ''
self._navigation_handlers = [] # keep refs: .NET must not GC these
# ── Availability ─────────────────────────────────────────────────
@staticmethod
def is_available():
"""True when pythonnet + the SDK DLLs + a runtime are all present."""
if sys.platform != 'win32':
return False
sdk = _find_sdk_dir()
if sdk is None:
return False
try:
import clr # noqa: F401 (pythonnet)
except Exception as exc:
WebView2Browser._import_error = f'pythonnet unavailable: {exc}'
return False
try:
cls = _load_webview2_types(sdk)
version = cls['env'].GetAvailableBrowserVersionString()
return bool(version)
except Exception as exc:
WebView2Browser._import_error = f'WebView2 unavailable: {exc}'
return False
# ── Public API (Kivy main thread) ────────────────────────────────
def show(self, url):
"""Begin displaying ``url``. Returns True once the request is accepted.
Rendering is asynchronous: the caller should poll :meth:`is_showing`
(the session's ``wait_visible`` does this on the watcher thread).
"""
with self._lock:
self._failed_reason = ''
self._pending_url = url
if self._stage == 'ready' and self._controller is not None:
return self._navigate(url)
if self._stage in ('env', 'controller'):
return True # already starting up; URL is queued
# Start-up order: environment → controller → navigate.
self._stage = 'env'
if not self._start_environment():
self._stage = 'idle'
return False
if self._stage != 'env':
# The environment resolved synchronously (fast path).
return self._after_environment()
return True
def hide(self):
"""Hide the page without destroying the controller (fast re-show)."""
with self._lock:
self._showing = False
self._pending_url = None
controller = self._controller
if controller is not None:
try:
controller.IsVisible = False
except Exception:
pass
def is_showing(self):
"""True while the page is actually on screen."""
with self._lock:
if self._failed_reason:
return False
if self._controller is None:
return False
return self._showing
def is_starting(self):
"""True while the environment/controller is still being created.
WebView2 start-up is asynchronous. A controller that does not exist yet
is NOT the same as a browser that has gone away, and conflating the two
made the *first* weblink after a cold start be skipped instantly (the
watcher saw "not alive" and advanced). Callers should treat
``is_starting()`` as "still alive, not yet painted".
"""
with self._lock:
if self._failed_reason:
return False
return self._stage in ('env', 'controller')
def is_alive(self):
"""True when the browser is starting up or showing. False only on failure."""
return self.is_showing() or self.is_starting()
@property
def failed_reason(self):
return self._failed_reason
def resize(self, width, height):
"""Fit the page to ``width`` x ``height`` physical pixels."""
width, height = int(width), int(height)
if width <= 0 or height <= 0:
return
with self._lock:
self._size = (width, height)
controller = self._controller
if controller is None:
return
try:
from System.Drawing import Rectangle
controller.Bounds = Rectangle(0, 0, width, height)
except Exception as exc:
_log(f'WebView2 resize failed (non-fatal): {exc}')
def shutdown(self):
"""Dispose the controller and environment. Never raises."""
with self._lock:
self._showing = False
self._stop_poll_locked()
controller, self._controller = self._controller, None
webview, self._webview = self._webview, None
env, self._env = self._env, None
self._stage = 'idle'
for obj, label in ((webview, 'webview'), (controller, 'controller')):
if obj is None:
continue
try:
dispose = getattr(obj, 'Dispose', None)
if dispose is not None:
dispose()
except Exception as exc:
_log(f'WebView2 {label} dispose failed (non-fatal): {exc}')
if ctypes is not None:
try:
ctypes.windll.ole32.CoUninitialize()
except Exception:
pass
del env
# ── Start-up ─────────────────────────────────────────────────────
def _start_environment(self):
sdk = _find_sdk_dir()
if sdk is None:
self._failed_reason = 'WebView2 SDK not found'
_log('WebView2: SDK DLLs not found (expected Microsoft.Web.WebView2.Core.dll)')
return False
try:
types = _load_webview2_types(sdk)
except Exception as exc:
self._failed_reason = f'WebView2 SDK load failed: {exc}'
_log(f'WebView2: SDK load failed: {exc}')
return False
# The controller must live on a thread with a message pump; Kivy's SDL
# thread qualifies, and COM must be initialised on it first.
try:
ctypes.windll.ole32.CoInitializeEx(None, 0x2) # STA
except Exception:
pass
try:
os.makedirs(self._user_data_dir, exist_ok=True)
except Exception as exc:
_log(f'WebView2: could not create profile dir ({exc}); using temp')
import tempfile
self._user_data_dir = tempfile.mkdtemp(prefix='kiwy-wv2-')
_log(f'WebView2: creating environment (profile={self._user_data_dir})')
try:
options = _build_environment_options()
task = _create_environment_async(types, self._user_data_dir, options)
except Exception as exc:
self._failed_reason = f'CreateAsync failed: {exc}'
_log(f'WebView2: environment creation failed: {exc}')
return False
self._task = task
self._task_kind = 'env'
self._start_poll()
return True
def _start_controller(self):
hwnd = 0
if self._hwnd_provider is not None:
try:
hwnd = self._hwnd_provider() or 0
except Exception as exc:
_log(f'WebView2: hwnd provider failed: {exc}')
if not hwnd:
self._failed_reason = 'Kivy window handle not found'
_log('WebView2: could not locate the Kivy SDL window handle')
return False
self._hwnd = int(hwnd)
_log(f'WebView2: creating controller inside hwnd=0x{self._hwnd:x}')
try:
# The parent window must be a .NET IntPtr; a plain Python int does
# not match the overload and pythonnet raises "No method matches
# given arguments".
from System import IntPtr
parent = IntPtr(self._hwnd)
# HWND hosting: WebView2 creates its own child window in `parent`.
task = self._env.CreateCoreWebView2ControllerAsync(parent)
except Exception as exc:
self._failed_reason = f'controller creation failed: {exc}'
_log(f'WebView2: controller creation failed: {exc}')
return False
self._task = task
self._task_kind = 'controller'
self._stage = 'controller'
self._start_poll()
return True
def _after_environment(self):
"""Called once the environment resolved."""
if self._env is None:
return False
started = self._start_controller()
if not started and self._stage == 'controller':
return True # still coming up asynchronously
return started
# ── Async task polling (Kivy Clock) ──────────────────────────────
def _start_poll(self):
try:
from kivy.clock import Clock
if self._poll_event is None:
self._poll_event = Clock.schedule_interval(self._tick, 0.05)
except Exception:
# No Kivy (or called off-thread): poll from a plain timer instead.
if self._poll_event is None:
self._poll_event = _ThreadTimer(0.05, self._tick, None)
def _stop_poll_locked(self):
event, self._poll_event = self._poll_event, None
if event is None:
return
try:
cancel = getattr(event, 'cancel', None)
if cancel is not None:
cancel()
else:
event.stop()
except Exception:
pass
def _tick(self, _dt):
"""Consume the in-flight Task once it completes."""
with self._lock:
task, kind = self._task, self._task_kind
if task is None:
self._stop_poll_locked()
return False
try:
done = bool(task.IsCompleted)
except Exception as exc:
self._failed_reason = f'task poll failed: {exc}'
self._task = None
self._stop_poll_locked()
return False
if not done:
return True
self._task, self._task_kind = None, None
self._stop_poll_locked()
try:
if task.IsFaulted:
exc = task.Exception
detail = ''
try:
detail = exc.GetBaseException().Message
except Exception:
detail = str(exc)
self._failed_reason = f'{kind} failed: {detail}'
_log(f'WebView2: {kind} task faulted: {detail}')
return False
result = task.Result
except Exception as exc:
self._failed_reason = f'{kind} task error: {exc}'
_log(f'WebView2: {kind} task error: {exc}')
return False
if kind == 'env':
self._env = result
_log('WebView2: environment ready')
if not self._start_controller():
self._stage = 'idle'
return False
if kind == 'controller':
self._controller = result
self._on_controller_ready()
return False
return False
def _on_controller_ready(self):
"""Wire up the page: bounds, settings, first navigation."""
controller = self._controller
try:
controller.IsVisible = False # stay hidden until navigated
except Exception:
pass
webview = None
try:
webview = controller.CoreWebView2
except Exception as exc:
_log(f'WebView2: CoreWebView2 unavailable: {exc}')
if webview is None:
self._failed_reason = 'CoreWebView2 was not created'
return
self._webview = webview
# Chrome-less, kiosk-like surface: no context menu, no devtools,
# no accelerators that could let an operator escape the signage.
try:
settings = webview.Settings
settings.AreDefaultContextMenusEnabled = False
settings.AreDevToolsEnabled = False
settings.IsStatusBarEnabled = False
settings.AreBrowserAcceleratorKeysEnabled = False
settings.IsZoomControlEnabled = False
settings.AreDefaultScriptDialogsEnabled = False
except Exception as exc:
_log(f'WebView2: settings tweak failed (non-fatal): {exc}')
self._hook_navigation_events(webview)
width, height = self._size
if width > 0 and height > 0:
self.resize(width, height)
self._stage = 'ready'
_log('WebView2: controller ready')
with self._lock:
url, self._pending_url = self._pending_url, None
if url:
self._navigate(url)
def _hook_navigation_events(self, webview):
"""Track whether the page actually loaded.
``is_showing()`` alone is misleading: it becomes True the moment
``Navigate()`` is called, before anything has been fetched. On a closed
network the weblink host is often unreachable, and Chromium then paints
an error page which the player must treat as a failure so the item is
skipped rather than shown as a broken screen for its whole slot.
Handlers are stored on the instance: if the delegate were only a local,
the .NET GC would collect it and the event would silently stop firing.
"""
try:
handler = _NavigationCompletedHandler(self)
webview.NavigationCompleted += handler
self._navigation_handlers.append(handler)
_log('WebView2: navigation tracking enabled')
except Exception as exc:
# Not fatal: without it we simply cannot distinguish a loaded page
# from an error page, and fall back to "visible means OK".
_log(f'WebView2: could not hook NavigationCompleted ({exc})')
def navigation_succeeded(self):
"""True / False once navigation finished, None while still pending."""
with self._lock:
return self._navigation_ok
def navigation_status(self):
with self._lock:
return self._navigation_status
def _navigate(self, url):
webview = self._webview
if webview is None:
return False
with self._lock:
self._navigation_ok = None
self._navigation_status = ''
try:
webview.Navigate(url)
except Exception as exc:
self._failed_reason = f'navigate failed: {exc}'
_log(f'WebView2: navigate failed: {exc}')
return False
width, height = self._size
if width > 0 and height > 0:
self.resize(width, height)
try:
self._controller.IsVisible = True
except Exception as exc:
_log(f'WebView2: could not show controller: {exc}')
return False
with self._lock:
self._showing = True
_log(f'WebView2: navigated to {url[:80]}')
return True
# ── Module helpers ───────────────────────────────────────────────────
_TYPES_CACHE = {}
def _load_webview2_types(sdk_dir):
"""Import the managed SDK and return the types we need (cached)."""
key = str(sdk_dir)
cached = _TYPES_CACHE.get(key)
if cached:
return cached
if hasattr(os, 'add_dll_directory'):
try:
os.add_dll_directory(str(sdk_dir)) # let the loader find WebView2Loader.dll
except Exception:
pass
if key not in sys.path:
sys.path.insert(0, key)
import clr
# Framework assemblies we rely on (Rectangle for Bounds).
try:
clr.AddReference('System.Drawing')
except Exception:
pass
clr.AddReference(str(sdk_dir / 'Microsoft.Web.WebView2.Core.dll'))
from Microsoft.Web.WebView2.Core import CoreWebView2Environment
types = {'env': CoreWebView2Environment}
_TYPES_CACHE[key] = types
return types
def _create_environment_async(types, user_data_dir, options=None):
"""Call CreateAsync with the options object.
The SDK exposes exactly one overload:
``CreateAsync(string browserExecutableFolder, string userDataFolder,
CoreWebView2EnvironmentOptions options)``.
"""
env_type = types['env']
last = None
# Preferred: explicit options (used to pass offline browser arguments).
if options is not None:
try:
return env_type.CreateAsync(None, user_data_dir, options)
except Exception as exc:
last = exc
attempts = (
(None, user_data_dir, None),
(None, user_data_dir),
)
for args in attempts:
try:
return env_type.CreateAsync(*args)
except Exception as exc:
last = exc
raise last if last is not None else RuntimeError('CreateAsync failed')
def _offline_browser_arguments():
"""Chromium flags that stop internet chatter on a closed network.
A signage player normally lives on an isolated LAN. By default Chromium
still tries to reach the internet for component updates, field trials,
safe-browsing lists, translate, and Google services. On a closed network
every one of those attempts has to time out, which costs start-up latency
(and, if DNS resolves but routes black-hole, can stall for many seconds).
These flags disable that background traffic. They do NOT affect loading
actual pages a weblink pointing at the local server still works, and one
pointing at the public internet simply fails fast with a normal
ERR_INTERNET_DISCONNECTED instead of hanging.
"""
return ' '.join([
'--disable-background-networking',
'--disable-component-update',
'--disable-domain-reliability',
'--disable-features=Translate,OptimizationHints,MediaRouter,'
'CalculateNativeWinOcclusion',
'--disable-sync',
'--no-first-run',
'--no-default-browser-check',
'--no-pings',
'--disable-breakpad',
'--metrics-recording-only',
'--disable-client-side-phishing-detection',
])
def _build_environment_options():
"""Create a CoreWebView2EnvironmentOptions with offline flags applied."""
try:
from Microsoft.Web.WebView2.Core import CoreWebView2EnvironmentOptions
options = CoreWebView2EnvironmentOptions()
options.AdditionalBrowserArguments = _offline_browser_arguments()
# Don't phone home with crash reports.
try:
options.IsCustomCrashReportingEnabled = False
except Exception:
pass
_log('WebView2: offline browser arguments applied')
return options
except Exception as exc:
_log(f'WebView2: could not build environment options ({exc}); '
'continuing with defaults')
return None
class _ThreadTimer:
"""Minimal fallback timer used only when Kivy's Clock is unavailable."""
def __init__(self, interval, func, _unused):
self._interval = float(interval)
self._func = func
self._stop = threading.Event()
self._thread = threading.Thread(target=self._run, daemon=True)
self._thread.start()
def _run(self):
while not self._stop.wait(self._interval):
try:
if self._func(None) is False:
return
except Exception:
return
def cancel(self):
self._stop.set()
class _NavigationCompletedHandler:
"""Adapter for WebView2's ``NavigationCompleted`` event.
The event is ``System.EventHandler<CoreWebView2NavigationCompletedEventArgs>``
there is no ``CoreWebView2NavigationCompletedEventHandler`` type to import
(attempting to import one fails). pythonnet converts a plain Python callable
to the generic delegate automatically, so that is what we pass.
The callable is kept on the browser instance: a delegate referenced only by
a local would be collected by the .NET GC, after which the event silently
stops firing.
"""
def __init__(self, browser):
self._browser = browser
def __call__(self, sender, args):
"""Fires on the WebView2 thread that owns the message loop."""
try:
success = bool(args.IsSuccess)
status = _describe_navigation_error(args, success) if not success else ''
with self._browser._lock:
self._browser._navigation_ok = success
self._browser._navigation_status = status
if success:
_log('WebView2: page loaded')
else:
_log(f'WebView2: page failed to load ({status or "unknown"})')
except Exception as exc:
_log(f'WebView2: navigation handler error ({exc})')
def _log(message):
try:
from kivy.logger import Logger
Logger.info(f'[WebView2] {message}')
except Exception:
print(f'[WebView2] {message}')
def _describe_navigation_error(args, success):
"""Human-readable reason for a failed navigation.
``WebErrorStatus`` is an enum whose numeric value is not useful on its own;
when it reports ``Unknown`` (common for connection-level failures) the HTTP
status is more informative, so prefer whichever actually says something.
"""
parts = []
try:
error_status = str(args.WebErrorStatus)
if error_status and error_status.lower() != 'unknown':
parts.append(error_status)
except Exception:
pass
try:
http_status = int(args.HttpStatusCode)
if http_status > 0:
parts.append(f'HTTP {http_status}')
except Exception:
pass
if parts:
return ', '.join(parts)
return 'connection failed (host unreachable or DNS failure)'
+446
View File
@@ -0,0 +1,446 @@
"""webview2_runtime.py — make sure the WebView2 Runtime is present.
Why this exists
---------------
WebView2 splits into two parts:
* the **SDK** (the ``Microsoft.Web.WebView2.Core.dll`` + ``WebView2Loader.dll``
we bundle in ``windows/webview2_sdk``), which is just the API surface, and
* the **Runtime** (``msedgewebview2.exe`` etc.), the actual Chromium engine.
The SDK is useless without the Runtime. The Runtime ships with Windows 11 and
is present on the vast majority of Windows 10 machines, but Microsoft still
recommends checking for it and installing it when missing so that is what
this module does.
Deployment notes (per Microsoft's distribution guidance):
* If the Runtime is missing we run an installer with ``/silent /install``.
* Run it **without elevation** per-user install, which never shows a UAC
prompt. That matters for an unattended signage player: a UAC dialog on a
kiosk screen is a failure, not a prompt.
* Two installers are supported, in order of preference:
1. ``MicrosoftEdgeWebView2RuntimeInstallerX64.exe`` the ~203 MB offline
*standalone* installer. Works with no internet (drop it in
``windows/webview2_runtime/`` to have it bundled).
2. ``MicrosoftEdgeWebview2Setup.exe`` the ~1.7 MB *bootstrapper*, which
downloads the Runtime from Microsoft. Bundled by default.
Nothing here ever raises: a failure just means web links fall back to the
Chrome/Edge subprocess engine, which is far better than the player crashing.
"""
from __future__ import annotations
import os
import subprocess
import sys
import threading
import time
from pathlib import Path
# Per Microsoft, the Runtime's presence/version lives in this registry value.
# (Edge Update client GUID for the Evergreen WebView2 Runtime.)
_CLIENT_GUID = '{F3017226-FE2A-4295-8BDF-00C3A9A7E4C5}'
_STANDALONE_NAME = 'MicrosoftEdgeWebView2RuntimeInstallerX64.exe'
_BOOTSTRAPPER_NAME = 'MicrosoftEdgeWebview2Setup.exe'
#: Don't re-attempt a failing install on every single start-up.
_RETRY_COOLDOWN_SECONDS = 6 * 60 * 60
_install_lock = threading.Lock()
_install_state = {
'attempted': False,
'installing': False,
'installed': None, # bool once known
'version': '',
'error': '',
}
#: Set once a background install has finished, so a weblink can wait for it.
_install_done = threading.Event()
# ── Detection ────────────────────────────────────────────────────────
def _parse_version(text):
"""Return a comparable tuple from a version string like '152.0.4191.66'."""
parts = []
for chunk in str(text or '').split('.'):
digits = ''.join(c for c in chunk if c.isdigit())
parts.append(int(digits) if digits else 0)
while len(parts) < 4:
parts.append(0)
return tuple(parts[:4])
def _read_registry_version():
"""Read the installed Runtime version from the registry, or ''.
Checks both install scopes: HKLM (per-machine) and HKCU (per-user). On
64-bit Windows the per-machine value lives under WOW6432Node because the
Edge Updater is a 32-bit component.
"""
if sys.platform != 'win32':
return ''
try:
import winreg
except Exception:
return ''
candidates = [
# (hive, subkey, access flag)
(winreg.HKEY_LOCAL_MACHINE,
rf'SOFTWARE\WOW6432Node\Microsoft\EdgeUpdate\Clients\{_CLIENT_GUID}', 0),
(winreg.HKEY_LOCAL_MACHINE,
rf'SOFTWARE\Microsoft\EdgeUpdate\Clients\{_CLIENT_GUID}',
getattr(winreg, 'KEY_WOW64_32KEY', 0)),
(winreg.HKEY_CURRENT_USER,
rf'SOFTWARE\Microsoft\EdgeUpdate\Clients\{_CLIENT_GUID}', 0),
]
for hive, subkey, access in candidates:
try:
with winreg.OpenKey(hive, subkey, 0,
winreg.KEY_READ | access) as key:
value, _ = winreg.QueryValueEx(key, 'pv')
value = str(value or '').strip()
if value and _parse_version(value) > (0, 0, 0, 0):
return value
except Exception:
continue
return ''
def get_runtime_version():
"""Version of the installed Evergreen Runtime, or '' when absent."""
version = _read_registry_version()
if version:
return version
# Fallback: ask the SDK itself (also covers preview channels).
try:
from webview2_browser import _find_sdk_dir, _load_webview2_types
sdk = _find_sdk_dir()
if sdk is not None:
types = _load_webview2_types(sdk)
reported = types['env'].GetAvailableBrowserVersionString()
return str(reported).strip() if reported else ''
except Exception:
pass
return ''
def is_runtime_installed():
"""True when a usable WebView2 Runtime is present."""
return bool(get_runtime_version())
# ── Installer discovery ──────────────────────────────────────────────
def _search_dirs():
"""Folders that may hold an installer, best (standalone) first."""
here = Path(__file__).resolve().parent
dirs = [here / 'webview2_runtime', here]
meipass = getattr(sys, '_MEIPASS', None)
if meipass:
dirs.append(Path(meipass) / 'webview2_runtime')
# Next to the .exe, so an operator can drop the offline installer in
# without rebuilding.
data_dir = os.environ.get('KIWY_DATA_DIR')
if data_dir:
dirs.append(Path(data_dir) / 'webview2_runtime')
dirs.append(Path(data_dir))
env = os.environ.get('KIWY_WEBVIEW2_INSTALLER')
if env:
dirs.insert(0, Path(env).parent)
return dirs
def find_installer():
"""Locate a usable installer. Returns ``(path, kind)`` or ``(None, None)``.
The standalone (offline) installer is preferred: it does not depend on the
target machine having internet access, which is the normal case for a
signage player on an isolated LAN.
"""
env = os.environ.get('KIWY_WEBVIEW2_INSTALLER')
if env and Path(env).is_file():
return Path(env), 'explicit'
found = {'standalone': None, 'bootstrapper': None}
for directory in _search_dirs():
try:
if found['standalone'] is None:
candidate = directory / _STANDALONE_NAME
if candidate.is_file():
found['standalone'] = candidate
if found['bootstrapper'] is None:
candidate = directory / _BOOTSTRAPPER_NAME
if candidate.is_file():
found['bootstrapper'] = candidate
except OSError:
continue
if found['standalone'] is not None:
return found['standalone'], 'standalone'
if found['bootstrapper'] is not None:
return found['bootstrapper'], 'bootstrapper'
return None, None
def _has_internet(timeout=4.0):
"""Quick reachability probe. A closed network returns False fast."""
try:
import socket
with socket.create_connection(('www.msftconnecttest.com', 80),
timeout=timeout):
return True
except Exception:
return False
# ── Cooldown bookkeeping ─────────────────────────────────────────────
def _marker_path():
data_dir = os.environ.get('KIWY_DATA_DIR') or os.getcwd()
return Path(data_dir) / 'logs' / '.webview2_install_attempt'
def _recent_failed_attempt():
try:
marker = _marker_path()
if not marker.is_file():
return False
age = time.time() - marker.stat().st_mtime
return age < _RETRY_COOLDOWN_SECONDS
except Exception:
return False
def _record_attempt():
try:
marker = _marker_path()
marker.parent.mkdir(parents=True, exist_ok=True)
marker.write_text(str(int(time.time())))
except Exception:
pass
def _clear_attempt_marker():
try:
marker = _marker_path()
if marker.is_file():
marker.unlink()
except Exception:
pass
# ── Install ──────────────────────────────────────────────────────────
def _create_no_window():
"""Keep the installer from flashing a console window on the signage."""
try:
return subprocess.CREATE_NO_WINDOW
except AttributeError:
return 0x08000000
def _run_installer(path, timeout):
"""Run the installer silently. Returns (ok, detail)."""
# `/silent /install` is the documented silent invocation. Deliberately NOT
# elevated: a non-elevated run performs a per-user install, which never
# raises a UAC prompt on the kiosk display.
args = [str(path), '/silent', '/install']
_log(f'WebView2: running installer {Path(path).name} /silent /install '
f'(per-user, no elevation)')
try:
result = subprocess.run(
args,
timeout=timeout,
creationflags=_create_no_window(),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
except subprocess.TimeoutExpired:
return False, f'installer timed out after {int(timeout)}s'
except Exception as exc:
return False, f'could not run installer: {exc}'
code = result.returncode
detail = (result.stdout or b'').decode('utf-8', 'replace').strip()
# Edge Update installers commonly report 0 (success) or 3010 (reboot
# required). They also return non-zero HRESULTs when the Runtime is already
# installed at an equal/newer version — which is why the caller decides
# success by re-reading the installed version rather than trusting this
# code. We only use it to explain a failure.
if code in (0, 3010):
return True, f'installer exit code {code}'
return False, f'installer exit code {code}{": " + detail if detail else ""}'
def ensure_runtime(timeout=600):
"""Install the Runtime when missing. Blocking; never raises.
Returns a dict describing the outcome (``installed``, ``version``,
``error``, ``action``).
"""
with _install_lock:
if is_runtime_installed():
version = get_runtime_version()
_install_state.update(
attempted=True, installing=False, installed=True,
version=version, error='',
)
return dict(_install_state, action='already-present')
if _recent_failed_attempt():
_install_state.update(
attempted=True, installing=False, installed=False, error='',
)
return dict(_install_state, action='skipped-recent-failure')
path, kind = find_installer()
if path is None:
message = ('no WebView2 installer found (expected '
f'{_STANDALONE_NAME} or {_BOOTSTRAPPER_NAME} in '
'windows/webview2_runtime/)')
_log(f'WebView2: {message}')
_install_state.update(
attempted=True, installing=False, installed=False,
error=message,
)
return dict(_install_state, action='installer-missing')
if kind == 'bootstrapper':
# The bootstrapper downloads the Runtime from Microsoft. On a closed
# network that can never succeed, so fail fast with an actionable
# message instead of hanging for the whole timeout.
if not _has_internet():
message = ('the WebView2 Runtime is missing and this machine has '
f'no internet access; only the ONLINE bootstrapper '
f'({_BOOTSTRAPPER_NAME}) is available. Bundle the '
f'offline installer ({_STANDALONE_NAME}, run '
'webview2_runtime/download_runtime_installers.ps1 '
'-Offline) to run on a closed network.')
_log(f'WebView2: {message}')
_install_state.update(
attempted=True, installing=False, installed=False,
error=message,
)
_install_done.set()
return dict(_install_state, action='offline-no-installer')
_log('WebView2: Runtime missing — using the ONLINE bootstrapper '
'(downloads ~150 MB). Add the offline standalone installer to '
'avoid needing internet.')
_install_state.update(attempted=True, installing=True, error='')
_record_attempt()
ok, detail = _run_installer(path, timeout)
if not ok and 'already installed' not in detail.lower():
_log(f'WebView2: installer reported {detail}')
# Decide success by RE-READING the installed version, not by the exit
# code: a non-zero HRESULT can simply mean "nothing to do".
version = ''
deadline = time.monotonic() + 30
while time.monotonic() < deadline:
version = get_runtime_version()
if version:
break
time.sleep(1.0)
if version:
_clear_attempt_marker()
_log(f'WebView2: Runtime available (v{version}) [{detail}]')
_install_state.update(
installing=False, installed=True, version=version,
error='', action=f'installed-{kind}',
)
else:
message = detail or 'Runtime still not detected after install'
_log(f'WebView2: install did not take effect ({message})')
_install_state.update(
installing=False, installed=False, version='', error=message,
)
_install_done.set()
return dict(_install_state, action=f'attempted-{kind}')
def ensure_runtime_async(timeout=600):
"""Kick off :func:`ensure_runtime` on a background thread.
Called at start-up so a missing Runtime installs while the player is still
syncing its playlist, instead of freezing the UI.
"""
if is_runtime_installed():
_install_done.set()
_install_state.update(installed=True, version=get_runtime_version())
return None
def _worker():
try:
ensure_runtime(timeout=timeout)
except Exception as exc: # defensive: never kill the player
_log(f'WebView2: background install failed: {exc}')
_install_state.update(installing=False, installed=False, error=str(exc))
_install_done.set()
thread = threading.Thread(target=_worker, name='webview2-install', daemon=True)
thread.start()
return thread
def wait_for_install(timeout):
"""Wait (briefly, on the watcher thread) for a pending install.
Returns True when a Runtime is available afterwards.
"""
if is_runtime_installed():
return True
if not _install_state.get('installing'):
return False
_install_done.wait(timeout=max(0.0, float(timeout)))
return is_runtime_installed()
def get_state():
"""Snapshot of the installer state, for logging/diagnostics."""
state = dict(_install_state)
if state.get('installed') is None:
state['installed'] = is_runtime_installed()
state['version'] = state['version'] or get_runtime_version()
return state
def describe():
"""One-line status for the startup log."""
version = get_runtime_version()
if version:
return f'WebView2 Runtime present (v{version})'
path, kind = find_installer()
if path is None:
return 'WebView2 Runtime MISSING and no bundled installer found'
if kind == 'standalone':
return (f'WebView2 Runtime MISSING (will install OFFLINE via '
f'{path.name} — no internet needed)')
return (f'WebView2 Runtime MISSING (will install via the ONLINE '
f'bootstrapper {path.name}; needs internet)')
def is_offline_ready():
"""True when a Runtime is present, or can be installed without internet.
This is the property that matters for a closed-network deployment: web
links will work on first start with no outbound connectivity.
"""
if is_runtime_installed():
return True
path, kind = find_installer()
return path is not None and kind in ('standalone', 'explicit')
def _log(message):
try:
from kivy.logger import Logger
Logger.info(f'[WebView2] {message}')
except Exception:
print(f'[WebView2] {message}')
@@ -0,0 +1,65 @@
# Downloads the WebView2 Runtime installers into windows\webview2_runtime\.
#
# build.spec bundles:
# - MicrosoftEdgeWebview2Setup.exe (~1.7 MB) always
# - MicrosoftEdgeWebView2RuntimeInstallerX64.exe (~203 MB) only if present
#
# The bootstrapper is the small online installer (it downloads the Runtime
# from Microsoft). Run this script with -Offline to also fetch the standalone
# installer for machines that have no internet access — note that it makes the
# built .exe about 200 MB larger.
#
# Usage:
# .\download_runtime_installers.ps1
# .\download_runtime_installers.ps1 -Offline
[CmdletBinding()]
param(
[switch]$Offline
)
$ErrorActionPreference = 'Stop'
$dest = Join-Path $PSScriptRoot 'webview2_runtime'
if (-not (Test-Path $dest)) {
New-Item -ItemType Directory -Force -Path $dest | Out-Null
}
$bootstrapperUrl = 'https://go.microsoft.com/fwlink/p/?LinkId=2124703'
$standaloneUrl = 'https://go.microsoft.com/fwlink/?linkid=2124701' # x64
function Get-Installer {
param([string]$Url, [string]$FileName, [string]$Label)
$target = Join-Path $dest $FileName
Write-Host "[INFO] Downloading $Label ..." -ForegroundColor Cyan
Invoke-WebRequest -Uri $Url -OutFile $target -UseBasicParsing -MaximumRedirection 10
$file = Get-Item -LiteralPath $target
$sig = Get-AuthenticodeSignature -LiteralPath $target
$sizeMb = [math]::Round($file.Length / 1MB, 1)
Write-Host (" {0} {1} MB" -f $file.Name, $sizeMb)
if ($sig.Status -eq 'Valid' -and $sig.SignerCertificate.Subject -like '*Microsoft*') {
Write-Host " signature: Valid (Microsoft)" -ForegroundColor Green
}
else {
Write-Warning " signature: $($sig.Status) — verify this download!"
}
}
Get-Installer -Url $bootstrapperUrl -FileName 'MicrosoftEdgeWebview2Setup.exe' -Label 'Runtime bootstrapper (online)'
if ($Offline) {
Get-Installer -Url $standaloneUrl -FileName 'MicrosoftEdgeWebView2RuntimeInstallerX64.exe' -Label 'Runtime standalone installer (offline, x64)'
Write-Host ''
Write-Host '[WARN] The standalone installer adds ~203 MB to the built .exe.' -ForegroundColor Yellow
}
else {
Write-Host ''
Write-Host '[INFO] Offline installer skipped. Re-run with -Offline to include it.' -ForegroundColor DarkGray
}
Write-Host ''
Write-Host "[OK] Installers are in $dest" -ForegroundColor Green
Write-Host ' Next: rebuild with venv\Scripts\python.exe -m PyInstaller build.spec --clean --noconfirm'
Binary file not shown.
Binary file not shown.