MedicalAI Observability Library API Reference

Table of Contents

• API Overview

• Image Format Support

  • Supported Image Formats by Modality

  • Universal Image Input Handlers

  • Modality-Specific Handlers

• Model Prediction Format Support

  • Supported Model Output Formats by Task Type

  • Universal Model Output Handler

  • Task-Specific Output Handlers

• Modality-Specific Metric Functions

  • CT-Specific Metrics

  • MRI-Specific Metrics

  • Ultrasound-Specific Metrics

  • X-ray-Specific Metrics

  • Mammography-Specific Metrics

  • Digital Pathology-Specific Metrics

  • PET/SPECT-Specific Metrics

  • OCT-Specific Metrics

• Data Distribution Metrics

  • Privacy-Preserving Distribution Tracking

• Model Performance Metrics

  • Classification Performance Metrics

  • Detection Performance Metrics

  • Segmentation Performance Metrics

  • Uncertainty Estimation

  • Cross-Task Consistency

• Clinical Value Metrics

• Privacy-Safe API Integration

• Integration Approaches for Development and Production

  • Development Environment Integration

  • Production Environment Integration

• Privacy and Security

  • Privacy-Preserving Metrics Collection

  • Secure Storage

  • Access Control

• Configuration Reference

  • YAML Configuration Format

  • Environment Variables

• Best Practices Guide

  • Integration Best Practices

  • Modality-Specific Recommendations

  • Task-Specific Recommendations

• Version History and Roadmap

  • Version History

  • Roadmap

API Overview

This document provides a comprehensive reference for the MedicalAI Observability Library, detailing all classes, methods, parameters, return values, and supported data formats for medical imaging AI observability across development and production environments.

This reference is specifically designed to be implementable within a third-party workflow :

• AI teams integrate their inference API into the thirdparty platform

• The observability module is integrated

• Provides observability dashboards to AI teams/companies

Image Format Support

Supported Image Formats by Modality

The library supports all major medical imaging formats across modalities. Below is a detailed reference of supported formats by modality and the specific conversion methods available.

Modality	Supported Formats	Primary Metadata Fields	Format Handler Functions
CT	DICOM, NIFTI, RAW, MHD	kVp, mAs, CTDI, reconstruction kernel, slice thickness	handle_ct_input(), ct_to_numpy(), extract_ct_metadata()
MRI	DICOM, NIFTI, ANALYZE	TR, TE, field strength, sequence name, flip angle	handle_mri_input(), mri_to_numpy(), extract_mri_metadata()
Ultrasound	DICOM, RAW, MP4, NRRD	transducer frequency, probe type, depth, gain	handle_us_input(), us_to_numpy(), extract_us_metadata()
X-Ray	DICOM, JPEG, PNG	exposure index, detector type, grid ratio, SID	handle_xray_input(), xray_to_numpy(), extract_xray_metadata()
Mammography	DICOM, JPEG	compression, view type, detector exposure, breast thickness	handle_mammo_input(), mammo_to_numpy(), extract_mammo_metadata()
Digital Pathology	TIFF, SVS, NDPI	magnification, stain type, slide ID, tissue type	handle_pathology_input(), pathology_to_numpy(), extract_pathology_metadata()
PET/SPECT	DICOM, NIFTI	radiopharmaceutical, activity, half-life, acquisition type	handle_pet_input(), pet_to_numpy(), extract_pet_metadata()
OCT	DICOM, Proprietary	signal strength, axial resolution, scan pattern	handle_oct_input(), oct_to_numpy(), extract_oct_metadata()
Multi-modal	Any combination	Combined fields, registration parameters	handle_multimodal_input(), multimodal_to_numpy(), extract_multimodal_metadata()

Universal Image Input Handlers

The following core handlers process any imaging data regardless of modality:

def handle_image_input(image_input, modality=None)

Parameters:

• image_input (Any): Input image in any supported format

• modality (str, optional): One of "CT", "MRI", "US", "XR", "MG", "PT", "OCT", "PATH", "MULTI"

Returns: Tuple[np.ndarray, Dict] containing pixel data and metadata

Exceptions:

• UnsupportedFormatError: If image format not recognized

• ProcessingError: If image processing fails

Supported Input Formats:

• NumPy Arrays:

  • 2D arrays: Single-slice grayscale images (shape: H×W)

  • 3D arrays: Volumetric data or RGB images (shapes: D×H×W or H×W×C)

  • 4D arrays: Multi-channel volumes or temporal sequences (shapes: D×H×W×C or T×H×W×C)

  • Supported dtypes: uint8, uint16, int16, float32, float64

• File Formats:

  • DICOM (.dcm): Raw or as pydicom.FileDataset

  • NIFTI (.nii, .nii.gz): Raw or as nibabel.Nifti1Image

  • TIFF (.tif, .tiff): Standard or BigTIFF

  • Slide formats: (.svs, .ndpi, .scn)

  • Common formats: (.png, .jpg, .jpeg)

  • Raw data: (.raw, .mhd/.mha pairs)

  • Videos: (.mp4, .avi) for ultrasound or fluoroscopy

• Python Objects:

  • PyTorch tensors: torch.Tensor with NCHW or NDHWC format

  • TensorFlow tensors: tf.Tensor with NHWC or NDHWC format

  • SimpleITK Images: sitk.Image

  • PIL Images: PIL.Image.Image

  • Bytes/BytesIO: Raw binary data

  • File paths: String paths to image files

  • File-like objects: Objects with read() method

• Structured Data:

  • Dictionaries: {"pixel_array": array, "metadata": dict}

  • Lists of images: [array1, array2, ...] for batched processing

Modality-Specific Handlers

Each modality has specialized handler functions:

def handle_ct_input(ct_input)

Parameters:

• ct_input (Any): CT scan in any supported format

Returns: Tuple[np.ndarray, Dict] containing standardized CT data and metadata

Supported CT-specific formats:

• DICOM series with modality tag "CT"

• NIFTI volumes with calibrated Hounsfield Units

• Raw volumes with rescale parameters

def handle_mri_input(mri_input)

Parameters:

• mri_input (Any): MRI scan in any supported format

Returns: Tuple[np.ndarray, Dict] containing standardized MRI data and metadata

Supported MRI-specific formats:

• DICOM series with modality tag "MR"

• NIFTI volumes with MRI-specific metadata

• DICOM multi-echo or multi-contrast sequences

def handle_us_input(us_input)

Parameters:

• us_input (Any): Ultrasound image or video in any supported format

Returns: Tuple[np.ndarray, Dict] containing standardized ultrasound data and metadata

Supported ultrasound-specific formats:

• DICOM with modality tag "US"

• Video formats (.mp4, .avi) for dynamic ultrasound

• Color or grayscale Doppler images

def handle_xray_input(xray_input)

Parameters:

• xray_input (Any): X-ray image in any supported format

Returns: Tuple[np.ndarray, Dict] containing standardized X-ray data and metadata

Supported X-ray-specific formats:

• DICOM with modality tag "DX" or "CR"

• Standard image formats with appropriate metadata

• Exposure-corrected or raw detector readings

def handle_mammo_input(mammo_input)

Parameters:

• mammo_input (Any): Mammography image in any supported format

Returns: Tuple[np.ndarray, Dict] containing standardized mammography data and metadata

Supported mammography-specific formats:

• DICOM with modality tag "MG"

• Standard image formats with appropriate metadata

• Tomosynthesis image sets

def handle_pathology_input(pathology_input, magnification=None)

Parameters:

• pathology_input (Any): Pathology image in any supported format

• magnification (float, optional): Magnification level for multi-resolution images

Returns: Tuple[np.ndarray, Dict] containing standardized pathology data and metadata

Supported pathology-specific formats:

• Whole slide images (.svs, .ndpi, .tiff)

• Pyramidal TIFF formats

• Multi-resolution tile-based formats

def handle_pet_input(pet_input)

Parameters:

• pet_input (Any): PET or SPECT image in any supported format

Returns: Tuple[np.ndarray, Dict] containing standardized PET/SPECT data and metadata

Supported PET/SPECT-specific formats:

• DICOM with modality tag "PT" or "NM"

• NIFTI with appropriate quantitative units

• SUV-calibrated or raw counts

def handle_oct_input(oct_input)

