Coverage for bioimageio/spec/partner_utils/imjoy/_plugin_parser.py: 0%
131 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-02-05 13:53 +0000
1# type: ignore
2"""ImJoy plugin parser module."""
3import copy
4import json
5import uuid
6import warnings
7from pathlib import Path
8from typing import Any, Callable, Dict, Tuple, Union
9from urllib.parse import urljoin
11import requests
12from lxml import etree
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 root = etree.HTML("<html>" + source + "</html>")
51 plugin_comp = dotdict()
52 for tag_type in tag_types:
53 elms = root.xpath(f".//{tag_type}")
54 values = []
55 for elm in elms:
56 values.append(
57 dotdict(
58 attrs=dotdict(elm.attrib),
59 content=elm.text,
60 )
61 )
62 plugin_comp[tag_type] = values
63 if plugin_comp.config[0].attrs.lang == "yaml":
64 config = yaml.load(plugin_comp.config[0].content)
65 elif plugin_comp.config[0].attrs.lang == "json":
66 config = json.loads(plugin_comp.config[0].content)
67 else:
68 raise Exception(
69 "Unsupported config language: " + plugin_comp.config[0].attrs.lang
70 )
72 overwrite_config = overwrite_config or {}
73 config["tag"] = overwrite_config.get("tag") or (
74 config.get("tags") and config.get("tags")[0]
75 )
76 config["hot_reloading"] = overwrite_config.get("hot_reloading")
77 config["scripts"] = []
78 # try to match the script with current tag
79 for elm in plugin_comp.script:
80 if elm.attrs.tag == config["tag"]:
81 config["script"] = elm.content
82 # exclude script with mismatched tag
83 if not elm.attrs.tag or elm.attrs.tag == config["tag"]:
84 config["scripts"].append(elm)
85 if not config.get("script") and len(plugin_comp.script) > 0:
86 config["script"] = plugin_comp.script[0].content
87 config["lang"] = plugin_comp.script[0].attrs.lang
88 config["links"] = plugin_comp.link or None
89 config["windows"] = plugin_comp.window or None
90 config["styles"] = plugin_comp.style or None
91 config["docs"] = plugin_comp.docs[0] if plugin_comp.docs else config.get("docs")
92 config["attachments"] = plugin_comp.attachment or None
94 config["_id"] = overwrite_config.get("_id") or config.get("name").replace(" ", "_")
95 config["uri"] = overwrite_config.get("uri")
96 config["origin"] = overwrite_config.get("origin")
97 config["namespace"] = overwrite_config.get("namespace")
98 config["code"] = source
99 config["id"] = (
100 config.get("name").strip().replace(" ", "_") + "_" + str(uuid.uuid4())
101 )
102 config["runnable"] = config.get("runnable", True)
103 config["requirements"] = config.get("requirements") or []
105 for field in CONFIGURABLE_FIELDS:
106 obj = config.get(field)
107 if obj and isinstance(obj, dict) and not isinstance(obj, list):
108 if config.get("tag"):
109 config[field] = obj.get(config.get("tag"))
110 if not obj.get(config.get("tag")):
111 print(
112 "WARNING: "
113 + field
114 + " do not contain a tag named: "
115 + config.get("tag")
116 )
117 else:
118 raise Exception("You must use 'tags' with configurable fields.")
119 config["lang"] = config.get("lang") or "javascript"
120 return config
123def convert_config_to_rdf(plugin_config, source_url=None) -> dict:
124 """Convert imjoy plugin config to RDF format."""
125 rdf = {
126 "type": "application",
127 }
128 if source_url:
129 rdf["source"] = source_url
130 fields = [
131 "icon",
132 "name",
133 "version",
134 "api_version",
135 "description",
136 "license",
137 "requirements",
138 "dependencies",
139 "env",
140 "passive",
141 "services",
142 ]
143 for field in fields:
144 if field in plugin_config:
145 rdf[field] = plugin_config[field]
146 tags = plugin_config.get("labels", []) + plugin_config.get("flags", [])
147 if "bioengine" not in tags:
148 tags.append("bioengine")
149 rdf["tags"] = tags
151 # docs = plugin_config.get("docs")
152 # if isinstance(docs, dict):
153 # rdf["documentation"] = docs.get("content")
154 # elif isinstance(docs, str):
155 # rdf["documentation"] = docs
156 rdf["covers"] = plugin_config.get("cover")
157 # make sure we have a list
158 if not rdf["covers"]:
159 rdf["covers"] = []
160 elif not isinstance(rdf["covers"], list):
161 rdf["covers"] = [rdf["covers"]]
163 rdf["badges"] = plugin_config.get("badge")
164 if not rdf["badges"]:
165 rdf["badges"] = []
166 elif not isinstance(rdf["badges"], list):
167 rdf["badges"] = [rdf["badges"]]
169 authors = plugin_config.get("author")
170 if authors:
171 if isinstance(authors, str):
172 authors = {"name": authors}
173 if not isinstance(authors, list):
174 authors = [authors]
175 rdf["authors"] = authors
177 return rdf
180def get_plugin_as_rdf(source_url: str) -> Dict[Any, Any]:
181 """Get imjoy plugin config in RDF format."""
182 req = requests.get(source_url, timeout=5)
183 source = req.text
184 plugin_config = parse_imjoy_plugin(source)
185 rdf = convert_config_to_rdf(plugin_config, source_url)
186 return rdf
189def enrich_partial_rdf_with_imjoy_plugin(
190 partial_rdf: Dict[str, Any],
191 root: Union[HttpUrl, DirectoryPath],
192 resolve_rdf_source: Callable[
193 [Union[HttpUrl, FilePath, str]],
194 Tuple[Dict[str, Any], str, Union[HttpUrl, DirectoryPath]],
195 ],
196) -> Dict[str, Any]:
197 """
198 a (partial) rdf may have 'rdf_source' or 'source' which resolve to rdf data that may be overwritten.
199 """
201 enriched_rdf: Dict[str, Any] = {}
202 if "rdf_source" in partial_rdf:
203 given_rdf_src = partial_rdf["rdf_source"]
204 if isinstance(given_rdf_src, str) and given_rdf_src.split("?")[0].endswith(
205 ".imjoy.html"
206 ):
207 # given_rdf_src is an imjoy plugin
208 rdf_source = dict(get_plugin_as_rdf(given_rdf_src))
209 else:
210 # given_rdf_src is an actual rdf
211 if isinstance(given_rdf_src, dict):
212 rdf_source: Dict[str, Any] = given_rdf_src
213 else:
214 try:
215 rdf_source, _, rdf_source_root = resolve_rdf_source(given_rdf_src)
216 except Exception as e:
217 try:
218 rdf_source, _, rdf_source_root = resolve_rdf_source(
219 root / given_rdf_src
220 if isinstance(root, Path)
221 else urljoin(str(root), given_rdf_src)
222 )
223 except Exception as ee:
224 rdf_source = {}
225 warnings.warn(
226 f"Failed to resolve `rdf_source`: 1. {e}\n2. {ee}"
227 )
228 else:
229 rdf_source["root_path"] = (
230 rdf_source_root # enables remote source content to be resolved
231 )
232 else:
233 rdf_source["root_path"] = (
234 rdf_source_root # enables remote source content to be resolved
235 )
237 assert isinstance(rdf_source, dict)
238 enriched_rdf.update(rdf_source)
240 if "source" in partial_rdf:
241 if partial_rdf["source"].split("?")[0].endswith(".imjoy.html"):
242 rdf_from_source = get_plugin_as_rdf(partial_rdf["source"])
243 enriched_rdf.update(rdf_from_source)
245 enriched_rdf.update(
246 partial_rdf
247 ) # initial partial rdf overwrites fields from rdf_source or source
248 return enriched_rdf