1
0
Fork 0

svc: Load URL list from file

Instead of hard-coding the list of URLs to open, we'll read it from a
JSON file on the disk.  The file contains a mapping of monitor names to
URLs, e.g.

```json
{
    "HDMI-1": "http://my.site.one",
    "HDMI-2": "http://my.site.two"
}
```
master
Dustin 2022-04-30 13:55:58 -05:00
parent 939f24d79f
commit dbf266de5b
2 changed files with 50 additions and 15 deletions

View File

@ -86,3 +86,8 @@ async def refresh_screen(number: int):
@app.on_event('shutdown')
async def on_shutdown():
await svc.shutdown()
@app.on_event('startup')
async def on_startup():
await svc.startup()

View File

@ -1,6 +1,8 @@
import asyncio
import json
import logging
from typing import List, Optional
from pathlib import Path
from typing import Dict, List, Optional
import pydantic
from aiomarionette import Marionette, WindowRect
@ -10,11 +12,6 @@ from .xrandr import MonitorConfig
log = logging.getLogger(__name__)
URLS = [
'https://grafana.pyrocufflink.blue/d/p3CPbi87z/hud?kiosk',
'https://grafana.pyrocufflink.blue/d/lGT5FHOnz/home?kiosk',
]
class NoMonitorConfig(Exception):
'''Raised when no monitor config has been provided yet'''
@ -33,6 +30,9 @@ class HUDService:
self.port: Optional[int] = None
self.monitor_config: Optional[MonitorConfig] = None
self.marionette: Optional[Marionette] = None
self.urls: Dict[str, str] = {}
self.urls_file = Path('urls.json')
async def get_screens(self) -> List[HUDScreen]:
assert self.marionette
@ -57,30 +57,56 @@ class HUDService:
'Initializing display for %d monitors',
len(self.monitor_config.monitors),
)
window = await self.marionette.new_window('window')
window: Optional[str] = await self.marionette.new_window('window')
for handle in await self.marionette.get_window_handles():
if handle == window:
continue
await self.marionette.switch_to_window(handle)
await self.marionette.close_window()
tasks = []
for i, monitor in enumerate(self.monitor_config.monitors):
for monitor in self.monitor_config.monitors:
try:
url = URLS[i]
except IndexError:
break
url = self.urls[monitor.name]
except KeyError:
continue
if window is None:
window = await self.marionette.new_window('window')
await self.marionette.switch_to_window(window)
await self.marionette.set_window_rect(
x=monitor.pos_x,
y=1,
)
await self.marionette.fullscreen()
log.info('Screen %d: Opening URL %s', i, url)
log.info('Screen %s: Opening URL %s', monitor.name, url)
tasks.append(asyncio.create_task(self.marionette.navigate(url)))
if i + 1 < min(len(URLS), len(self.monitor_config.monitors)):
window = await self.marionette.new_window('window')
window = None
if tasks:
await asyncio.wait(tasks)
def load_urls(self) -> None:
try:
f = self.urls_file.open(encoding='utf-8')
except FileNotFoundError:
return
except OSError as e:
log.error('Could not load URL list: %s', e)
return
log.info('Loading URL list from %s', self.urls_file)
with f:
try:
value = json.load(f)
except ValueError as e:
log.error('Failed to load URL list: %s', e)
return
if isinstance(value, dict):
self.urls = value
else:
log.error(
'Invalid URL list: found %r, expected %r',
type(value),
dict,
)
async def refresh_screen(self, number: int) -> None:
assert self.marionette
windows = await self.marionette.get_window_handles()
@ -101,3 +127,7 @@ class HUDService:
if self.marionette is not None:
await self.marionette.close()
self.marionette = None
async def startup(self) -> None:
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self.load_urls)