Parameters:

• oct_input (Any): OCT image in any supported format

Returns: Tuple[np.ndarray, Dict] containing standardized OCT data and metadata

Supported OCT-specific formats:

• DICOM with appropriate OCT-specific tags

• Vendor-specific proprietary formats

• B-scan or volume scans

Model Prediction Format Support

Supported Model Output Formats by Task Type

The library supports all common model output formats across different medical AI tasks. Each format is handled by specialized functions that extract relevant metrics.

Task Type	Supported Output Formats	Primary Metrics	Format Handler Functions
Classification	Probabilities, class indices, medical scores (PI-RADS, BI-RADS, etc.)	Confidence distribution, uncertainty, calibration	handle_classification_output(), analyze_classification_confidence()
Detection	Bounding boxes, heatmaps, coordinates (multiple formats)	Size distribution, location patterns, confidence thresholds	handle_detection_output(), analyze_detection_characteristics()
Segmentation	Binary masks, multi-class masks, probability maps, RLE	Volume statistics, boundary smoothness, region properties	handle_segmentation_output(), analyze_segmentation_characteristics()
Regression	Scalar values, measurement arrays, named measurements	Distribution, bias patterns, correlation with inputs	handle_regression_output(), analyze_regression_characteristics()
Enhancement	Enhanced images, paired before/after, quality metrics	Quality improvement, artifact reduction, detail preservation	handle_enhancement_output(), analyze_enhancement_characteristics()
Multi-task	Combined outputs from multiple tasks	Inter-task relationships, consistency across tasks	handle_multitask_output(), analyze_multitask_consistency()

Universal Model Output Handler

def handle_model_output(output_data, prediction_type=None)

Parameters:

• output_data (Any): Model output in any supported format

• prediction_type (str, optional): One of "classification", "detection", "segmentation", "regression", "enhancement", "multitask"

Returns: Dict containing standardized prediction representation

Exceptions:

• UnsupportedFormatError: If output format not recognized

• ValidationError: If output data is invalid

Supported Output Formats:

• Raw Numerical Formats:

  • NumPy arrays of various shapes depending on task

  • Python lists, tuples, or scalar values

  • PyTorch and TensorFlow tensors

• Framework-Specific Formats:

  • PyTorch detection format: {"boxes": tensor, "scores": tensor, "labels": tensor}

  • TensorFlow detection format: {"detection_boxes": tensor, "detection_scores": tensor, "detection_classes": tensor}

  • COCO format: [{"bbox": [x, y, w, h], "category_id": id, "score": score}, ...]

  • YOLO format: [x_center, y_center, width, height, class_conf, class_pred]

• Medical Domain Formats:

  • PI-RADS scores: {"system": "PI-RADS", "score": 4, "confidence": 0.92}

  • BI-RADS assessment: {"system": "BI-RADS", "category": "4A", "probability": 0.15}

  • TNM staging: {"system": "TNM", "t": "T2", "n": "N0", "m": "M0", "confidence": 0.85}

  • Structured measurements: {"measurements": [{"name": "volume", "value": 45.7, "unit": "cm³"}]}

• Task-Specific Structured Formats:

  • See detailed formats in the task-specific handlers below

Task-Specific Output Handlers

Classification Output Handlers

def handle_classification_output(classification_output)

Parameters:

• classification_output (Any): Classification output in any supported format

Returns: Dict containing standardized classification output

Supported Classification Formats:

• Raw Probabilities:

  • NumPy array: np.array([0.05, 0.85, 0.10]) (class probabilities)

  • List/tuple: [0.05, 0.85, 0.10] (class probabilities)

  • PyTorch tensor: torch.tensor([[0.05, 0.85, 0.10]])

  • TensorFlow tensor: tf.constant([[0.05, 0.85, 0.10]])

• Class Index Format:

  • Single integer: 1 (predicted class index)

  • With confidence: (1, 0.85) (class index, confidence)

  • One-hot encoded: [0, 1, 0]

• Named Class Format:

  • Dictionary: {"predicted_class": "pneumonia", "confidence": 0.85}

  • Named probabilities: {"normal": 0.05, "pneumonia": 0.85, "covid": 0.10}

• Multi-Label Format:

  • Dictionary: {"mass": 0.85, "pneumothorax": 0.12, "effusion": 0.65}

  • Binary indicators: {"mass": 1, "pneumothorax": 0, "effusion": 1}

• Medical Scoring Systems:

  • PI-RADS: {"system": "PI-RADS", "score": 4, "confidence": 0.92}

  • BI-RADS: {"system": "BI-RADS", "category": "4A", "probability": 0.15}

  • ACR TI-RADS: {"system": "TI-RADS", "score": 3, "confidence": 0.78}

  • Lung-RADS: {"system": "Lung-RADS", "category": "3", "malignancy_probability": 0.15}

Detection Output Handlers

def handle_detection_output(detection_output)

Parameters:

• detection_output (Any): Detection output in any supported format

Returns: Dict containing standardized detection output

Supported Detection Formats:

• Bounding Box Formats:

  • List of [x1, y1, x2, y2, conf, class]: [[100, 150, 250, 300, 0.9, 1], ...]

  • List of [x, y, w, h, conf, class]: [[100, 150, 150, 150, 0.9, 1], ...]

  • List of [x_center, y_center, w, h, conf, class]: [[175, 225, 150, 150, 0.9, 1], ...]

• PyTorch Format:

  • Dictionary: {"boxes": torch.tensor([[100, 150, 250, 300]]), "scores": torch.tensor([0.9]), "labels": torch.tensor([1])}

• TensorFlow Format:

  • Dictionary: {"detection_boxes": tf.constant([[0.1, 0.2, 0.3, 0.4]]), "detection_scores": tf.constant([0.9]), "detection_classes": tf.constant([1])}

• COCO Format:

  • List of dictionaries: [{"bbox": [100, 150, 150, 150], "category_id": 1, "score": 0.9}, ...]

• Medical Detection Format:

  • Dictionary: {"detections": [{"bbox": [100, 150, 250, 300], "class": "lesion", "confidence": 0.9, "attributes": {"malignancy": 0.7, "location": "peripheral_zone", "pi_rads_score": 4}}]}

• Localization Formats:

  • Keypoints: np.array([[120, 230, 0.95], [250, 180, 0.80]]) (x, y, confidence)

  • Heatmap: np.array(shape=(512, 512), dtype=float32) (probability map)

Segmentation Output Handlers

def handle_segmentation_output(segmentation_output)

Parameters:

• segmentation_output (Any): Segmentation output in any supported format

Returns: Dict containing standardized segmentation output

Supported Segmentation Formats:

• Mask Formats:

  • Binary mask: np.array(shape=(512, 512), dtype=bool) (foreground/background)

  • Multi-class mask: np.array(shape=(512, 512), dtype=np.uint8) (class indices)

  • One-hot encoded: np.array(shape=(num_classes, H, W), dtype=bool) or np.array(shape=(H, W, num_classes), dtype=bool)

  • Probability maps: np.array(shape=(num_classes, H, W), dtype=np.float32) or np.array(shape=(H, W, num_classes), dtype=np.float32)

• Instance Segmentation:

  • List of masks: [mask1, mask2, mask3] (one mask per instance)

  • Instance IDs: np.array(shape=(H, W), dtype=np.int32) (pixel values = instance IDs)

• Named Masks:

  • Dictionary: {"tumor": tumor_mask, "organ": organ_mask}

• Compressed Formats:

  • RLE: {"counts": [49, 2, 18, 14...], "size": [512, 512]}

• Boundary Representations:

  • Contours: [np.array([[100, 100], [120, 100], [120, 120]...]), ...]

  • Surface mesh: {"vertices": vertices_array, "faces": faces_array}

• Medical Segmentation Format:

  • Dictionary: {"segmentation": {"mask": mask_array, "classes": ["background", "prostate", "lesion"], "confidence_map": conf_array, "volumetric_statistics": {"total_volume_mm3": 45678.9, "class_volumes": {"prostate": 45000.0, "lesion": 678.9}}}}

Regression/Measurement Output Handlers

def handle_regression_output(regression_output)

Parameters:

• regression_output (Any): Regression output in any supported format

Returns: Dict containing standardized regression output

