Coverage for nexusLIMS/exporters/destinations/labarchives.py: 100%

1"""LabArchives export destination plugin.

3Exports NexusLIMS XML records to a LabArchives electronic lab notebook by

4creating a page per session with an HTML summary entry and the full XML record

5attached as a file.

6"""

8from __future__ import annotations

10import base64

11import logging

12import re

13import traceback

14from collections import Counter

15from pathlib import Path

17from nexusLIMS.config import settings

18from nexusLIMS.exporters.base import ExportContext, ExportResult

19from nexusLIMS.utils.labarchives import (

20 LabArchivesAuthenticationError,

21 LabArchivesClient,

22 LabArchivesError,

23 get_labarchives_client,

24)

26_logger = logging.getLogger(__name__)

29class LabArchivesDestination:

30 """LabArchives export destination plugin.

32 Uploads NexusLIMS XML records to a LabArchives instance. For each session,

33 creates a page under ``NexusLIMS Records/{instrument}/`` in the configured

34 notebook (or the user's Inbox if no notebook is configured) with:

36 - An HTML-formatted session summary as a text entry

37 - The full XML record attached as a file

39 Attributes

40 ----------

41 name : str

42 Destination identifier: "labarchives"

43 priority : int

44 Export priority: 90 (runs after CDCS at 100, allowing access to CDCS URLs)

45 """

47 name = "labarchives"

48 priority = 90

50 @property

51 def enabled(self) -> bool:

52 """Check if LabArchives is configured and enabled.

54 Returns

55 -------

56 bool

57 True if ``NX_LABARCHIVES_ACCESS_KEY_ID``,

58 ``NX_LABARCHIVES_ACCESS_PASSWORD``, ``NX_LABARCHIVES_USER_ID``,

59 and ``NX_LABARCHIVES_URL`` are all configured.

60 """

61 return all(

62 [

63 settings.NX_LABARCHIVES_ACCESS_KEY_ID,

64 settings.NX_LABARCHIVES_ACCESS_PASSWORD,

65 settings.NX_LABARCHIVES_USER_ID,

66 settings.NX_LABARCHIVES_URL,

67 ]

68 )

70 def validate_config(self) -> tuple[bool, str | None]:

71 """Validate LabArchives configuration.

73 Checks that required fields are present and tests authentication by

74 making a lightweight API call.

76 Returns

77 -------

78 tuple[bool, str | None]

79 ``(is_valid, error_message)`` — ``error_message`` is ``None`` on success.

80 """

81 for attr, label in [

82 ("NX_LABARCHIVES_ACCESS_KEY_ID", "NX_LABARCHIVES_ACCESS_KEY_ID"),

83 ("NX_LABARCHIVES_ACCESS_PASSWORD", "NX_LABARCHIVES_ACCESS_PASSWORD"),

84 ("NX_LABARCHIVES_USER_ID", "NX_LABARCHIVES_USER_ID"),

85 ("NX_LABARCHIVES_URL", "NX_LABARCHIVES_URL"),

86 ]:

87 if not getattr(settings, attr, None):

88 return False, f"{label} not configured"

90 # Test authentication by making a live API call

91 try: # pragma: no cover

92 client = get_labarchives_client()

93 nbid = settings.NX_LABARCHIVES_NOTEBOOK_ID

94 if nbid:

95 # Verify notebook is accessible

96 client.get_tree_level(nbid, "0")

97 else:

98 # No notebook configured — just verify credentials with a minimal call

99 client.get_tree_level("0", "0")

100 except LabArchivesAuthenticationError as e: # pragma: no cover

101 return False, f"LabArchives authentication failed: {e}"

102 except LabArchivesError as e: # pragma: no cover

103 # Non-auth errors (e.g. notebook not found) still mean config is usable

104 _logger.debug("LabArchives validate_config non-fatal error: %s", e)

105 except Exception as e: # pragma: no cover

106 return False, f"LabArchives configuration error: {e}"

107

108 return True, None # pragma: no cover

109

110 def export(self, context: ExportContext) -> ExportResult:

111 """Export record to LabArchives.

112

113 Creates a notebook page with an HTML summary entry and attaches the

114 XML record file. Never raises exceptions — all errors are caught and

115 returned as :class:`~nexusLIMS.exporters.base.ExportResult` with

116 ``success=False``.

117

118 Parameters

119 ----------

120 context : ExportContext

121 Export context with file path, session metadata, and results from

122 higher-priority destinations (e.g., CDCS).

123

124 Returns

125 -------

126 ExportResult

127 Result of the export attempt.

128 """

129 try:

130 client = get_labarchives_client()

131

132 # Find or create target page

133 nbid, page_tree_id = self._find_or_create_page(client, context)

134

135 # Collect preview images from activities

136 previews = self._collect_previews(context)

137

138 # Build and upload HTML summary entry

139 html_summary = self._build_html_summary(context, previews=previews)

140 entry_id = client.add_entry(nbid, page_tree_id, html_summary)

