Crawl pvi screens

src/techui_builder/generate_jsonmap.py

@@ -11,6 +11,7 @@
 
 from techui_builder._logger import Logger
 from techui_builder.jsonmap.crawl import CrawlContext, crawl
+from techui_builder.jsonmap.fetch import ScreenFetcher
 from techui_builder.jsonmap.nodes import ScreenNode, serialise_node
 from techui_builder.models import TechUi
 
@@ -46,6 +47,7 @@ class JsonMapGenerator:
     bob_path: Path = field(default=Path("index.bob"))
     techui: Path = field(default=Path("techui.yaml"))
     output: Path | None = field(default=None)
+    fetcher: ScreenFetcher = field(default_factory=ScreenFetcher)
 
     def __post_init__(self):
         # Determine the directory to write the json map file to.
@@ -72,17 +74,16 @@ def __post_init__(self):
     def generate_json_map(
         self,
         screen_path: Path,
-        dest_path: Path,
         current_component_name: str | None = None,
         name_elem: str | None = None,
     ) -> ScreenNode:
         """Recursively generate JSON map from .bob file tree"""
         ctx = CrawlContext(
             components=self.techui_yaml.components,
             synoptic_dir=self._parent_path,
-            link_base_dir=dest_path,
+            fetcher=self.fetcher,
             component_name=current_component_name,
-            service_name="",
+            screen=screen_path,
         )
         return crawl(screen_path, ctx, link_name=name_elem)
 
@@ -95,7 +96,7 @@ def write_json_map(
                 f"Cannot generate json map for {self.bob_path}. Has it been generated?"
             )
 