Supported Regression Formats:

• Scalar Formats:

  • Single value: 0.75

  • Multiple values: [0.75, 12.3, 45.6]

• Named Measurements:

  • Dictionary: {"ejection_fraction": 0.65, "stroke_volume": 75.2}

  • With confidence: {"value": 0.75, "confidence": 0.92, "range": [0.70, 0.80]}

• Structured Measurements:

  • List of measurements: {"measurements": [{"name": "tumor_size", "value": 15.2, "unit": "mm", "confidence": 0.9}]}

• Medical Measurements:

  • Cardiac measurements: {"measurements": [{"name": "ejection_fraction", "value": 0.65, "unit": "%", "reference_range": [0.55, 0.70]}]}

  • Tumor measurements: {"measurements": [{"name": "long_axis", "value": 24.3, "unit": "mm"}, {"name": "short_axis", "value": 16.7, "unit": "mm"}]}

  • Organ volumes: {"measurements": [{"name": "liver_volume", "value": 1520, "unit": "cm³", "reference_range": [1200, 1600]}]}

• Time Series:

  • Waveform: np.array(shape=(128), dtype=np.float32) (e.g., for cardiac cycle analysis)

Enhancement/Reconstruction Output Handlers

def handle_enhancement_output(enhancement_output)

Parameters:

• enhancement_output (Any): Enhancement output in any supported format

Returns: Dict containing standardized enhancement output

Supported Enhancement Formats:

• Enhanced Image:

  • NumPy array: np.array(shape=(512, 512), dtype=np.float32)

• Before/After Pair:

  • Dictionary: {"original": original_array, "enhanced": enhanced_array}

• Multiple Reconstructions:

  • Dictionary: {"standard": standard_array, "bone": bone_array, "soft_tissue": soft_tissue_array}

• With Quality Metrics:

  • Dictionary: {"image": enhanced_array, "metrics": {"psnr": 32.6, "ssim": 0.92}}

• Medical Enhancement Formats:

  • Denoised MRI: {"original": noisy_mri, "enhanced": denoised_mri, "metrics": {"snr_improvement": 6.8, "detail_preservation": 0.94}}

  • Super-resolution CT: {"original": low_res_ct, "enhanced": high_res_ct, "metrics": {"resolution_scale": 3, "structure_preservation": 0.91}}

  • Artifact-corrected ultrasound: {"original": artifact_us, "enhanced": corrected_us, "artifacts_removed": ["shadowing", "reverberation"]}

Modality-Specific Metric Functions

The library provides specialized metric functions for each imaging modality. These functions are designed to extract and analyze metrics that are uniquely relevant to each modality.

CT-Specific Metrics

def compute_ct_metrics(image, metadata)

Compute comprehensive metrics for CT images.

Parameters:

• image (np.ndarray): CT image data

• metadata (Dict): CT metadata

Returns: Dict containing CT-specific metrics

Metrics computed:

• HU statistics (mean, std, min, max, percentiles)

• Noise index

• Contrast-to-noise ratio

• Spatial resolution estimate

• Slice thickness verification

• Reconstruction kernel characteristics

• Dose metrics (if available in metadata)

• Metal artifact detection and quantification

• Beam hardening artifact detection

• Motion artifact detection

def analyze_hu_distribution(ct_image, roi=None)

Analyze Hounsfield Unit distribution in CT image.

Parameters:

• ct_image (np.ndarray): CT image data in HU

• roi (np.ndarray or tuple, optional): Region of interest mask or coordinates

Returns: Dict containing HU distribution metrics

def detect_ct_artifacts(ct_image, metadata)

Detect common artifacts in CT images.

Parameters:

• ct_image (np.ndarray): CT image data

• metadata (Dict): CT acquisition metadata

Returns: Dict containing detected artifacts and their severity

def compute_ct_noise_metrics(ct_image, background_roi=None)

Compute noise characteristics in CT images.

Parameters:

• ct_image (np.ndarray): CT image data

• background_roi (np.ndarray or tuple, optional): Background region for noise analysis

Returns: Dict containing noise metrics

MRI-Specific Metrics

def compute_mri_metrics(image, metadata)

Compute comprehensive metrics for MRI images.

Parameters:

• image (np.ndarray): MRI image data

• metadata (Dict): MRI metadata

Returns: Dict containing MRI-specific metrics

Metrics computed:

• SNR (signal-to-noise ratio)

• CNR (contrast-to-noise ratio)

• Ghosting ratio

• Image uniformity

• Resolution and sharpness

• Sequence-specific quality metrics

• B0 field homogeneity estimation

• Motion artifact detection

• Specific MRI artifacts (aliasing, chemical shift, etc.)

• Signal intensity characteristics

def compute_mri_snr(mri_image, signal_roi=None, noise_roi=None)

Compute signal-to-noise ratio in MRI images.

Parameters:

• mri_image (np.ndarray): MRI image data

• signal_roi (np.ndarray or tuple, optional): Signal region

• noise_roi (np.ndarray or tuple, optional): Noise region

Returns: float representing SNR

def analyze_mri_sequence(mri_image, metadata, sequence_type=None)

Analyze MRI quality based on sequence type.

Parameters:

• mri_image (np.ndarray): MRI image data

• metadata (Dict): MRI acquisition metadata

• sequence_type (str, optional): One of "T1", "T2", "FLAIR", "DWI", "ADC", etc.

Returns: Dict containing sequence-specific quality metrics

def detect_mri_artifacts(mri_image, metadata)

Detect common artifacts in MRI images.

Parameters:

• mri_image (np.ndarray): MRI image data

• metadata (Dict): MRI acquisition metadata

Returns: Dict containing detected artifacts and their severity

Ultrasound-Specific Metrics

def compute_us_metrics(image, metadata)

Compute comprehensive metrics for ultrasound images.

Parameters:

• image (np.ndarray): Ultrasound image data

• metadata (Dict): Ultrasound metadata

Returns: Dict containing ultrasound-specific metrics

Metrics computed:

• Signal-to-noise ratio

• Contrast ratio

• Penetration depth

• Resolution (axial, lateral)

• Speckle characteristics

• Cyst detection and analysis

• Doppler quality metrics (if applicable)

• Specific artifacts (shadowing, reverberation, enhancement)

• Gain and dynamic range appropriateness

def compute_us_penetration(us_image, metadata)

Compute ultrasound penetration depth.

Parameters:

• us_image (np.ndarray): Ultrasound image data

• metadata (Dict): Ultrasound acquisition metadata

Returns: Dict containing penetration metrics

def analyze_us_speckle(us_image, roi=None)

Analyze speckle characteristics in ultrasound image.

Parameters:

• us_image (np.ndarray): Ultrasound image data

• roi (np.ndarray or tuple, optional): Region for speckle analysis

Returns: Dict containing speckle metrics

def detect_us_artifacts(us_image)

Detect common artifacts in ultrasound images.

Parameters:

• us_image (np.ndarray): Ultrasound image data

Returns: Dict containing detected artifacts and their characteristics

def analyze_doppler_quality(doppler_image, metadata)

Analyze quality of Doppler ultrasound images.

Parameters:

• doppler_image (np.ndarray): Doppler ultrasound image

• metadata (Dict): Doppler acquisition metadata

Returns: Dict containing Doppler quality metrics

X-ray-Specific Metrics

def compute_xray_metrics(image, metadata)

Compute comprehensive metrics for X-ray images.

Parameters:

• image (np.ndarray): X-ray image data

• metadata (Dict): X-ray metadata

Returns: Dict containing X-ray-specific metrics

Metrics computed:

• Exposure index

• Deviation index

• Signal-to-noise ratio

• Contrast

• Dynamic range

• Resolution and sharpness

• Histogram analysis

• Collimation quality

• Patient positioning assessment

• Specific artifacts (grid lines, foreign objects)

def compute_exposure_metrics(xray_image, metadata)

Compute exposure-related metrics for X-ray images.

Parameters:

• xray_image (np.ndarray): X-ray image data

• metadata (Dict): X-ray acquisition metadata

Returns: Dict containing exposure metrics

def analyze_xray_positioning(xray_image, exam_type=None)

Analyze patient positioning in X-ray images.

Parameters:

• xray_image (np.ndarray): X-ray image data