141 _logger.info(

142 "Created LabArchives entry %s on page %s", entry_id, page_tree_id

143 )

144

145 # Upload XML file as attachment

146 xml_bytes = context.xml_file_path.read_bytes()

147 filename = context.xml_file_path.name

148 client.add_attachment(

149 nbid,

150 page_tree_id,

151 filename,

152 xml_bytes,

153 caption="NexusLIMS XML record",

154 )

155 _logger.info("Uploaded XML attachment %s to LabArchives", filename)

156

157 # Build a best-effort entry URL

158 entry_url = _build_entry_url(

159 base_url=str(settings.NX_LABARCHIVES_URL),

160 nbid=nbid,

161 page_tree_id=page_tree_id,

162 )

163

164 return ExportResult(

165 success=True,

166 destination_name=self.name,

167 record_id=entry_id,

168 record_url=entry_url,

169 )

170

171 except Exception:

172 _logger.exception(

173 "Failed to export to LabArchives: %s",

174 context.xml_file_path.name,

175 )

176 return ExportResult(

177 success=False,

178 destination_name=self.name,

179 error_message=traceback.format_exc(),

180 )

181

182 # ------------------------------------------------------------------ #

183 # Private helpers #

184 # ------------------------------------------------------------------ #

185

186 def _find_or_create_page(

187 self,

188 client: LabArchivesClient,

189 context: ExportContext,

190 ) -> tuple[str, str]:

191 """Find or create the target notebook page for this session.

192

193 When ``NX_LABARCHIVES_NOTEBOOK_ID`` is configured, navigates to (or

194 creates) the path ``NexusLIMS Records/{instrument_pid}/`` and creates

195 a new page named ``{YYYY-MM-DD} — {session_identifier}``.

196

197 When no notebook ID is configured, returns ``("0", "0")`` which causes

198 the API calls to target the user's Inbox.

199

200 Parameters

201 ----------

202 client : LabArchivesClient

203 Authenticated API client

204 context : ExportContext

205 Export context with session metadata

206

207 Returns

208 -------

209 tuple[str, str]

210 ``(nbid, page_tree_id)``

211 """

212 nbid = settings.NX_LABARCHIVES_NOTEBOOK_ID

213 if not nbid:

214 # No notebook configured — upload to Inbox

215 return ("0", "0")

216

217 # Find or create "NexusLIMS Records" folder at root

218 root_nodes = client.get_tree_level(nbid, "0")

219 nexuslims_folder_id = _find_node_by_text(root_nodes, "NexusLIMS Records")

220 if nexuslims_folder_id is None:

221 nexuslims_folder_id = client.insert_folder(nbid, "0", "NexusLIMS Records")

222 _logger.info("Created 'NexusLIMS Records' folder in notebook %s", nbid)

223

224 # Find or create instrument sub-folder

225 instrument_nodes = client.get_tree_level(nbid, nexuslims_folder_id)

226 instrument_folder_id = _find_node_by_text(

227 instrument_nodes, context.instrument_pid

228 )

229 if instrument_folder_id is None:

230 instrument_folder_id = client.insert_folder(

231 nbid, nexuslims_folder_id, context.instrument_pid

232 )

233 _logger.info(

234 "Created instrument folder '%s' in LabArchives",

235 context.instrument_pid,

236 )

237

238 # Create a new page for this session

239 page_name = f"{context.dt_from:%Y-%m-%d} \u2014 {context.session_identifier}"

240 page_tree_id = client.insert_page(nbid, instrument_folder_id, page_name)

241 _logger.info("Created page '%s' (tree_id=%s)", page_name, page_tree_id)

242

243 return (nbid, page_tree_id)

244

245 def _collect_previews(self, context: ExportContext) -> list[tuple[str, bytes]]:

246 """Collect (name, image_bytes) pairs from context activities.

247

248 Iterates over all activities and their preview paths, reading image

249 bytes for each available preview. Multi-signal files that appear more

250 than once in an activity's file list are labelled ``"(N of M)"``.

251

252 Parameters

253 ----------

254 context : ExportContext

255 Export context with activities

256

257 Returns

258 -------

259 list[tuple[str, bytes]]

260 List of ``(caption_name, image_bytes)`` pairs, one per available preview.

261 """

262 previews: list[tuple[str, bytes]] = []

263 for act in context.activities:

264 file_counts: Counter = Counter(act.files)

265 file_seen: Counter = Counter()

266 for fname, preview_path in zip(act.files, act.previews):

267 file_seen[fname] += 1

268 if not preview_path or not Path(preview_path).exists():

269 continue

270 name = Path(fname).name

271 if file_counts[fname] > 1:

272 name = f"{name} ({file_seen[fname]} of {file_counts[fname]})"

273 try:

274 previews.append((name, Path(preview_path).read_bytes()))

275 except Exception:

276 _logger.debug("Could not read preview: %s", preview_path)

277 return previews

278