-        json_map = self.generate_json_map(self.bob_path, self._parent_path)
+        json_map = self.generate_json_map(self.bob_path)
         with open(self._write_directory / "JsonMap.json", "w") as f:
             f.write(
                 json.dumps(json_map, indent=4, default=lambda o: serialise_node(o))

src/techui_builder/jsonmap/crawl.py

@@ -1,18 +1,21 @@
 """Recursively crawl a tree of .bob screens into a tree of ScreenNodes."""
 
+import logging
 from collections.abc import Mapping
-from dataclasses import dataclass, replace
+from dataclasses import dataclass, field, replace
 from pathlib import Path
+from urllib.error import HTTPError
+from urllib.parse import urlsplit
 
 from lxml import etree, objectify
 from lxml.objectify import ObjectifiedElement
 
+from techui_builder.jsonmap.fetch import ScreenFetcher
 from techui_builder.jsonmap.links import (
     WidgetLink,
-    WidgetType,
-    assumed_exists,
     extract_links,
-    find_local_screen,
+    resolve_link,
+    substitute_macros,
 )
 from techui_builder.jsonmap.naming import (
     find_techui_label,
@@ -21,31 +24,53 @@
 )
 from techui_builder.jsonmap.nodes import ScreenNode
 from techui_builder.models import Component
+from techui_builder.utils import WidgetType
+
+logger_ = logging.getLogger(__name__)
 
 
 @dataclass
 class CrawlContext:
     """State passed down the recursion."""
 
     components: Mapping[str, Component]
-    synoptic_dir: Path  # ScreenNode.file is relative to this
-    link_base_dir: Path  # link files are resolved against this; never changes
+    synoptic_dir: Path  # local ScreenNode.file is relative to this
+    fetcher: ScreenFetcher
     component_name: str | None
-    service_name: str
+    screen: Path | str  # the screen being crawled, as a local path or URL
+    macros: dict[str, str] = field(default_factory=dict)  # including inherited ones
+
+    def with_screen(self, screen: Path | str) -> "CrawlContext":
+        """A copy for crawling a screen's links, in its component if it names one."""
+        component_name = self.component_name
+        # Only a local screen in the synoptic directory can name a component
+        if component_name is None and isinstance(screen, Path):
+            if screen.stem in self.components:
+                component_name = screen.stem
+
+        return replace(self, screen=screen, component_name=component_name)
+
+
+def find_screen_file(screen: Path | str) -> Path:
+    """The screen's file name, for a local path or a URL with a query or fragment."""
+    if isinstance(screen, str):
+        return Path(urlsplit(screen).path)
+    return screen
+
 
-    def with_screen_component(self, screen_path: Path) -> "CrawlContext":
-        """A copy for the screen's component, if it is one and none is set yet."""
-        if self.component_name is not None or screen_path.stem not in self.components:
-            return self
+def format_screen(screen: Path | str, synoptic_dir: Path) -> str:
+    """The URL, or the local path relative to the synoptic directory."""
+    if isinstance(screen, str):
+        return screen
+    return str(screen.resolve().relative_to(synoptic_dir.resolve(), walk_up=True))
 
-        component_name = screen_path.stem
-        # We know from the if statement that it exists
-        component = self.components.get(component_name)
-        assert isinstance(component, Component)
-        # TODO: How to find the screens if PV prefix is not the service name???
-        return replace(
-            self, component_name=component_name, service_name=component.prefix.lower()
-        )
+
+def inherit_macros(
+    parent_macros: Mapping[str, str], macros: Mapping[str, str]
+) -> dict[str, str]:
+    """The parent's macros overridden by a link's macros, expanded like Phoebus."""
+    expanded = {k: substitute_macros(v, parent_macros) for k, v in macros.items()}
+    return {**parent_macros, **expanded}
 
 
 def crawl_link(
@@ -57,65 +82,81 @@ def crawl_link(
 
     If it can't be found, a leaf ScreenNode is returned.
     """
-    local_path = find_local_screen(link.file, ctx.link_base_dir, ctx.service_name)
+    macros = inherit_macros(ctx.macros, link.macros)
+    screen = resolve_link(link.file, macros, ctx.screen)
+    leaf = ScreenNode(
+        format_screen(screen, ctx.synoptic_dir), display_name, macros=macros
+    )
+
+    if isinstance(screen, str):
+        try:
+            ctx.fetcher.fetch(screen)
+        except HTTPError as e:
+            leaf.exists = False
+            leaf.error = f"Could not fetch screen: {e}"
+            return leaf
+        except OSError as e:
+            # The server may just be unreachable from here, so assume it exists
+            leaf.error = f"Could not fetch screen: {e}"
+            return leaf
+
+    elif not screen.is_file():
+        leaf.exists = False
+        logger_.debug(f"Link {link.file} -> {screen}: not found")
+        return leaf
+
+    logger_.debug(f"Link {link.file} -> {screen}: found")
 
     # Crawl the next file
-    if local_path is not None:
-        # TODO: investigate non-recursive approaches?
-        return crawl(local_path, ctx, link_name=link.name)
-
-    return ScreenNode(
-        link.file,
-        display_name,
-        exists=assumed_exists(link.file, link.macros),
-    )
+    # TODO: investigate non-recursive approaches?
+    node = crawl(screen, replace(ctx, macros=macros), link_name=link.name)
+    node.macros = macros
+    return node
+
+
+def parse_screen(screen: Path | str, fetcher: ScreenFetcher) -> ObjectifiedElement:
+    """Parse a local or remote .bob screen."""
+    if isinstance(screen, str):
+        return objectify.fromstring(fetcher.fetch(screen), base_url=screen)
+    return objectify.parse(screen.absolute()).getroot()
 
 
 def crawl(
-    screen_path: Path, ctx: CrawlContext, link_name: str | None = None
+    screen: Path | str, ctx: CrawlContext, link_name: str | None = None
 ) -> ScreenNode:
     """Crawl a .bob screen and the screens it links to into a ScreenNode."""
 
     # Create initial node at top of .bob file
     current_node = ScreenNode(
-        str(
-            screen_path.resolve().relative_to(ctx.synoptic_dir.resolve(), walk_up=True)
-        ),
-        display_name=None,
+        format_screen(screen, ctx.synoptic_dir), display_name=None
     )
 
-    ctx = ctx.with_screen_component(screen_path)
+    ctx = ctx.with_screen(screen)
 
     try:
         # Create xml tree from .bob file
-        tree = objectify.parse(screen_path.absolute())
-        root: ObjectifiedElement = tree.getroot()
+        root = parse_screen(screen, ctx.fetcher)
 
         # Label for the linking widget, else the screen's own <name>, else file stem
-        own_name = name_or_file_stem(root.name.text, screen_path)
+        own_name = name_or_file_stem(root.name.text, find_screen_file(screen))
         label = find_techui_label(ctx.components, ctx.component_name, link_name)
         current_node.display_name = label if label is not None else own_name
 
         for link in extract_links(root):
             # Label, else widget <name>, else file stem
             label = find_techui_label(ctx.components, ctx.component_name, link.name)
             display_name = name_or_file_stem(
-                label if label is not None else link.name, Path(link.file)
+                label if label is not None else link.name, find_screen_file(link.file)
             )
 
             child_node = crawl_link(link, display_name, ctx)
 
             if link.type == WidgetType.EMBEDDED:
                 for embedded_child in child_node.children:
-                    embedded_child.macros = {**embedded_child.macros, **link.macros}
                     embedded_child.display_name = display_name
-                    embedded_child.exists = "IOC" in link.macros or (
-                        "https://" in str(embedded_child.file)
-                    )
                     current_node.children.append(embedded_child)
 
             else:
-                child_node.macros = link.macros
                 # TODO: make this work for only list[ScreenNode]
                 assert isinstance(current_node.children, list)
                 # TODO: fix typing

src/techui_builder/jsonmap/fetch.py

@@ -0,0 +1,48 @@
+"""Fetching remote .bob screens over http(s)."""
+
+import logging
+from dataclasses import dataclass, field
+from urllib.error import HTTPError
+from urllib.parse import urlparse
+from urllib.request import urlopen
+
+logger_ = logging.getLogger(__name__)
+
+TIMEOUT_SECONDS = 10
+
+
+def download_url(url: str) -> bytes:
+    """Download the contents of a URL."""
+    with urlopen(url, timeout=TIMEOUT_SECONDS) as response:
+        return response.read()
+
+
+@dataclass
+class ScreenFetcher:
+    """Fetches remote screens."""
+
+    _screens: dict[str, bytes] = field(default_factory=dict)
+    _unreachable_hosts: dict[str, OSError] = field(default_factory=dict)
+
+    def fetch(self, url: str) -> bytes:
+        """Fetch a screen, raising HTTPError if missing or OSError if unreachable."""
+        if url in self._screens:
+            return self._screens[url]
+
+        # Don't wait for a timeout on every screen from a host that is down
+        host = urlparse(url).netloc
+        if host in self._unreachable_hosts:
+            raise self._unreachable_hosts[host]
+
+        try:
+            screen = download_url(url)
+        except HTTPError as e:
+            logger_.warning(f"Could not fetch {url}: {e}")
+            raise
+        except OSError as e:
+            logger_.warning(f"Could not reach {host}, not crawling its screens: {e}")
+            self._unreachable_hosts[host] = e
+            raise
+
+        self._screens[url] = screen
+        return screen

src/techui_builder/jsonmap/links.py

@@ -1,9 +1,11 @@
 """Links from a .bob screen to other screens."""
 
+import logging
 import re
 from collections.abc import Iterator, Mapping
 from dataclasses import dataclass
 from pathlib import Path
+from urllib.parse import urljoin, urlsplit
 
 from lxml.objectify import ObjectifiedElement
 
@@ -14,7 +16,9 @@
     _get_nav_tabs,
 )
 
-PVI_FILE_RE = re.compile(r"^(?:\$\(IOC\))\/([a-zA-Z]+[.a-zA-Z]+)$")
+logger_ = logging.getLogger(__name__)
+
+MACRO_RE = re.compile(r"\$(?:\((\w+)\)|\{(\w+)\})")
 
 
 @dataclass
@@ -34,8 +38,8 @@ def extract_file_text(file_elem: ObjectifiedElement) -> str:
 
 
 def is_bob(file: str) -> bool:
-    """Whether the file is a .bob screen."""
-    return Path(file).suffix == ".bob"
+    """Whether the link is to a .bob screen"""
+    return urlsplit(file).path.endswith(".bob")  # ignores url queries
 
 
 def extract_links(root: ObjectifiedElement) -> Iterator[WidgetLink]:
@@ -82,6 +86,7 @@ def extract_links(root: ObjectifiedElement) -> Iterator[WidgetLink]:
                     file = extract_file_text(file_elem)
                     # Skip links that are not .bob screens
                     if not is_bob(file):
+                        logger_.debug(f"Skipping link to {file}: not a .bob screen")
                         continue
 
                     yield WidgetLink(file, name, widget_type, macros)
@@ -95,29 +100,34 @@ def extract_links(root: ObjectifiedElement) -> Iterator[WidgetLink]:
         file = extract_file_text(file_elem)
         # Skip links that are not .bob screens
         if not is_bob(file):
+            logger_.debug(f"Skipping link to {file}: not a .bob screen")
             continue
 
         yield WidgetLink(file, name, widget_type, macros)
 
 
-def resolve_link_path(file: str, base_dir: Path, service_name: str) -> Path:
-    """Resolve a link's file to a local path."""
+def is_url(file: str) -> bool:
+    """Whether the file is an http(s) URL."""
+    return file.startswith(("http://", "https://"))
 
-    match = PVI_FILE_RE.fullmatch(file)
-    # The file path is a PVI screen, so attempt to find that screen
-    if match:
-        file_name = match.group(1)
-        return base_dir / f"../{service_name}/{file_name}"
 
-    return base_dir / file
+def substitute_macros(text: str, macros: Mapping[str, str]) -> str:
+    """Replace $(NAME) and ${NAME} with macro values, leaving unknown macros as-is."""
+    return MACRO_RE.sub(
+        lambda m: macros.get(m.group(1) or m.group(2), m.group(0)), text
+    )
 
 
-def find_local_screen(file: str, base_dir: Path, service_name: str) -> Path | None:
-    """Resolve a link's file, returning the path if it can be crawled locally."""
-    path = resolve_link_path(file, base_dir, service_name)
-    return path if path.is_file() else None
+def resolve_link(
+    file: str, macros: Mapping[str, str], screen: Path | str
+) -> Path | str:
+    """Resolve a link's file to a URL or local path, relative to the linking screen."""
+    file = substitute_macros(file, macros)
+    if is_url(file):
+        return file
 
+    # Phoebus resolves relative files against the display containing the link
+    if isinstance(screen, str):
+        return urljoin(screen, file)
 
-def assumed_exists(file: str, macros: Mapping[str, str]) -> bool:
-    """Whether a link's file that could not be found locally is assumed to exist."""
-    return "IOC" in macros or ("https:/" in file)
+    return screen.parent / file