• exam_type (str, optional): Type of examination (e.g., "chest", "abdomen", "extremity")

Returns: Dict containing positioning quality metrics

def detect_xray_artifacts(xray_image, metadata)

Detect common artifacts in X-ray images.

Parameters:

• xray_image (np.ndarray): X-ray image data

• metadata (Dict): X-ray acquisition metadata

Returns: Dict containing detected artifacts and their severity

Mammography-Specific Metrics

def compute_mammo_metrics(image, metadata)

Compute comprehensive metrics for mammography images.

Parameters:

• image (np.ndarray): Mammography image data

• metadata (Dict): Mammography metadata

Returns: Dict containing mammography-specific metrics

Metrics computed:

• Exposure index

• Contrast

• Signal-to-noise ratio

• Breast tissue coverage

• Compression thickness

• Positioning quality

• Pectoral muscle visualization

• Skin line visualization

• Technical artifacts

• MQSA compliance metrics

def analyze_breast_density(mammo_image, metadata)

Analyze breast density in mammography images.

Parameters:

• mammo_image (np.ndarray): Mammography image data

• metadata (Dict): Mammography acquisition metadata

Returns: Dict containing breast density metrics

def evaluate_mammo_positioning(mammo_image, view=None)

Evaluate positioning quality in mammography.

Parameters:

• mammo_image (np.ndarray): Mammography image data

• view (str, optional): Mammographic view (e.g., "MLO", "CC", "ML", "LM")

Returns: Dict containing positioning quality metrics

def detect_mammo_artifacts(mammo_image, metadata)

Detect common artifacts in mammography images.

Parameters:

• mammo_image (np.ndarray): Mammography image data

• metadata (Dict): Mammography acquisition metadata

Returns: Dict containing detected artifacts and their characteristics

Digital Pathology-Specific Metrics

def compute_pathology_metrics(image, metadata, stain_type=None)

Compute comprehensive metrics for digital pathology images.

Parameters:

• image (np.ndarray): Pathology image data

• metadata (Dict): Pathology metadata

• stain_type (str, optional): Stain type (e.g., "H&E", "IHC", "special")

Returns: Dict containing pathology-specific metrics

Metrics computed:

• Focus quality

• Color consistency

• Stain quality and normalization

• Tissue coverage

• Scanning artifacts

• Tissue fold detection

• Air bubble detection

• Pen marks detection

• Color balance and white balance

• Section thickness consistency

def analyze_stain_quality(path_image, stain_type="H&E")

Analyze stain quality in pathology images.

Parameters:

• path_image (np.ndarray): Pathology image data

• stain_type (str): Stain type

Returns: Dict containing stain quality metrics

def detect_focus_quality(path_image, tile_size=512)

Analyze focus quality in pathology images.

Parameters:

• path_image (np.ndarray): Pathology image data

• tile_size (int, optional): Size of tiles for localized focus analysis

Returns: Dict containing focus quality metrics

def detect_pathology_artifacts(path_image)

Detect common artifacts in pathology images.

Parameters:

• path_image (np.ndarray): Pathology image data

Returns: Dict containing detected artifacts and their severity

PET/SPECT-Specific Metrics

def compute_pet_metrics(image, metadata)

Compute comprehensive metrics for PET/SPECT images.

Parameters:

• image (np.ndarray): PET/SPECT image data

• metadata (Dict): PET/SPECT metadata

Returns: Dict containing PET/SPECT-specific metrics

Metrics computed:

• SUV calibration verification

• Noise equivalent count rate

• Uniformity

• Reconstruction quality

• Resolution

• Motion artifacts

• Attenuation correction quality

• Registration quality (for hybrid imaging)

• Specific artifacts (attenuation, scatter, randoms)

• Quantitative accuracy metrics

def analyze_suv_calibration(pet_image, metadata)

Analyze SUV calibration in PET images.

Parameters:

• pet_image (np.ndarray): PET image data

• metadata (Dict): PET acquisition metadata

Returns: Dict containing SUV calibration metrics

def compute_pet_uniformity(pet_image, background_roi=None)

Compute uniformity in PET background regions.

Parameters:

• pet_image (np.ndarray): PET image data

• background_roi (np.ndarray or tuple, optional): Background region

Returns: Dict containing uniformity metrics

def detect_pet_artifacts(pet_image, metadata)

Detect common artifacts in PET images.

Parameters:

• pet_image (np.ndarray): PET image data

• metadata (Dict): PET acquisition metadata

Returns: Dict containing detected artifacts and their characteristics

OCT-Specific Metrics

def compute_oct_metrics(image, metadata)

Compute comprehensive metrics for OCT images.

Parameters:

• image (np.ndarray): OCT image data

• metadata (Dict): OCT metadata

Returns: Dict containing OCT-specific metrics

Metrics computed:

• Signal strength/quality

• Signal-to-noise ratio

• Axial resolution

• Lateral resolution

• Depth penetration

• Motion artifacts

• Segmentation quality estimation

• Layer visibility

• Specific artifacts (shadowing, mirror artifacts)

• Signal attenuation characteristics

def analyze_oct_signal_quality(oct_image)

Analyze signal quality in OCT images.

Parameters:

• oct_image (np.ndarray): OCT image data

Returns: Dict containing signal quality metrics

def compute_oct_resolution(oct_image, metadata)

Estimate resolution in OCT images.

Parameters:

• oct_image (np.ndarray): OCT image data

• metadata (Dict): OCT acquisition metadata

Returns: Dict containing resolution metrics

def detect_oct_artifacts(oct_image)

Detect common artifacts in OCT images.

Parameters:

• oct_image (np.ndarray): OCT image data

Returns: Dict containing detected artifacts and their severity

Data Distribution Metrics

Functions for tracking input data distributions and detecting shifts in deployment.

def track_metadata_distribution(metadata_entries, field_name, sketch_method=None)

Track distribution of a specific metadata field across multiple images.

Parameters:

• metadata_entries (List[Dict]): List of metadata dictionaries from multiple images

• field_name (str): Name of metadata field to track

• sketch_method (str, optional): Method for distribution tracking ("exact", "kll", "count_min", "none")

Returns: Dict containing distribution statistics

Example fields for tracking:

• Scanner manufacturers/models

• Protocol names

• kVp settings (CT)

• TE/TR values (MRI)

• Field strengths (MRI)

• Slice thickness

• Reconstruction kernels

• Detector exposure settings

• Study descriptions

• Patient positioning

def track_scanner_distribution(metadata_entries, sketch_method=None)

Track distribution of scanner manufacturers and models.

Parameters:

• metadata_entries (List[Dict]): List of metadata dictionaries

• sketch_method (str, optional): Method for distribution tracking

Returns: Dict containing scanner distribution statistics

def track_protocol_distribution(metadata_entries, sketch_method=None)

Track distribution of imaging protocols.

Parameters:

• metadata_entries (List[Dict]): List of metadata dictionaries

• sketch_method (str, optional): Method for distribution tracking

Returns: Dict containing protocol distribution statistics

def analyze_image_dimensions(images, metadata_entries=None)

Analyze distribution of image dimensions.

Parameters:

• images (List[np.ndarray]): List of image arrays

• metadata_entries (List[Dict], optional): List of metadata dictionaries for pixel spacing

Returns: Dict containing dimension statistics

def analyze_intensity_distribution(images, modality=None, sketch_method=None)

Analyze distribution of pixel intensities.

Parameters:

• images (List[np.ndarray]): List of image arrays

• modality (str, optional): Imaging modality for scaling/windowing

• sketch_method (str, optional): Method for distribution tracking

Returns: Dict containing intensity distribution statistics

def detect_distribution_drift(current_distribution, baseline_distribution, field_name=None)

Detect drift between current and baseline distributions.

Parameters:

• current_distribution (Dict): Current distribution statistics

• baseline_distribution (Dict): Baseline distribution statistics

• field_name (str, optional): Field name for specific distribution

Returns: Dict containing drift metrics

Metrics computed:

• Drift magnitude

• Statistical significance

• Maximum divergence

• Distribution comparison statistics (KL divergence, Wasserstein distance, etc.)

• Top contributors to drift

• Visualization data

def compute_distribution_distance(dist1, dist2, method="kl_divergence")

Compute distance between two distributions.

