Coverage for bioimageio/spec/partner_utils/imjoy/_plugin_parser.py: 17%
131 statements
« prev ^ index » next coverage.py v7.9.1, created at 2025-06-27 09:20 +0000
1# pragma: no cover
2# type: ignore
3"""ImJoy plugin parser module."""
4import copy
5import json
6import uuid
7import warnings
8from pathlib import Path
9from typing import Any, Callable, Dict, Tuple, Union
10from urllib.parse import urljoin
12import httpx
13from pydantic import DirectoryPath, FilePath, HttpUrl
14from ruyaml import YAML
16yaml = YAML(typ="safe")
18tag_types = ["config", "script", "link", "window", "style", "docs", "attachment"]
20CONFIGURABLE_FIELDS = [
21 "env",
22 "requirements",
23 "dependencies",
24 "icon",
25 "ui",
26 "type",
27 "flags",
28 "labels",
29 "cover",
30 "base_frame",
31 "base_worker",
32 "passive",
33]
36class dotdict(dict): # pylint: disable=invalid-name
37 """Access dictionary attributes with dot.notation."""
39 __getattr__ = dict.get
40 __setattr__ = dict.__setitem__
41 __delattr__ = dict.__delitem__
43 def __deepcopy__(self, memo=None):
44 """Make a deep copy."""
45 return dotdict(copy.deepcopy(dict(self), memo=memo))
48def parse_imjoy_plugin(source, overwrite_config=None):
49 """Parse ImJoy plugin file and return a dict with all the fields."""
50 from lxml import etree
52 root = etree.HTML("<html>" + source + "</html>")
53 plugin_comp = dotdict()
54 for tag_type in tag_types:
55 elms = root.xpath(f".//{tag_type}")
56 values = []
57 for elm in elms:
58 values.append(
59 dotdict(
60 attrs=dotdict(elm.attrib),
61 content=elm.text,
62 )
63 )
64 plugin_comp[tag_type] = values
65 if plugin_comp.config[0].attrs.lang == "yaml":
66 config = yaml.load(plugin_comp.config[0].content)
67 elif plugin_comp.config[0].attrs.lang == "json":
68 config = json.loads(plugin_comp.config[0].content)
69 else:
70 raise Exception(
71 "Unsupported config language: " + plugin_comp.config[0].attrs.lang
72 )
74 overwrite_config = overwrite_config or {}
75 config["tag"] = overwrite_config.get("tag") or (
76 config.get("tags") and config.get("tags")[0]
77 )
78 config["hot_reloading"] = overwrite_config.get("hot_reloading")
79 config["scripts"] = []
80 # try to match the script with current tag
81 for elm in plugin_comp.script:
82 if elm.attrs.tag == config["tag"]:
83 config["script"] = elm.content
84 # exclude script with mismatched tag
85 if not elm.attrs.tag or elm.attrs.tag == config["tag"]:
86 config["scripts"].append(elm)
87 if not config.get("script") and len(plugin_comp.script) > 0:
88 config["script"] = plugin_comp.script[0].content
89 config["lang"] = plugin_comp.script[0].attrs.lang
90 config["links"] = plugin_comp.link or None
91 config["windows"] = plugin_comp.window or None
92 config["styles"] = plugin_comp.style or None
93 config["docs"] = plugin_comp.docs[0] if plugin_comp.docs else config.get("docs")
94 config["attachments"] = plugin_comp.attachment or None
96 config["_id"] = overwrite_config.get("_id") or config.get("name").replace(" ", "_")
97 config["uri"] = overwrite_config.get("uri")
98 config["origin"] = overwrite_config.get("origin")
99 config["namespace"] = overwrite_config.get("namespace")
100 config["code"] = source
101 config["id"] = (
102 config.get("name").strip().replace(" ", "_") + "_" + str(uuid.uuid4())
103 )
104 config["runnable"] = config.get("runnable", True)
105 config["requirements"] = config.get("requirements") or []
107 for field in CONFIGURABLE_FIELDS:
108 obj = config.get(field)
109 if obj and isinstance(obj, dict) and not isinstance(obj, list):
110 if config.get("tag"):
111 config[field] = obj.get(config.get("tag"))
112 if not obj.get(config.get("tag")):
113 print(
114 "WARNING: "
115 + field
116 + " do not contain a tag named: "
117 + config.get("tag")
118 )
119 else:
120 raise Exception("You must use 'tags' with configurable fields.")
121 config["lang"] = config.get("lang") or "javascript"
122 return config
125def convert_config_to_rdf(plugin_config, source_url=None) -> dict:
126 """Convert imjoy plugin config to RDF format."""
127 rdf = {
128 "type": "application",
129 }
130 if source_url:
131 rdf["source"] = source_url
132 fields = [
133 "icon",
134 "name",
135 "version",
136 "api_version",
137 "description",
138 "license",
139 "requirements",
140 "dependencies",
141 "env",
142 "passive",
143 "services",
144 ]
145 for field in fields:
146 if field in plugin_config:
147 rdf[field] = plugin_config[field]
148 tags = plugin_config.get("labels", []) + plugin_config.get("flags", [])
149 if "bioengine" not in tags:
150 tags.append("bioengine")
151 rdf["tags"] = tags
153 # docs = plugin_config.get("docs")
154 # if isinstance(docs, dict):
155 # rdf["documentation"] = docs.get("content")
156 # elif isinstance(docs, str):
157 # rdf["documentation"] = docs
158 rdf["covers"] = plugin_config.get("cover")
159 # make sure we have a list
160 if not rdf["covers"]:
161 rdf["covers"] = []
162 elif not isinstance(rdf["covers"], list):
163 rdf["covers"] = [rdf["covers"]]
165 rdf["badges"] = plugin_config.get("badge")
166 if not rdf["badges"]:
167 rdf["badges"] = []
168 elif not isinstance(rdf["badges"], list):
169 rdf["badges"] = [rdf["badges"]]
171 authors = plugin_config.get("author")
172 if authors:
173 if isinstance(authors, str):
174 authors = {"name": authors}
175 if not isinstance(authors, list):
176 authors = [authors]
177 rdf["authors"] = authors
179 return rdf
182def get_plugin_as_rdf(source_url: str) -> Dict[Any, Any]:
183 """Get imjoy plugin config in RDF format."""
184 req = httpx.get(source_url, timeout=5)
185 source = req.text
186 plugin_config = parse_imjoy_plugin(source)
187 rdf = convert_config_to_rdf(plugin_config, source_url)
188 return rdf
191def enrich_partial_rdf_with_imjoy_plugin(
192 partial_rdf: Dict[str, Any],
193 root: Union[HttpUrl, DirectoryPath],
194 resolve_rdf_source: Callable[
195 [Union[HttpUrl, FilePath, str]],
196 Tuple[Dict[str, Any], str, Union[HttpUrl, DirectoryPath]],
197 ],
198) -> Dict[str, Any]:
199 """
200 a (partial) rdf may have 'rdf_source' or 'source' which resolve to rdf data that may be overwritten.
201 """
203 enriched_rdf: Dict[str, Any] = {}
204 if "rdf_source" in partial_rdf:
205 given_rdf_src = partial_rdf["rdf_source"]
206 if isinstance(given_rdf_src, str) and given_rdf_src.split("?")[0].endswith(
207 ".imjoy.html"
208 ):
209 # given_rdf_src is an imjoy plugin
210 rdf_source = dict(get_plugin_as_rdf(given_rdf_src))
211 else:
212 # given_rdf_src is an actual rdf
213 if isinstance(given_rdf_src, dict):
214 rdf_source: Dict[str, Any] = given_rdf_src
215 else:
216 try:
217 rdf_source, _, rdf_source_root = resolve_rdf_source(given_rdf_src)
218 except Exception as e:
219 try:
220 rdf_source, _, rdf_source_root = resolve_rdf_source(
221 root / given_rdf_src
222 if isinstance(root, Path)
223 else urljoin(str(root), given_rdf_src)
224 )
225 except Exception as ee:
226 rdf_source = {}
227 warnings.warn(
228 f"Failed to resolve `rdf_source`: 1. {e}\n2. {ee}"
229 )
230 else:
231 rdf_source["root_path"] = (
232 rdf_source_root # enables remote source content to be resolved
233 )
234 else:
235 rdf_source["root_path"] = (
236 rdf_source_root # enables remote source content to be resolved
237 )
239 assert isinstance(rdf_source, dict)
240 enriched_rdf.update(rdf_source)
242 if "source" in partial_rdf:
243 if partial_rdf["source"].split("?")[0].endswith(".imjoy.html"):
244 rdf_from_source = get_plugin_as_rdf(partial_rdf["source"])
245 enriched_rdf.update(rdf_from_source)
247 enriched_rdf.update(
248 partial_rdf
249 ) # initial partial rdf overwrites fields from rdf_source or source
250 return enriched_rdf