svc: Save config after navigation

When a makes a *POST /screen/{name}/navigate* HTTP request, the URL list
is updated and saved to the configuration file.  This provides a
mechanism for configuring the URL list, since it is now part of the
display configuration file and not a separate file manged externally.

svc/.gitignore

@@ -4,3 +4,4 @@
 .mypy_cache/
 __pycache__/
 *.py[co]
+/config.json

svc/src/hudctrl/api.py

@@ -1,7 +1,5 @@
 import io
 import logging
-import os
-from pathlib import Path
 from typing import Optional
 
 from PIL import Image
@@ -17,16 +15,11 @@ log = logging.getLogger(__name__)
 logging.basicConfig(level=logging.DEBUG)
 
 
-HUDCTRL_URLS_FILE = os.environ.get('HUDCTRL_URLS_FILE')
-
-
 app = fastapi.FastAPI(
     docs_url='/api-doc/',
 )
 
 svc = HUDService()
-if HUDCTRL_URLS_FILE:
-    svc.urls_file = Path(HUDCTRL_URLS_FILE)
 
 
 class PNGImageResponse(fastapi.Response):
@@ -61,7 +54,8 @@ async def get_monitor_config():
 
 @app.put('/display/monitors')
 async def put_monitor_config(
-    monitors: str = fastapi.Body(..., media_type='text/plain')
+    bgtasks: fastapi.BackgroundTasks,
+    monitors: str = fastapi.Body(..., media_type='text/plain'),
 ):
     try:
         svc.monitor_config = MonitorConfig.from_string(monitors)
@@ -71,6 +65,7 @@ async def put_monitor_config(
             fastapi.status.HTTP_400_BAD_REQUEST,
             detail=f'Invalid monitor config: {e}',
         )
+    bgtasks.add_task(svc.save_config)
     return {'monitor_config': svc.monitor_config}
 
 
@@ -142,8 +137,14 @@ async def get_screenshot(
     '/screen/{name}/navigate',
     response_class=fastapi.responses.PlainTextResponse,
 )
-async def navigate(name: str, url: str = fastapi.Form(...)):
+async def navigate(
+    bgtasks: fastapi.BackgroundTasks,
+    name: str,
+    url: str = fastapi.Form(...),
+):
     await svc.navigate(name, url)
+    svc.urls[name] = url
+    bgtasks.add_task(svc.save_config)
 
 
 @app.on_event('shutdown')

svc/src/hudctrl/config.py

@@ -0,0 +1,12 @@
+from typing import Optional
+
+import pydantic
+
+from .xrandr import Monitor
+
+
+class Configuration(pydantic.BaseModel):
+    monitors: list[Monitor] = pydantic.Field(default_factory=list)
+    urls: dict[str, str] = pydantic.Field(default_factory=dict)
+    host: Optional[str] = None
+    port: Optional[int] = None

svc/src/hudctrl/hud.py

@@ -2,18 +2,23 @@ import asyncio
 import base64
 import json
 import logging
+import os
 from pathlib import Path
 from typing import Dict, Optional
 
 import pydantic
 from aiomarionette import Marionette, WindowRect
 
+from .config import Configuration
 from .xrandr import MonitorConfig
 
 
 log = logging.getLogger(__name__)
 
 
+CONFIG_PATH = os.environ.get('HUDCTRL_CONFIG_PATH', 'config.json')
+
+
 class NoMonitorConfig(Exception):
     '''Raised when no monitor config has been provided yet'''
 
@@ -34,7 +39,7 @@ class HUDService:
         self.urls: Dict[str, str] = {}
         self.windows: Dict[str, str] = {}
 
-        self.urls_file = Path('urls.json')
+        self.config_file = Path(CONFIG_PATH)
         self.lock = asyncio.Lock()
 
     async def get_screen(self, name: str) -> HUDScreen:
@@ -104,29 +109,27 @@ class HUDService:
             if tasks:
                 await asyncio.wait(tasks)
 
-    def load_urls(self) -> None:
+    def load_config(self) -> None:
         try:
-            f = self.urls_file.open(encoding='utf-8')
+            f = self.config_file.open(encoding='utf-8')
         except FileNotFoundError:
             return
         except OSError as e:
-            log.error('Could not load URL list: %s', e)
+            log.error('Could not load config: %s', e)
             return
-        log.info('Loading URL list from %s', self.urls_file)
+        log.info('Loading config from %s', f.name)
         with f:
             try:
-                value = json.load(f)
+                config = Configuration.parse_obj(json.load(f))
             except ValueError as e:
-                log.error('Failed to load URL list: %s', e)
+                log.error('Failed to load config: %s', e)
                 return
-            if isinstance(value, dict):
+            self.urls = config.urls
-                self.urls = value
+            if config.monitors:
-            else:
+                self.monitor_config = MonitorConfig()
-                log.error(
+                self.monitor_config.monitors = config.monitors
-                    'Invalid URL list: found %r, expected %r',
+            self.host = config.host
-                    type(value),
+            self.port = config.port
-                    dict,
-                )
 
     async def navigate(self, name: str, url: str) -> None:
         assert self.marionette
@@ -140,6 +143,19 @@ class HUDService:
             await self.marionette.switch_to_window(self.windows[name])
             await self.marionette.refresh()
 
+    def save_config(self) -> None:
+        log.info('Saving configuration to %s', self.config_file)
+        config = Configuration(
+            monitors=self.monitor_config.monitors
+            if self.monitor_config
+            else [],
+            urls=self.urls,
+            host=self.host,
+            port=self.port,
+        )
+        f = self.config_file.open('w', encoding='utf-8')
+        f.write(config.json())
+
     async def set_display(self, host: str, port: int) -> None:
         if self.marionette:
             log.warning('Closing existing Marionette connection')
@@ -154,10 +170,17 @@ class HUDService:
         if self.marionette is not None:
             await self.marionette.close()
             self.marionette = None
+        self.save_config()
 
     async def startup(self) -> None:
-        loop = asyncio.get_running_loop()
+        asyncio.create_task(self._startup())
-        await loop.run_in_executor(None, self.load_urls)
+
+    async def _startup(self) -> None:
+        await asyncio.to_thread(self.load_config)
+        if self.host and self.port:
+            await self.set_display(self.host, self.port)
+            if self.monitor_config:
+                await self.initialize()
 
     async def take_screenshot(self, screen: str) -> bytes:
         assert self.marionette