Parameters:

• dist1 (Dict or np.ndarray): First distribution

• dist2 (Dict or np.ndarray): Second distribution

• method (str, optional): Distance method ("kl_divergence", "wasserstein", "js_divergence", "earth_movers")

Returns: float representing distance between distributions

def visualize_distribution_drift(current_distribution, baseline_distribution, field_name)

Generate visualization data for distribution drift.

Parameters:

• current_distribution (Dict): Current distribution statistics

• baseline_distribution (Dict): Baseline distribution statistics

• field_name (str): Field name for specific distribution

Returns: Dict containing visualization data

Privacy-Preserving Distribution Tracking

The library provides privacy-preserving methods for tracking distributions without storing raw data.

KLL Sketch for Continuous Distributions

class KLLSketch:
    def __init__(self, k=200, epsilon=0.01, is_float=True)

Parameters:

• k (int): Size parameter controlling accuracy/memory tradeoff

• epsilon (float): Error bound

• is_float (bool): Whether values are floating point (True) or integer (False)

Methods:

• update(value): Add a value to the sketch

• merge(other_sketch): Merge with another sketch

• get_quantile(q): Get quantile value (0 ≤ q ≤ 1)

• get_quantiles(qs): Get multiple quantile values

• get_rank(value): Get rank of a value (0 to 1)

• get_min_value(): Get minimum value

• get_max_value(): Get maximum value

• get_num_items(): Get number of items

• to_bytes(): Serialize sketch to bytes

• from_bytes(data): Deserialize sketch from bytes (class method)

Example uses:

• SNR distribution across studies

• Confidence score distribution

• Lesion size/volume distribution

• Pixel intensity distribution

• Processing time distribution

CountMinSketch for Categorical Distributions

class CountMinSketch:
    def __init__(self, width=2000, depth=5, seed=None)

Parameters:

• width (int): Width of sketch (larger = more accurate)

• depth (int): Depth of sketch (more hash functions = fewer collisions)

• seed (int, optional): Random seed for hash functions

Methods:

• update(item, count=1): Add an item to the sketch with specified count

• merge(other_sketch): Merge with another sketch

• estimate_count(item): Estimate item count

• estimate_frequency(item): Estimate item frequency (0 to 1)

• get_heavy_hitters(threshold): Get frequent items above threshold

• get_total_count(): Get total count of all items

• to_bytes(): Serialize sketch to bytes

• from_bytes(data): Deserialize sketch from bytes (class method)

Example uses:

• Scanner manufacturer distribution

• Protocol name distribution

• Artifact type distribution

• Error type distribution

• Diagnosis code distribution

HyperLogLog for Cardinality Estimation

class HyperLogLog:
    def __init__(self, precision=14)

Parameters:

• precision (int): Precision parameter (4-16, higher = more accurate)

Methods:

• update(item): Add an item to the sketch

• merge(other_sketch): Merge with another sketch

• get_cardinality(): Estimate unique item count

• to_bytes(): Serialize sketch to bytes

• from_bytes(data): Deserialize sketch from bytes (class method)

Example uses:

• Count of unique study descriptions

• Count of unique protocols

• Count of unique scanners

• Count of unique error messages

• Count of unique patients (with appropriate anonymization) file

• api_key (str, optional): API key

• endpoint (str, optional): API endpoint

Returns: Dict containing API response

Exceptions:

• APIError: If API call fails

Model Performance Metrics

Functions for analyzing model performance without ground truth labels.

Classification Performance Metrics

def analyze_classification_confidence(classification_outputs, threshold=0.5)

Analyze confidence distribution and uncertainty for classification outputs.

Parameters:

• classification_outputs (List[Dict]): List of standardized classification outputs

• threshold (float, optional): Classification threshold

Returns: Dict containing confidence metrics

Metrics computed:

• Confidence distribution statistics (mean, median, std, min, max)

• Entropy of predictions

• Calibration metrics

• Uncertainty estimates

• Prediction stability

• Threshold analysis

• Class distribution analysis

def analyze_classification_calibration(probabilities, bins=10)

Analyze calibration of classification probabilities.

Parameters:

• probabilities (List[np.ndarray]): List of probability distributions

• bins (int, optional): Number of bins for reliability diagram

Returns: Dict containing calibration metrics

def detect_classification_anomalies(classification_outputs, baseline_stats=None)

Detect anomalous predictions based on confidence patterns.

Parameters:

• classification_outputs (List[Dict]): List of standardized classification outputs

• baseline_stats (Dict, optional): Baseline statistics for comparison

Returns: Dict containing anomaly metrics

Detection Performance Metrics

def analyze_detection_characteristics(detection_outputs)

Analyze statistical properties of detection outputs.

Parameters:

• detection_outputs (List[Dict]): List of standardized detection outputs

Returns: Dict containing detection metrics

Metrics computed:

• Bounding box size distribution (area, aspect ratio)

• Spatial distribution analysis

• Confidence distribution

• Number of detections per image

• Detection clustering analysis

• Consistency across similar inputs

• Overlap analysis

def analyze_detection_spatial_distribution(detection_outputs, image_shapes=None)

Analyze spatial distribution of detections.

Parameters:

• detection_outputs (List[Dict]): List of standardized detection outputs

• image_shapes (List[Tuple], optional): Shapes of corresponding images

Returns: Dict containing spatial distribution metrics

def compute_detection_stability(detection_outputs_series)

Measure stability of detections across a series of similar inputs.

Parameters:

• detection_outputs_series (List[List[Dict]]): Series of detection outputs on similar inputs

Returns: Dict containing stability metrics

Segmentation Performance Metrics

def analyze_segmentation_characteristics(segmentation_outputs)

Analyze statistical properties of segmentation outputs.

Parameters:

• segmentation_outputs (List[Dict]): List of standardized segmentation outputs

Returns: Dict containing segmentation metrics

Metrics computed:

• Volume/area distribution

• Shape analysis (compactness, elongation, etc.)

• Boundary smoothness

• Confidence distribution (for probability maps)

• Topology analysis

• Region properties

• Class distribution (for multi-class segmentation)

• Connected component analysis

def analyze_segmentation_boundaries(segmentation_mask)

Analyze boundary characteristics of segmentation mask.

Parameters:

• segmentation_mask (np.ndarray): Segmentation mask

Returns: Dict containing boundary metrics

def compute_segmentation_stability(segmentation_outputs_series)

Measure stability of segmentations across a series of similar inputs.

Parameters:

• segmentation_outputs_series (List[Dict]): Series of segmentation outputs on similar inputs

Returns: Dict containing stability metrics

Uncertainty Estimation

def compute_mc_dropout_uncertainty(model_func, input_data, num_samples=20, **kwargs)

Estimate uncertainty using Monte Carlo dropout.

Parameters:

• model_func (Callable): Function that runs inference with dropout enabled

• input_data (Any): Input data for inference

• num_samples (int, optional): Number of forward passes

• **kwargs: Additional arguments for model_func

Returns: Dict containing uncertainty metrics

def compute_ensemble_uncertainty(predictions)

Estimate uncertainty from ensemble predictions.

Parameters:

• predictions (List[Dict]): Predictions from ensemble members

Returns: Dict containing uncertainty metrics

def compute_predictive_uncertainty(predictions)

Decompose predictive uncertainty into aleatoric and epistemic components.

Parameters:

• predictions (List[Dict]): Multiple predictions (from ensemble or MC dropout)

Returns: Dict containing uncertainty decomposition

Cross-Task Consistency

def analyze_task_consistency(classification_outputs, detection_outputs, segmentation_outputs)

Analyze consistency between different task outputs on the same inputs.

Parameters:

• classification_outputs (List[Dict]): Classification outputs

• detection_outputs (List[Dict]): Detection outputs

• segmentation_outputs (List[Dict]): Segmentation outputs

Returns: Dict containing consistency metrics

Clinical Value Metrics

Functions for evaluating the clinical utility of AI outputs.

def analyze_user_feedback(feedback_entries)

Analyze patterns in user feedback.

Parameters:

• feedback_entries (List[Dict]): User feedback entries

Returns: Dict containing feedback analysis

Metrics computed:

• Overall rating statistics

• Rating distribution by modality/task

• Common issue categories

• Acceptance rate

