wsi_annotation_tests/
probe.rs

1use std::ffi::OsString;
2use std::fmt;
3use std::path::Path;
4use std::time::Duration;
5
6use serde::Serialize;
7use serde_json::Value;
8
9use crate::process::{CommandSpec, ProcessError, ProcessOutput, run};
10use crate::schema::{validate_conversion_report, validate_probe_report};
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum PayloadMode {
14    Full,
15    Digest,
16}
17
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19enum ProbeOperation {
20    Inspect,
21    Roundtrip,
22    ConvertGeoJson,
23    ConvertRaster,
24}
25
26impl ProbeOperation {
27    const fn label(self) -> &'static str {
28        match self {
29            Self::Inspect => "inspect",
30            Self::Roundtrip => "roundtrip",
31            Self::ConvertGeoJson => "convert-geojson",
32            Self::ConvertRaster => "convert-raster",
33        }
34    }
35}
36
37impl PayloadMode {
38    const fn label(self) -> &'static str {
39        match self {
40            Self::Full => "full",
41            Self::Digest => "digest",
42        }
43    }
44}
45
46#[derive(Debug, Clone, Copy, PartialEq, Eq)]
47pub enum GeoJsonCoordinateSpace {
48    Level0Pixels,
49    SourcePixels,
50    SlideMillimeters,
51}
52
53impl GeoJsonCoordinateSpace {
54    const fn label(self) -> &'static str {
55        match self {
56            Self::Level0Pixels => "level0-pixels",
57            Self::SourcePixels => "source-pixels",
58            Self::SlideMillimeters => "slide-mm",
59        }
60    }
61}
62
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
64pub enum GeoJsonTarget {
65    Ann,
66    Seg,
67    Sr,
68}
69
70impl GeoJsonTarget {
71    const fn label(self) -> &'static str {
72        match self {
73            Self::Ann => "ann",
74            Self::Seg => "seg",
75            Self::Sr => "sr",
76        }
77    }
78}
79
80#[derive(Debug, Clone, PartialEq, Eq)]
81pub enum RasterChannels {
82    Auto,
83    One(String),
84    All,
85}
86
87#[derive(Debug, Clone)]
88pub struct ProbeObservation {
89    pub command: Vec<String>,
90    pub report: Value,
91    pub stderr: String,
92    pub elapsed_ms: f64,
93    pub peak_rss_bytes: u64,
94    pub rss_sampled: bool,
95    pub sample_interval_ms: Option<u64>,
96    pub stdout_total_bytes: u64,
97    pub stderr_total_bytes: u64,
98    pub stderr_truncated: bool,
99}
100
101#[derive(Debug, Clone, Copy, Serialize)]
102pub(crate) struct ProbeMeasurements {
103    pub runtime_ms: Option<f64>,
104    pub peak_rss_bytes: Option<u64>,
105    pub peak_tracked_heap_bytes: Option<u64>,
106}
107
108pub(crate) fn tracked_heap_bytes(report: &Value) -> Option<u64> {
109    report
110        .pointer("/runtime/peak_tracked_heap_bytes")
111        .or_else(|| report.get("peak_tracked_heap_bytes"))
112        .and_then(Value::as_u64)
113}
114
115impl ProbeObservation {
116    pub(crate) fn measurements(&self) -> ProbeMeasurements {
117        ProbeMeasurements {
118            runtime_ms: Some(self.elapsed_ms),
119            peak_rss_bytes: self.rss_sampled.then_some(self.peak_rss_bytes),
120            peak_tracked_heap_bytes: tracked_heap_bytes(&self.report),
121        }
122    }
123}
124
125#[derive(Debug, Clone, Serialize)]
126pub struct ProbeError {
127    #[serde(flatten)]
128    details: Box<ProbeErrorDetails>,
129    pub report: Option<Value>,
130    pub stderr: String,
131    pub returncode: Option<i32>,
132    pub peak_rss_bytes: Option<u64>,
133    pub process: Option<Box<ProbeProcessEvidence>>,
134    pub command: Option<Box<Vec<String>>>,
135}
136
137#[derive(Debug, Clone, Serialize)]
138struct ProbeErrorDetails {
139    message: String,
140    elapsed_ms: Option<f64>,
141}
142
143#[derive(Debug, Clone, Serialize)]
144pub struct ProbeProcessEvidence {
145    pub stdout_truncated: bool,
146    pub stderr_truncated: bool,
147    pub stdout_total_bytes: u64,
148    pub stderr_total_bytes: u64,
149    pub sample_interval_ms: Option<u64>,
150}
151
152impl fmt::Display for ProbeError {
153    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
154        formatter.write_str(&self.details.message)
155    }
156}
157
158impl std::error::Error for ProbeError {}
159
160impl ProbeError {
161    pub(crate) fn measurements(&self) -> ProbeMeasurements {
162        ProbeMeasurements {
163            runtime_ms: self.elapsed_ms(),
164            peak_rss_bytes: self.peak_rss_bytes,
165            peak_tracked_heap_bytes: self.report.as_ref().and_then(tracked_heap_bytes),
166        }
167    }
168
169    /// Measured subprocess time, absent when the process could not be launched.
170    #[must_use]
171    pub(crate) const fn elapsed_ms(&self) -> Option<f64> {
172        self.details.elapsed_ms
173    }
174
175    fn with_command(mut self, command: &[String]) -> Self {
176        self.command = Some(Box::new(command.to_vec()));
177        self
178    }
179}
180
181#[derive(Debug, Clone)]
182pub struct ViewerProbe {
183    executable: CommandSpec,
184    timeout: Duration,
185}
186
187impl ViewerProbe {
188    /// Construct a process-only viewer adapter.
189    ///
190    /// # Errors
191    ///
192    /// Returns an error for an empty command or a zero timeout.
193    pub fn new(executable: Vec<String>, timeout: Option<Duration>) -> Result<Self, String> {
194        let executable = CommandSpec::from_strings(executable, "annotation_probe")?;
195        Self::from_spec(executable, timeout)
196    }
197
198    /// Construct a viewer adapter without converting the program or fixed arguments to UTF-8.
199    ///
200    /// # Errors
201    ///
202    /// Returns an error for an empty program or a zero timeout.
203    pub fn from_program(
204        program: impl Into<OsString>,
205        arguments: Vec<OsString>,
206        timeout: Option<Duration>,
207    ) -> Result<Self, String> {
208        let executable = CommandSpec::new(program.into(), arguments)?;
209        Self::from_spec(executable, timeout)
210    }
211
212    fn from_spec(executable: CommandSpec, timeout: Option<Duration>) -> Result<Self, String> {
213        let timeout = timeout.unwrap_or(Duration::from_mins(10));
214        if timeout.is_zero() {
215            return Err("timeout must be positive".to_owned());
216        }
217        Ok(Self {
218            executable,
219            timeout,
220        })
221    }
222
223    #[must_use]
224    pub fn program(&self) -> &Path {
225        Path::new(self.executable.program())
226    }
227
228    /// Inspect one ANN or SEG object through the public probe contract.
229    ///
230    /// # Errors
231    ///
232    /// Returns a structured error for process, timeout, JSON, schema, or probe failures.
233    pub fn inspect(
234        &self,
235        source: &Path,
236        annotation: &Path,
237        canonical_source: Option<&Path>,
238        payload: PayloadMode,
239    ) -> Result<ProbeObservation, ProbeError> {
240        let mut command = self.executable.clone();
241        command.extend(["inspect", "--source"]);
242        command.push(source.as_os_str());
243        if let Some(canonical_source) = canonical_source {
244            command.push("--canonical-source");
245            command.push(canonical_source.as_os_str());
246        }
247        command.extend([
248            OsString::from("--payload"),
249            OsString::from(payload.label()),
250            annotation.as_os_str().to_owned(),
251        ]);
252        self.execute(&command, ProbeOperation::Inspect, validate_probe_report)
253    }
254
255    /// Rewrite one ANN or SEG object through the public probe contract.
256    ///
257    /// # Errors
258    ///
259    /// Returns a structured error for process, timeout, JSON, schema, or probe failures.
260    pub fn roundtrip(
261        &self,
262        source: &Path,
263        annotation: &Path,
264        output: &Path,
265        canonical_source: Option<&Path>,
266        payload: PayloadMode,
267        allow_lossy: bool,
268    ) -> Result<ProbeObservation, ProbeError> {
269        let mut command = self.executable.clone();
270        command.extend(["roundtrip", "--source"]);
271        command.push(source.as_os_str());
272        command.push("--output");
273        command.push(output.as_os_str());
274        if let Some(canonical_source) = canonical_source {
275            command.push("--canonical-source");
276            command.push(canonical_source.as_os_str());
277        }
278        if allow_lossy {
279            command.push("--allow-lossy");
280        }
281        command.extend([
282            OsString::from("--payload"),
283            OsString::from(payload.label()),
284            annotation.as_os_str().to_owned(),
285        ]);
286        self.execute(&command, ProbeOperation::Roundtrip, validate_probe_report)
287    }
288
289    /// Convert profiled `GeoJSON` into a deterministic multi-object bundle.
290    ///
291    /// # Errors
292    ///
293    /// Returns a structured error for process, schema, or conversion failures.
294    #[allow(clippy::too_many_arguments)]
295    pub fn convert_geojson_bundle(
296        &self,
297        source: &Path,
298        canonical_source: Option<&Path>,
299        mapping: &Path,
300        coordinate_space: GeoJsonCoordinateSpace,
301        targets: &[GeoJsonTarget],
302        output_directory: &Path,
303        geojson: &Path,
304        allow_lossy: bool,
305    ) -> Result<ProbeObservation, ProbeError> {
306        if targets.is_empty() {
307            return Err(configuration_error(
308                "GeoJSON conversion requires at least one target",
309            ));
310        }
311        let mut command = self.executable.clone();
312        command.extend([
313            OsString::from("convert-geojson"),
314            OsString::from("--source"),
315            source.as_os_str().to_owned(),
316        ]);
317        if let Some(canonical_source) = canonical_source {
318            command.extend([
319                OsString::from("--canonical-source"),
320                canonical_source.as_os_str().to_owned(),
321            ]);
322        }
323        command.extend([
324            OsString::from("--mapping"),
325            mapping.as_os_str().to_owned(),
326            OsString::from("--coordinate-space"),
327            OsString::from(coordinate_space.label()),
328        ]);
329        for target in targets {
330            command.extend(["--target", target.label()]);
331        }
332        if allow_lossy {
333            command.push("--allow-lossy");
334        }
335        command.extend([
336            OsString::from("--output-dir"),
337            output_directory.as_os_str().to_owned(),
338            geojson.as_os_str().to_owned(),
339        ]);
340        self.execute(
341            &command,
342            ProbeOperation::ConvertGeoJson,
343            validate_conversion_report,
344        )
345    }
346
347    /// Convert a profiled raster into a Parametric Map bundle.
348    ///
349    /// # Errors
350    ///
351    /// Returns a structured error for process, schema, or conversion failures.
352    #[allow(clippy::too_many_arguments)]
353    pub fn convert_raster_bundle(
354        &self,
355        source: &Path,
356        canonical_source: Option<&Path>,
357        profile: &Path,
358        channels: RasterChannels,
359        output_directory: &Path,
360        maximum_instance_bytes: Option<u64>,
361        raster: &Path,
362    ) -> Result<ProbeObservation, ProbeError> {
363        let mut command = self.executable.clone();
364        command.extend([
365            OsString::from("convert-raster"),
366            OsString::from("--source"),
367            source.as_os_str().to_owned(),
368        ]);
369        if let Some(canonical_source) = canonical_source {
370            command.extend([
371                OsString::from("--canonical-source"),
372                canonical_source.as_os_str().to_owned(),
373            ]);
374        }
375        command.extend([OsString::from("--profile"), profile.as_os_str().to_owned()]);
376        match channels {
377            RasterChannels::Auto => {}
378            RasterChannels::One(value) => {
379                command.extend([OsString::from("--channel"), OsString::from(value)]);
380            }
381            RasterChannels::All => command.push("--all-channels"),
382        }
383        if let Some(maximum) = maximum_instance_bytes {
384            command.extend([
385                OsString::from("--max-instance-bytes"),
386                OsString::from(maximum.to_string()),
387            ]);
388        }
389        command.extend([
390            OsString::from("--output-dir"),
391            output_directory.as_os_str().to_owned(),
392            raster.as_os_str().to_owned(),
393        ]);
394        self.execute(
395            &command,
396            ProbeOperation::ConvertRaster,
397            validate_conversion_report,
398        )
399    }
400
401    fn execute(
402        &self,
403        command: &CommandSpec,
404        expected_operation: ProbeOperation,
405        validate: fn(&Value) -> Result<(), String>,
406    ) -> Result<ProbeObservation, ProbeError> {
407        let display_command = command.display();
408        let output = run(&command.process_spec(self.timeout))
409            .map_err(|error| process_error(error).with_command(&display_command))?;
410        let stderr = String::from_utf8_lossy(&output.stderr.bytes).into_owned();
411        if output.stdout.truncated {
412            return Err(observed_error(
413                format!(
414                    "annotation_probe report exceeded the {} byte stdout limit ({} bytes observed)",
415                    output.stdout.bytes.len(),
416                    output.stdout.total_bytes
417                ),
418                None,
419                stderr,
420                &output,
421                &display_command,
422            ));
423        }
424        let report: Value = serde_json::from_slice(&output.stdout.bytes).map_err(|error| {
425            observed_error(
426                format!("annotation_probe emitted invalid JSON: {error}"),
427                None,
428                stderr.clone(),
429                &output,
430                &display_command,
431            )
432        })?;
433        validate(&report).map_err(|message| {
434            observed_error(
435                message,
436                Some(report.clone()),
437                stderr.clone(),
438                &output,
439                &display_command,
440            )
441        })?;
442        let operation = report.get("operation").and_then(Value::as_str);
443        if operation != Some(expected_operation.label()) {
444            return Err(observed_error(
445                format!(
446                    "annotation_probe reported operation {} for requested {}",
447                    operation.unwrap_or("<missing>"),
448                    expected_operation.label()
449                ),
450                Some(report),
451                stderr,
452                &output,
453                &display_command,
454            ));
455        }
456        if !output.status.success() || report.get("status").and_then(Value::as_str) != Some("ok") {
457            let message = report
458                .pointer("/error/message")
459                .and_then(Value::as_str)
460                .map_or_else(
461                    || format!("annotation_probe exited with status {}", output.status),
462                    str::to_owned,
463                );
464            return Err(observed_error(
465                message,
466                Some(report),
467                stderr,
468                &output,
469                &display_command,
470            ));
471        }
472        Ok(ProbeObservation {
473            command: display_command,
474            report,
475            stderr,
476            elapsed_ms: output.elapsed.as_secs_f64() * 1000.0,
477            peak_rss_bytes: output.peak_rss_bytes,
478            rss_sampled: output.rss_sampled,
479            sample_interval_ms: output
480                .sample_interval
481                .and_then(|interval| u64::try_from(interval.as_millis()).ok()),
482            stdout_total_bytes: output.stdout.total_bytes,
483            stderr_total_bytes: output.stderr.total_bytes,
484            stderr_truncated: output.stderr.truncated,
485        })
486    }
487}
488
489fn configuration_error(message: &str) -> ProbeError {
490    ProbeError {
491        details: Box::new(ProbeErrorDetails {
492            message: message.to_owned(),
493            elapsed_ms: None,
494        }),
495        report: None,
496        stderr: String::new(),
497        returncode: None,
498        peak_rss_bytes: None,
499        process: None,
500        command: None,
501    }
502}
503
504fn observed_error(
505    message: String,
506    report: Option<Value>,
507    stderr: String,
508    output: &ProcessOutput,
509    command: &[String],
510) -> ProbeError {
511    ProbeError {
512        details: Box::new(ProbeErrorDetails {
513            message,
514            elapsed_ms: Some(output.elapsed.as_secs_f64() * 1000.0),
515        }),
516        report,
517        stderr,
518        returncode: output.status.code(),
519        peak_rss_bytes: output.rss_sampled.then_some(output.peak_rss_bytes),
520        process: Some(Box::new(ProbeProcessEvidence {
521            stdout_truncated: output.stdout.truncated,
522            stderr_truncated: output.stderr.truncated,
523            stdout_total_bytes: output.stdout.total_bytes,
524            stderr_total_bytes: output.stderr.total_bytes,
525            sample_interval_ms: output
526                .sample_interval
527                .and_then(|interval| u64::try_from(interval.as_millis()).ok()),
528        })),
529        command: Some(Box::new(command.to_vec())),
530    }
531}
532
533fn process_error(error: ProcessError) -> ProbeError {
534    match error {
535        ProcessError::TimedOut { timeout, output } => ProbeError {
536            details: Box::new(ProbeErrorDetails {
537                elapsed_ms: Some(output.elapsed.as_secs_f64() * 1000.0),
538                message: format!(
539                    "annotation_probe timed out after {} seconds",
540                    timeout.as_secs_f64()
541                ),
542            }),
543            report: (!output.stdout.truncated)
544                .then(|| serde_json::from_slice(&output.stdout.bytes).ok())
545                .flatten(),
546            stderr: String::from_utf8_lossy(&output.stderr.bytes).into_owned(),
547            returncode: None,
548            peak_rss_bytes: output.rss_sampled.then_some(output.peak_rss_bytes),
549            process: Some(Box::new(ProbeProcessEvidence {
550                stdout_truncated: output.stdout.truncated,
551                stderr_truncated: output.stderr.truncated,
552                stdout_total_bytes: output.stdout.total_bytes,
553                stderr_total_bytes: output.stderr.total_bytes,
554                sample_interval_ms: output
555                    .sample_interval
556                    .and_then(|interval| u64::try_from(interval.as_millis()).ok()),
557            })),
558            command: None,
559        },
560        error => ProbeError {
561            details: Box::new(ProbeErrorDetails {
562                elapsed_ms: None,
563                message: format!("annotation_probe execution failed: {error}"),
564            }),
565            report: None,
566            stderr: String::new(),
567            returncode: None,
568            peak_rss_bytes: None,
569            process: None,
570            command: None,
571        },
572    }
573}
wsi_annotation_tests/probe.rs

wsi_annotation_tests/
probe.rs
