Skills › Scientific Agent Skills › latchbio-integration › inspect_latch_sdk.py
inspect_latch_sdk.py
292 lines · skills/latchbio-integration/scripts/inspect_latch_sdk.py at 5f2bc03
1#!/usr/bin/env python3
2"""Inspect the installed Latch SDK without authentication or network access."""
4from __future__ import annotations
6import argparse
7import importlib
8import inspect
9import json
10import platform
11import re
12import sys
13from importlib.metadata import PackageNotFoundError, version
14from typing import Any
17SYMBOL_GROUPS: dict[str, list[tuple[str, str]]] = {
18 "core": [
19 ("latch", "workflow"),
20 ("latch", "map_task"),
21 ("latch", "create_conditional_section"),
22 ("latch", "workflow_reference"),
23 ],
24 "tasks": [
25 ("latch.resources.tasks", "small_task"),
26 ("latch.resources.tasks", "medium_task"),
27 ("latch.resources.tasks", "large_task"),
28 ("latch.resources.tasks", "small_gpu_task"),
29 ("latch.resources.tasks", "large_gpu_task"),
30 ("latch.resources.tasks", "custom_task"),
31 ("latch.resources.tasks", "custom_memory_optimized_task"),
32 ("latch.resources.tasks", "v100_x1_task"),
33 ("latch.resources.tasks", "v100_x4_task"),
34 ("latch.resources.tasks", "v100_x8_task"),
35 ("latch.resources.tasks", "g6e_xlarge_task"),
36 ("latch.resources.tasks", "g6e_2xlarge_task"),
37 ("latch.resources.tasks", "g6e_4xlarge_task"),
38 ("latch.resources.tasks", "g6e_8xlarge_task"),
39 ("latch.resources.tasks", "g6e_12xlarge_task"),
40 ("latch.resources.tasks", "g6e_16xlarge_task"),
41 ("latch.resources.tasks", "g6e_24xlarge_task"),
42 ("latch.resources.tasks", "g6e_48xlarge_task"),
43 ],
44 "data": [
45 ("latch.ldata.path", "LPath"),
46 ("latch.types", "LatchFile"),
47 ("latch.types", "LatchDir"),
48 ("latch.types", "LatchOutputFile"),
49 ("latch.types", "LatchOutputDir"),
50 ("latch.types", "file_glob"),
51 ],
52 "metadata": [
53 ("latch.types.metadata", "LatchMetadata"),
54 ("latch.types.metadata", "LatchParameter"),
55 ("latch.resources.launch_plan", "LaunchPlan"),
56 ("latch.types.samplesheet_item", "SamplesheetItem"),
57 ],
58 "registry": [
59 ("latch.account", "Account"),
60 ("latch.registry.project", "Project"),
61 ("latch.registry.table", "Table"),
62 ("latch.registry.record", "Record"),
63 ],
64 "execution": [
65 ("latch_cli.services.launch.launch_v2", "launch"),
66 ("latch_cli.services.launch.launch_v2", "launch_from_launch_plan"),
67 ("latch_cli.services.launch.launch_v2", "Execution"),
68 ("latch_cli.services.launch.launch_v2", "CompletedExecution"),
69 ],
70 "verified": [
71 ("latch.verified", "rnaseq"),
72 ("latch.verified", "deseq2_wf"),
73 ("latch.verified", "gene_ontology_pathway_analysis"),
74 ("latch.verified", "mafft"),
75 ("latch.verified", "trim_galore"),
76 ],
77}
80METHODS: dict[str, tuple[str, str, list[str]]] = {
81 "LPath": (
82 "latch.ldata.path",
83 "LPath",
84 [
85 "exists",
86 "iterdir",
87 "mkdirp",
88 "copy_to",
89 "upload_from",
90 "download",
91 "rmr",
92 "fetch_metadata",
93 "modify_time",
94 "version_id",
95 ],
96 ),
97 "Table": (
98 "latch.registry.table",
99 "Table",
100 ["load", "list_records", "get_dataframe", "update"],
101 ),
102 "Execution": (
103 "latch_cli.services.launch.launch_v2",
104 "Execution",
105 ["poll", "wait", "abort"],
106 ),
107}
109OBJECT_REPR = re.compile(r"<(?P<name>[^<>]+) object at 0x[0-9a-fA-F]+>")
110ADDRESS_REPR = re.compile(r"0x[0-9a-fA-F]+")
113def normalize_repr(value: str) -> str:
114 """Remove process-specific memory addresses from introspection output."""
115 value = OBJECT_REPR.sub(lambda match: f"<{match.group('name')}>", value)
116 return ADDRESS_REPR.sub("0x<address>", value)
119def safe_signature(obj: Any) -> str | None:
120 try:
121 return normalize_repr(str(inspect.signature(obj)))
122 except (TypeError, ValueError):
123 return None
126def inspect_symbol(module_name: str, symbol_name: str) -> dict[str, Any]:
127 qualified_name = f"{module_name}.{symbol_name}"
128 try:
129 module = importlib.import_module(module_name)
130 except Exception as exc: # Import failures are the diagnostic output.
131 return {
132 "qualified_name": qualified_name,
133 "available": False,
134 "error": normalize_repr(f"{type(exc).__name__}: {exc}"),
135 }
137 if not hasattr(module, symbol_name):
138 return {
139 "qualified_name": qualified_name,
140 "available": False,
141 "error": "symbol not found",
142 }
144 obj = getattr(module, symbol_name)
145 result = {
146 "qualified_name": qualified_name,
147 "available": True,
148 "kind": type(obj).__name__,
149 "signature": safe_signature(obj),
150 }
152 python_interface = getattr(obj, "python_interface", None)
153 if python_interface is not None:
154 result["python_interface"] = {
155 "inputs": {
156 name: normalize_repr(inspect.formatannotation(annotation))
157 for name, annotation in python_interface.inputs.items()
158 },
159 "outputs": {
160 name: normalize_repr(inspect.formatannotation(annotation))
161 for name, annotation in python_interface.outputs.items()
162 },
163 }
165 return result
168def inspect_methods(
169 module_name: str,
170 class_name: str,
171 method_names: list[str],
172) -> dict[str, Any]:
173 try:
174 cls = getattr(importlib.import_module(module_name), class_name)
175 except Exception as exc:
176 return {"error": normalize_repr(f"{type(exc).__name__}: {exc}")}
178 result: dict[str, Any] = {}
179 for method_name in method_names:
180 method = getattr(cls, method_name, None)
181 result[method_name] = {
182 "available": method is not None,
183 "signature": safe_signature(method) if method is not None else None,
184 }
185 return result
188def build_report() -> dict[str, Any]:
189 try:
190 latch_version = version("latch")
191 except PackageNotFoundError:
192 latch_version = None
194 symbols = {
195 group: [
196 inspect_symbol(module_name, symbol_name)
197 for module_name, symbol_name in entries
198 ]
199 for group, entries in SYMBOL_GROUPS.items()
200 }
202 methods = {
203 label: inspect_methods(module_name, class_name, method_names)
204 for label, (module_name, class_name, method_names) in METHODS.items()
205 }
207 return {
208 "python": platform.python_version(),
209 "platform": platform.platform(),
210 "latch_version": latch_version,
211 "symbols": symbols,
212 "methods": methods,
213 }
216def print_text(report: dict[str, Any]) -> None:
217 print(f"Python: {report['python']}")
218 print(f"Platform: {report['platform']}")
219 print(f"Latch SDK: {report['latch_version'] or 'not installed'}")
221 for group, entries in report["symbols"].items():
222 print(f"\n[{group}]")
223 for entry in entries:
224 status = "ok" if entry["available"] else "missing"
225 line = f"- {status}: {entry['qualified_name']}"
226 if entry.get("signature"):
227 line += entry["signature"]
228 if entry.get("error"):
229 line += f" ({entry['error']})"
230 print(line)
231 if entry.get("python_interface"):
232 print(f" inputs: {entry['python_interface']['inputs']}")
233 print(f" outputs: {entry['python_interface']['outputs']}")
235 print("\n[methods]")
236 for class_name, methods in report["methods"].items():
237 print(f"- {class_name}")
238 if "error" in methods:
239 print(f" error: {methods['error']}")
240 continue
241 for method_name, details in methods.items():
242 status = "ok" if details["available"] else "missing"
243 signature = details["signature"] or ""
244 print(f" - {status}: {method_name}{signature}")
247def has_required_failures(report: dict[str, Any]) -> bool:
248 required = {
249 "latch.workflow",
250 "latch.resources.tasks.small_task",
251 "latch.resources.tasks.custom_task",
252 "latch.ldata.path.LPath",
253 "latch.registry.table.Table",
254 "latch_cli.services.launch.launch_v2.launch",
255 }
256 observed = {
257 entry["qualified_name"]: entry["available"]
258 for entries in report["symbols"].values()
259 for entry in entries
260 }
261 return any(not observed.get(name, False) for name in required)
264def main() -> int:
265 parser = argparse.ArgumentParser(description=__doc__)
266 parser.add_argument(
267 "--json",
268 action="store_true",
269 help="Emit machine-readable JSON.",
270 )
271 parser.add_argument(
272 "--strict",
273 action="store_true",
274 help="Exit non-zero when a core symbol is missing.",
275 )
276 args = parser.parse_args()
278 report = build_report()
279 if args.json:
280 print(json.dumps(report, indent=2, sort_keys=True))
281 else:
282 print_text(report)
284 if report["latch_version"] is None:
285 return 2
286 if args.strict and has_required_failures(report):
287 return 1
288 return 0
291if __name__ == "__main__":
292 sys.exit(main())