279 def _build_html_summary(

280 self,

281 context: ExportContext,

282 previews: list[tuple[str, bytes]] | None = None,

283 ) -> str:

284 """Build HTML summary content for the notebook entry.

285

286 Parameters

287 ----------

288 context : ExportContext

289 Export context with session metadata

290 previews : list[tuple[str, bytes]] | None

291 Optional list of ``(caption, image_bytes)`` pairs to embed as a

292 preview gallery (4 images per row, base64 data URIs).

293

294 Returns

295 -------

296 str

297 HTML-formatted session summary

298 """

299 lines = [

300 "<h1>NexusLIMS Microscopy Session</h1>",

301 "<h2>Session Details</h2>",

302 "<ul>",

303 f"<li><strong>Session ID</strong>: {context.session_identifier}</li>",

304 f"<li><strong>Instrument</strong>: {context.instrument_pid}</li>",

305 ]

306

307 if context.user:

308 lines.append(f"<li><strong>User</strong>: {context.user}</li>")

309

310 lines.extend(

311 [

312 f"<li><strong>Start</strong>: {context.dt_from.isoformat()}</li>",

313 f"<li><strong>End</strong>: {context.dt_to.isoformat()}</li>",

314 "</ul>",

315 ]

316 )

317

318 # Add CDCS link if available

319 cdcs_result = context.get_result("cdcs")

320 if cdcs_result and cdcs_result.success and cdcs_result.record_url:

321 lines.extend(

322 [

323 "<h2>Related Records</h2>",

324 "<ul>",

325 f'<li><a href="{cdcs_result.record_url}">View in CDCS</a></li>',

326 "</ul>",

327 ]

328 )

329

330 if previews:

331 lines.append("<h2>Preview Images</h2>")

332 lines.append('<table style="border-collapse:collapse;">')

333 for i, (name, img_bytes) in enumerate(previews):

334 if i % 4 == 0:

335 if i > 0:

336 lines.append("</tr>")

337 lines.append("<tr>")

338 b64 = base64.b64encode(img_bytes).decode()

339 lines.append(

340 f'<td style="padding:8px;text-align:center;vertical-align:top;">'

341 f'<img src="data:image/png;base64,{b64}" alt="{name}" '

342 f'style="max-width:250px;max-height:250px;border:1px solid #ccc;">'

343 f"<br><small>{name}</small></td>"

344 )

345 lines.append("</tr></table>")

346

347 lines.append("<h2>Files</h2>")

348 lines.append("<p>The following data files are referenced in this record:</p>")

349

350 # Collect unique file locations across all activities, preserving order.

351 # Strip NX_INSTRUMENT_DATA_PATH prefix so paths match the XML <location>

352 # element (relative to the instrument storage root).

353 instr_root = str(settings.NX_INSTRUMENT_DATA_PATH)

354 seen: set[str] = set()

355 file_names: list[str] = []

356 for act in context.activities:

357 for fname in act.files:

358 if fname not in seen:

359 seen.add(fname)

360 rel = fname.replace(instr_root, "")

361 file_names.append(rel)

362

363 if file_names:

364 lines.append("<ul>")

365 for name in file_names:

366 lines.append(f"<li>{name}</li>")

367 lines.append("</ul>")

368

369 return "\n".join(lines)

370

371

372# ------------------------------------------------------------------ #

373# Module-level helpers #

374# ------------------------------------------------------------------ #

375

376

377def _find_node_by_text(

378 nodes: list[dict],

379 display_text: str,

380) -> str | None:

381 """Return tree_id of the first node matching display_text, or None."""

382 for node in nodes:

383 if node.get("display_text") == display_text:

384 return node["tree_id"]

385 return None

386

387

388_LA_API_HOST = "api.labarchives.com"

389_LA_WEB_HOST = "mynotebook.labarchives.com"

390

391

392def _build_entry_url(base_url: str, nbid: str, page_tree_id: str) -> str:

393 """Build a best-effort URL to the LabArchives notebook page.

394

395 Parameters

396 ----------

397 base_url : str

398 API base URL (e.g. ``"https://api.labarchives.com/api"``). For the

399 cloud LabArchives service, ``api.labarchives.com`` is replaced with

400 ``mynotebook.labarchives.com`` so the link opens in the web interface.

401 For other hosts, the ``/api`` path suffix is stripped.

402 nbid : str

403 Notebook ID

404 page_tree_id : str

405 Tree ID of the page

406

407 Returns

408 -------

409 str

410 URL pointing to the page in the LabArchives web interface (best effort)

411 """

412 # For cloud LabArchives: swap API host → web host and drop the /api path.

413 # For self-hosted instances: just strip the /api suffix.

414 base = re.sub(r"/api/?$", "", base_url.rstrip("/"))

415 base = base.replace(_LA_API_HOST, _LA_WEB_HOST)

416 if nbid and nbid != "0":

417 return f"{base}/#/{nbid}/{page_tree_id}"

418 return base