• Modification patterns

• Time savings estimation

• User satisfaction trends

• Free text comment analysis

def analyze_acceptance_rate(feedback_entries, segment_by=None)

Analyze result acceptance patterns.

Parameters:

• feedback_entries (List[Dict]): User feedback entries

• segment_by (str, optional): Field to segment by (e.g., "modality", "user_role")

Returns: Dict containing acceptance metrics

def analyze_modification_patterns(original_outputs, modified_outputs)

Analyze how clinicians modify AI outputs.

Parameters:

• original_outputs (List[Dict]): Original AI outputs

• modified_outputs (List[Dict]): Clinician-modified outputs

Returns: Dict containing modification pattern metrics

def estimate_time_savings(feedback_entries)

Estimate time saved through AI assistance.

Parameters:

• feedback_entries (List[Dict]): User feedback entries with timing information

Returns: Dict containing time saving metrics

def analyze_clinical_confidence(feedback_entries)

Analyze clinician confidence in AI outputs.

Parameters:

• feedback_entries (List[Dict]): User feedback entries

Returns: Dict containing clinical confidence metrics

def detect_feedback_trends(feedback_entries, time_window=None)

Detect trends in user feedback over time.

Parameters:

• feedback_entries (List[Dict]): User feedback entries

• time_window (str, optional): Time window for trend analysis (e.g., "day", "week", "month")

Returns: Dict containing trend metrics

Integration with third party Platform

The library provides specialized components for integrating with the third party platform, facilitating the third-party workflow where:

• AI teams integrate their inference API into Thirdpar

• Third party provides the PACS software interface to hospitals

• The observability module is integrated by third party

• Third party provides dashboards to AI teams

ThirdpartyObservabilityMiddleware

class ThirdpartyObservabilityMiddleware:
    def __init__(self, config_path=None, vendor_id=None, model_id=None, privacy_level="high")

Parameters:

• config_path (str, optional): Path to configuration file

• vendor_id (str, optional): Vendor ID for AI model

• model_id (str, optional): Model ID for tracking

• privacy_level (str, optional): Privacy strictness level ("standard", "high", "extreme")

Methods:

def pre_inference(self, dicom_data, model_id=None)

Hook to be called before inference by Thirdparty.

Parameters:

• dicom_data (Any): Input DICOM data in any supported format

• model_id (str, optional): Model ID if different from initialization

Returns: Dict containing input metrics for logging (PHI-free)

def post_inference(self, prediction, model_id=None, inference_time=None, input_metadata=None)

Hook to be called after inference byThirdparty.

Parameters:

• prediction (Any): Model prediction in any supported format

• model_id (str, optional): Model ID if different from initialization

• inference_time (float, optional): Inference time in seconds

• input_metadata (Dict, optional): Additional input metadata

Returns: Dict containing output metrics for logging (PHI-free)

def on_feedback(self, feedback_data, model_id=None)

Hook to be called when user feedback is received.

Parameters:

• feedback_data (Dict): User feedback in any supported format

• model_id (str, optional): Model ID if different from initialization

Returns: Dict containing feedback metrics for logging (PHI-free)

def generate_report(self, time_range=None, metrics_filter=None)

Generate aggregated metrics report for the vendor/model.

Parameters:

• time_range (Tuple[datetime, datetime], optional): Time range for report

• metrics_filter (Dict, optional): Filter for specific metrics

Returns: Dict containing aggregated metrics (PHI-free)

Configuration Management for Thirdparty.

def push_config_to_thirdparty(config_path, api_key=None, endpoint=None)

Push observability configuration to Thirdparty.ai platform.

Parameters:

• config_path (str): Path to configuration file

• api_key (str, optional): Thirdpartyai API key

• endpoint (str, optional): Thirdpartyai API endpoint

Returns: Dict containing API response

Exceptions:

• APIError: If API call fails

• ConfigurationError: If configuration is invalid

• AuthenticationError: If authentication fails

def get_config_from_carpl(vendor_id, model_id, api_key=None, endpoint=None)

Get observability configuration from Carpl.ai platform.

Parameters:

• vendor_id (str): Vendor ID

• model_id (str): Model ID

• api_key (str, optional): Carpl.ai API key

• endpoint (str, optional): Carpl.ai API endpoint

Returns: Dict containing configuration

Exceptions:

• APIError: If API call fails

• AuthenticationError: If authentication fails

• NotFoundError: If configuration not found

Privacy-Safe API Integration

class CarplMetricsAPI:
    def __init__(self, api_key=None, api_endpoint=None)

Parameters:

• api_key (str, optional): Carpl.ai API key

• api_endpoint (str, optional): Carpl.ai API endpoint

Methods:

def send_metrics(self, metrics_data, vendor_id, model_id)

Send metrics to Carpl.ai platform.

Parameters:

• metrics_data (Dict): Metrics data (must be PHI-free)

• vendor_id (str): Vendor ID

• model_id (str): Model ID

Returns: Dict containing API response

def get_metrics(self, vendor_id, model_id, time_range=None, metrics_filter=None)

Get metrics from Carpl.ai platform.

Parameters:

• vendor_id (str): Vendor ID

• model_id (str): Model ID

• time_range (Tuple[datetime, datetime], optional): Time range for metrics

• metrics_filter (Dict, optional): Filter for specific metrics

Returns: Dict containing metrics

def get_dashboard_url(self, vendor_id, model_id, dashboard_type="overview")

Get URL for Carpl.ai metrics dashboard.

Parameters:

• vendor_id (str): Vendor ID

• model_id (str): Model ID

• dashboard_type (str, optional): Dashboard type

Returns: str containing dashboard URL

Integration Approaches for Development and Production

The library provides multiple integration approaches to support both development and production environments in the medical AI workflow.

Development Environment Integration

Training Integration with PyTorch

class PyTorchObserver:
    def __init__(self, config_path=None, log_dir=None)

Parameters:

• config_path (str, optional): Path to configuration file

• log_dir (str, optional): Directory for logging training metrics

Methods:

def create_lightning_callback(self)

Create a PyTorch Lightning callback for observability.

Returns: ObservabilityCallback instance for Lightning

def hook_model(self, model)

Hook a PyTorch model for observability.

Parameters:

• model (torch.nn.Module): PyTorch model

Returns: Hooked model

def track_batch(self, inputs, outputs, targets=None, loss=None, batch_idx=None, epoch=None)

Track a training/validation batch.

Parameters:

• inputs (Any): Batch inputs

• outputs (Any): Model outputs

• targets (Any, optional): Ground truth targets

• loss (float, optional): Loss value

• batch_idx (int, optional): Batch index

• epoch (int, optional): Epoch number

Returns: Dict containing batch metrics

def track_epoch(self, epoch, metrics, phase="train")

Track epoch-level metrics.

Parameters:

• epoch (int): Epoch number

• metrics (Dict): Epoch metrics

• phase (str, optional): "train", "validation", or "test"

Returns: Dict containing epoch metrics

def track_validation(self, model, dataloader, device="cuda", metrics_fn=None)

Run and track validation.

Parameters:

• model (torch.nn.Module): PyTorch model

• dataloader (torch.utils.data.DataLoader): Validation dataloader

• device (str, optional): Device for validation

• metrics_fn (callable, optional): Function to compute metrics

Returns: Dict containing validation metrics

Training Integration with TensorFlow/Keras

class TensorFlowObserver:
    def __init__(self, config_path=None, log_dir=None)

Parameters:

• config_path (str, optional): Path to configuration file

• log_dir (str, optional): Directory for logging training metrics

Methods:

def create_keras_callback(self)

Create a Keras callback for observability.

Returns: KerasCallback instance for Keras

def track_batch(self, inputs, outputs, targets=None, loss=None, batch_idx=None, epoch=None)

Track a training/validation batch.

Parameters:

• inputs (Any): Batch inputs

• outputs (Any): Model outputs

• targets (Any, optional): Ground truth targets

• loss (float, optional): Loss value

• batch_idx (int, optional): Batch index

• epoch (int, optional): Epoch number

Returns: Dict containing batch metrics

def track_epoch(self, epoch, metrics, phase="train")

Track epoch-level metrics.

Parameters:

• epoch (int): Epoch number

• metrics (Dict): Epoch metrics

• phase (str, optional): "train", "validation", or "test"

Returns: Dict containing epoch metrics

Inference Pipeline Integration

class InferencePipeline:
    def __init__(self, model, preprocessor=None, postprocessor=None, observer=None)

Parameters:

• model (Callable): Model function or object with call method

• preprocessor (Callable, optional): Preprocessing function

• postprocessor (Callable, optional): Postprocessing function

• observer (ObservabilityClient, optional): Observability client

Methods:

def predict(self, input_data, **kwargs)

Run inference with observability.

Parameters:

• input_data (Any): Input data in any supported format

• **kwargs: Additional arguments for model

Returns: Model prediction

def batch_predict(self, input_batch, **kwargs)

Run batch inference with observability.

Parameters:

• input_batch (List[Any]): Batch of input data

• **kwargs: Additional arguments for model

Returns: List of model predictions

Production Environment Integration

Web Service Integration

def create_fastapi_middleware(observer)

Create FastAPI middleware for observability.

Parameters:

• observer (ObservabilityClient): Observability client

Returns: FastAPI middleware class

def create_flask_middleware(observer)

Create Flask middleware for observability.

Parameters:

• observer (ObservabilityClient): Observability client

Returns: Flask middleware function

class ModelServingApp:
    def __init__(self, model_path, config_path=None, framework="auto", host="0.0.0.0", port=8000)

Parameters:

• model_path (str): Path to saved model

• config_path (str, optional): Path to configuration file

• framework (str, optional): "pytorch", "tensorflow", "onnx", or "auto"

• host (str, optional): Host for serving

• port (int, optional): Port for serving

Methods:

def start(self)

Start the model serving application with observability.

Returns: None

Docker Container Integration

def build_observability_container(model_path, config_path, output_path, base_image=None)

Build Docker container with model and observability.

Parameters:

• model_path (str): Path to saved model

• config_path (str): Path to configuration file

• output_path (str): Path for output container

• base_image (str, optional): Base Docker image

Returns: str containing container ID

DICOM Integration

class DicomNodeObserver:
    def __init__(self, aetitle, port, config_path=None)

Parameters:

• aetitle (str): AE title for DICOM node

• port (int): Port for DICOM node

• config_path (str, optional): Path to configuration file

Methods:

def start(self, model_path, output_directory)

Start DICOM node with observability.

Parameters:

• model_path (str): Path to saved model

• output_directory (str): Directory for output DICOM files

Returns: None

Third-Party Integration with Carpl.ai

For AI Model Developers

class CarplDevSDK:
    def __init__(self, api_key=None, config_path=None)

Parameters:

• api_key (str, optional): Carpl.ai API key

• config_path (str, optional): Path to configuration file

Methods:

def configure_observability(self, model_id, privacy_level="high")

Configure observability for model on Carpl.ai platform.

Parameters:

• model_id (str): Model ID

• privacy_level (str, optional): Privacy level ("standard", "high", "extreme")

Returns: Dict containing configuration status

def get_metrics_dashboard(self, model_id, time_range=None)

Get metrics dashboard URL for model.

Parameters:

• model_id (str): Model ID

• time_range (Tuple[datetime, datetime], optional): Time range for metrics

Returns: str containing dashboard URL

def download_metrics(self, model_id, time_range=None, format="json")

Download metrics for model.

Parameters:

• model_id (str): Model ID

• time_range (Tuple[datetime, datetime], optional): Time range for metrics

• format (str, optional): "json", "csv", or "parquet"

Returns: Dict or bytes containing metrics data

For Carpl.ai Platform Integration

class CarplPlatformSDK:
    def __init__(self, config_path=None)

Parameters:

• config_path (str, optional): Path to configuration file

Methods:

def initialize_vendor(self, vendor_id, vendor_config=None)

Initialize vendor in observability system.

Parameters:

• vendor_id (str): Vendor ID

• vendor_config (Dict, optional): Vendor configuration

Returns: Dict containing initialization status

def register_model(self, vendor_id, model_id, model_type, model_version=None)

Register model in observability system.

Parameters:

• vendor_id (str): Vendor ID

• model_id (str): Model ID

• model_type (str): Model type

• model_version (str, optional): Model version

Returns: Dict containing registration status

def track_inference(self, vendor_id, model_id, dicom_data, prediction, feedback=None)

Track inference for vendor/model.

Parameters:

• vendor_id (str): Vendor ID

• model_id (str): Model ID

• dicom_data (Any): Input DICOM data

• prediction (Any): Model prediction

• feedback (Dict, optional): User feedback

Returns: Dict containing tracking status

def generate_vendor_report(self, vendor_id, model_id=None, time_range=None)

Generate vendor-specific report.

Parameters:

• vendor_id (str): Vendor ID

• model_id (str, optional): Model ID (if None, report for all models)

• time_range (Tuple[datetime, datetime], optional): Time range for report

Returns: Dict containing report data

Privacy and Security

The library provides specialized components for ensuring privacy and security in healthcare settings, particularly important when deployed through platforms like Carpl.ai.

Privacy-Preserving Metrics Collection

class PrivacyManager:
    def __init__(self, privacy_level="high", config=None)

Parameters:

• privacy_level (str): Privacy strictness level ("standard", "high", "extreme")

• config (Dict, optional): Privacy configuration

Methods:

def process_dicom_metadata(self, dicom_metadata)

Process DICOM metadata to remove/transform PHI.

Parameters:

• dicom_metadata (Dict): DICOM metadata

Returns: Dict containing privacy-safe metadata

def process_image_data(self, image_data)

Process image data to remove identifiable information.

Parameters:

• image_data (np.ndarray): Image data

Returns: Privacy-safe derived metrics (not the image itself)

def apply_differential_privacy(self, metrics, epsilon=0.1, delta=1e-6)

Apply differential privacy to aggregated metrics.

Parameters:

• metrics (Dict): Metrics to protect

• epsilon (float, optional): Privacy parameter

• delta (float, optional): Privacy parameter

Returns: Dict containing privacy-protected metrics

def check_phi_leakage(self, data)

Check if data contains potential PHI leakage.

Parameters:

• data (Any): Data to check

Returns: Tuple[bool, List[str]] containing leakage status and fields

Secure Storage

class SecureStorage:
    def __init__(self, storage_path, encryption_key=None)

Parameters:

• storage_path (str): Path to secure storage

• encryption_key (str, optional): Encryption key

Methods:

def store(self, data, data_id)

Securely store data.

Parameters:

• data (Any): Data to store

• data_id (str): Identifier for data

Returns: bool indicating success

def retrieve(self, data_id)

Securely retrieve data.

Parameters:

• data_id (str): Identifier for data

Returns: Retrieved data

def delete(self, data_id)

Securely delete data.

Parameters:

• data_id (str): Identifier for data

Returns: bool indicating success

Access Control

class AccessControlManager:
    def __init__(self, config=None)

Parameters:

• config (Dict, optional): Access control configuration

Methods:

def validate_access(self, user_id, resource_id, access_level)

Validate user access to resource.

Parameters:

• user_id (str): User identifier

• resource_id (str): Resource identifier

• access_level (str): Requested access level

Returns: bool indicating access granted

def create_access_token(self, user_id, resource_ids, access_level, expiry=None)

Create access token for resources.

Parameters:

• user_id (str): User identifier

• resource_ids (List[str]): Resource identifiers

• access_level (str): Access level

• expiry (datetime, optional): Token expiry

Returns: str containing access token

def validate_token(self, token)

Validate access token.

Parameters:

• token (str): Access token

Returns: Dict containing validation result

Configuration Reference

YAML Configuration Format

The library uses YAML configuration files to control its behavior. Below is a reference of the configuration format.

# Main configuration
version: "1.0"
environment: "auto"  # "development", "production", or "auto"

# General settings
general:
  log_level: "info"  # "debug", "info", "warning", "error"
  log_file: "/var/log/medical_ai_observe.log"
  temp_directory: "/tmp/medical_ai_observe"
  cache_enabled: true
  cache_ttl_seconds: 3600

# Tracking settings
tracking:
  input_tracking:
    enabled: true
    track_metadata: true
    track_image_quality: true
    
  output_tracking:
    enabled: true
    confidence_tracking: true
    boundary_analysis: true
    
  feedback_tracking:
    enabled: true
    
  performance_tracking:
    enabled: true
    track_memory: true
    track_inference_time: true

# Privacy settings
privacy:
  level: "high"  # "standard", "high", "extreme"
  
  differential_privacy:
    enabled: true
    epsilon: 0.1
    delta: 1e-6
    
  data_sketching:
    enabled: true
    kll_k: 200
    cm_width: 2000
    cm_depth: 5
    hll_precision: 14
    
  phi_removal:
    enabled: true
    remove_patient_info: true
    remove_institution_info: false
    remove_date_precision: "day"  # "none", "year", "month", "day"

# Storage settings
storage:
  type: "local"  # "local", "remote", "hybrid"
  
  local:
    path: "./observability_data"
    format: "json"  # "json", "sqlite", "parquet"
    retention_days: 90
    
  remote:
    endpoint: "https://api.example.com/metrics"
    auth_method: "oauth2"
    client_id: "${CLIENT_ID}"
    client_secret: "${CLIENT_SECRET}"
    ssl_verify: true
    batch_size: 10
    max_retries: 3
    
  buffer:
    enabled: true
    max_size: 100
    flush_interval_seconds: 60

# Modality-specific settings
modalities:
  ct:
    enabled: true
    metrics:
      - "hu_statistics"
      - "noise_index"
      - "contrast"
      
  mri:
    enabled: true
    metrics:
      - "snr"
      - "cnr"
      - "uniformity"
      
  ultrasound:
    enabled: true
    metrics:
      - "penetration_depth"
      - "speckle_index"
      - "artifact_detection"
      
  xray:
    enabled: true
    metrics:
      - "exposure_index"
      - "contrast"
      - "noise"
      
  # Other modalities...

# Task-specific settings
tasks:
  classification:
    enabled: true
    metrics:
      - "confidence_distribution"
      - "entropy"
      - "calibration"
      
  detection:
    enabled: true
    metrics:
      - "size_distribution"
      - "spatial_distribution"
      - "confidence_distribution"
      
  segmentation:
    enabled: true
    metrics:
      - "volume_statistics"
      - "boundary_smoothness"
      - "confidence_map_analysis"
      
  # Other tasks...

# Framework integrations
frameworks:
  pytorch:
    enabled: true
    hook_forward: true
    track_gradients: false
    
  tensorflow:
    enabled: true
    hook_model: true
    track_gradients: false

# Third-party integrations
integrations:
  carpl:
    enabled: true
    api_endpoint: "https://api.carpl.ai/v1"
    vendor_id: "${VENDOR_ID}"
    
  # Other integrations...

Environment Variables

The library supports configuration through environment variables:

• MEDICAL_AI_OBSERVE_CONFIG: Path to configuration file

• MEDICAL_AI_OBSERVE_ENV: Environment ("development", "production")

• MEDICAL_AI_OBSERVE_LOG_LEVEL: Log level

• MEDICAL_AI_OBSERVE_PRIVACY_LEVEL: Privacy level

• MEDICAL_AI_OBSERVE_STORAGE_PATH: Path for local storage

• MEDICAL_AI_OBSERVE_API_ENDPOINT: API endpoint for remote storage

• MEDICAL_AI_OBSERVE_API_KEY: API key for remote storage

• MEDICAL_AI_OBSERVE_VENDOR_ID: Vendor ID for Carpl.ai integration

• MEDICAL_AI_OBSERVE_MODEL_ID: Model ID for Carpl.ai integration

Best Practices Guide

Integration Best Practices

1. Start Simple

   • Begin with basic metrics before enabling advanced features

   • Add modality-specific metrics incrementally

   • Test with a limited set of studies before full deployment

2. Privacy Considerations

   • Set privacy level to "high" or "extreme" for clinical deployments

   • Use data sketching for distributional data

   • Apply differential privacy for sensitive count metrics

   • Never transmit PHI or PII to third-party services

3. Performance Optimization

   • Enable batching for high-volume deployments

   • Use asynchronous processing for non-blocking operation

   • Set appropriate buffer sizes based on deployment scale

   • Configure appropriate retention policies for local storage

4. Integration with Carpl.ai

   • Use the CarplObservabilityMiddleware for seamless integration

   • Configure vendor-specific metrics dashboards

   • Implement proper access controls for metrics visibility

   • Test integration thoroughly before production deployment

Modality-Specific Recommendations

1. CT Imaging

   • Track HU calibration and consistency

   • Monitor noise index across studies

   • Track dose metrics when available

2. MRI Imaging

   • Monitor sequence-specific quality metrics

   • Track artifact prevalence by sequence type

   • Track protocol parameter distributions

3. Ultrasound Imaging

   • Monitor gain settings and dynamic range

   • Track penetration depth and speckle characteristics

   • Monitor artifact prevalence (shadowing, enhancement)

4. X-ray Imaging

   • Track exposure index and deviation index

   • Monitor positioning quality metrics

   • Track artifact prevalence (grid lines, foreign objects)

5. Mammography

   • Track compression and positioning metrics

   • Monitor breast density distribution

   • Track technical recall rates

6. Digital Pathology

   • Monitor focus quality across slides

   • Track stain characteristics and normalization

   • Monitor scanning artifacts

Task-Specific Recommendations

1. Classification Models

   • Track confidence distribution by class

   • Monitor calibration metrics

   • Track entropy of predictions

2. Detection Models

   • Track size and location distribution of detections

   • Monitor confidence distribution

   • Track number of detections per image

3. Segmentation Models

   • Track volume/area distribution

   • Monitor boundary smoothness

   • Track topology consistency

4. Multi-Task Models

   • Track consistency between tasks

   • Monitor task-specific metrics

   • Track performance correlations between tasks

Version History and Roadmap

Version History

• v1.0.0 (2024-05-18)

  • Initial public release

  • Support for all major imaging modalities

  • Basic metrics for image quality and model performance

  • Integration with Carpl.ai platform

  • Privacy-preserving analytics

• v0.9.0 (2024-04-15)

  • Beta release with limited feature set

  • Core functionality for CT, MRI, and X-ray

  • Basic privacy-preserving analytics

Roadmap

• Q3 2024

  • Advanced uncertainty metrics

  • Expanded modality-specific metrics

  • Enhanced visualization capabilities

• Q4 2024

  • Federated analytics capabilities

  • Advanced clinical impact metrics

  • Expanded API integrations

• Q1 2025

  • Multi-vendor comparative analytics

  • Automated insight generation

  • Predictive maintenance features

• Q2 2025

  • Regulatory compliance reporting

  • Advanced anomaly detection

  • Continuous learning framework

Table of Contents

• API Overview

• Image Format Support

  • Supported Image Formats by Modality

  • Universal Image Input Handlers

  • Modality-Specific Handlers

• Model Prediction Format Support

  • Supported Model Output Formats by Task Type

  • Universal Model Output Handler

  • Task-Specific Output Handlers

• Modality-Specific Metric Functions

  • CT-Specific Metrics

  • MRI-Specific Metrics

  • Ultrasound-Specific Metrics

  • X-ray-Specific Metrics

  • Mammography-Specific Metrics

  • Digital Pathology-Specific Metrics

  • PET/SPECT-Specific Metrics

  • OCT-Specific Metrics

• Data Distribution Metrics

  • Privacy-Preserving Distribution Tracking

• Model Performance Metrics

  • Classification Performance Metrics

  • Detection Performance Metrics

  • Segmentation Performance Metrics

  • Uncertainty Estimation

  • Cross-Task Consistency

• Clinical Value Metrics

• Integration with Carpl.ai Platform

  • CarplObservabilityMiddleware

  • Configuration Management for Carpl.ai

  • Privacy-Safe API Integration

• Integration Approaches for Development and Production

  • Development Environment Integration

  • Production Environment Integration

  • Third-Party Integration with Carpl.ai

• Privacy and Security

  • Privacy-Preserving Metrics Collection

  • Secure Storage

  • Access Control

• Configuration Reference

  • YAML Configuration Format

  • Environment Variables

• Best Practices Guide

  • Integration Best Practices

  • Modality-Specific Recommendations

  • Task-Specific Recommendations

• Version History and Roadmap

  • Version History

  • Roadmap