quantification-MobileNetV5-V4 / scripts /litho_reference_engine.py

Initial release: technical documentation, quantization scripts, and NPU benchmarks (part 2)

2248c41 verified 2 days ago

4.16 kB

"""
Lithotheque Edge - Reference AI Engine (Python)
(c) 2026 GeoStratum

This script provides a non-optimized, purely scientific reference implementation 
 of the processing strategies used in the Lithotheque Edge application.
Specifically:
1. 21-Pass Multi-Scale Tiling (21-MST) for Rock Analysis.
2. High-Resolution Preprocessing for Metrology/Scale Detection.
"""

import numpy as np
import cv2  # Used for high-quality bilinear interpolation
from typing import List, Tuple, Dict

class LithoReferenceEngine:
    def __init__(self, rock_input_size: int = 256, scale_input_size: int = 1024):
        self.rock_size = rock_input_size
        self.scale_size = scale_input_size
        
        # Scientific Weights for 21-pass aggregation
        self.WEIGHT_FINE = 0.50   # 16 tiles (4x4)
        self.WEIGHT_MEDIUM = 0.30 # 4 tiles (2x2)
        self.WEIGHT_GLOBAL = 0.20 # 1 tile (1x1)

    def preprocess_image(self, image: np.ndarray, mode: str = "mobilenet") -> np.ndarray:
        """
        Normalizes pixel values to match model requirements.
        - mobilenet: [-1.0, 1.0]
        - standard: [0.0, 1.0]
        """
        if mode == "mobilenet":
            return (image.astype(np.float32) / 127.5) - 1.0
        return image.astype(np.float32) / 255.0

    def extract_rock_tiles(self, full_image: np.ndarray) -> Dict[str, List[np.ndarray]]:
        """
        Implements the 21-MST strategy.
        Returns a dictionary containing:
        - 'fine': 16 tiles (4x4 grid)
        - 'medium': 4 tiles (2x2 grid)
        - 'global': 1 tile (1x1 full resize)
        """
        h, w, _ = full_image.shape
        passes = {
            "fine": [],
            "medium": [],
            "global": []
        }

        # 1. Global Pass (1x1)
        passes["global"].append(cv2.resize(full_image, (self.rock_size, self.rock_size), interpolation=cv2.INTER_LINEAR))

        # 2. Medium Pass (2x2)
        for r in range(2):
            for c in range(2):
                left, top = (c * w // 2), (r * h // 2)
                right, bottom = ((c + 1) * w // 2), ((r + 1) * h // 2)
                tile = full_image[top:bottom, left:right]
                passes["medium"].append(cv2.resize(tile, (self.rock_size, self.rock_size), interpolation=cv2.INTER_LINEAR))

        # 3. Fine Pass (4x4)
        for r in range(4):
            for c in range(4):
                left, top = (c * w // 4), (r * h // 4)
                right, bottom = ((c + 1) * w // 4), ((r + 1) * h // 4)
                tile = full_image[top:bottom, left:right]
                passes["fine"].append(cv2.resize(tile, (self.rock_size, self.rock_size), interpolation=cv2.INTER_LINEAR))

        return passes

    def process_scale_detection(self, full_image: np.ndarray) -> np.ndarray:
        """
        High-Resolution direct metrology resize (1024x1024).
        """
        return cv2.resize(full_image, (self.scale_size, self.scale_size), interpolation=cv2.INTER_LINEAR)

    def aggregate_21pass_scores(self, fine_scores: List[np.ndarray], medium_scores: List[np.ndarray], global_score: np.ndarray) -> np.ndarray:
        """
        Aggregates classification scores from 21 passes using weighted averages.
        Expected input: lists/arrays of softmax probabilities.
        """
        avg_fine = np.mean(fine_scores, axis=0)
        avg_medium = np.mean(medium_scores, axis=0)
        
        final_score = (
            avg_fine * self.WEIGHT_FINE +
            avg_medium * self.WEIGHT_MEDIUM +
            global_score * self.WEIGHT_GLOBAL
        )
        # Re-normalize to ensure it's a valid distribution
        return final_score / np.sum(final_score)

if __name__ == "__main__":
    print("Lithotheque Reference Engine Loaded.")
    print("Scientific Weights: Fine=50%, Medium=30%, Global=20%")
    # Example usage:
    # engine = LithoReferenceEngine()
    # tiles = engine.extract_rock_tiles(my_high_res_image)
    # print(f"Extracted {len(tiles['fine'])} fine tiles, {len(tiles['medium'])} medium tiles, and {len(tiles['global'])} global pass.")

	"""
	Lithotheque Edge - Reference AI Engine (Python)
	(c) 2026 GeoStratum

	This script provides a non-optimized, purely scientific reference implementation
	of the processing strategies used in the Lithotheque Edge application.
	Specifically:
	1. 21-Pass Multi-Scale Tiling (21-MST) for Rock Analysis.
	2. High-Resolution Preprocessing for Metrology/Scale Detection.
	"""

	import numpy as np
	import cv2 # Used for high-quality bilinear interpolation
	from typing import List, Tuple, Dict

	class LithoReferenceEngine:
	def __init__(self, rock_input_size: int = 256, scale_input_size: int = 1024):
	self.rock_size = rock_input_size
	self.scale_size = scale_input_size

	# Scientific Weights for 21-pass aggregation
	self.WEIGHT_FINE = 0.50 # 16 tiles (4x4)
	self.WEIGHT_MEDIUM = 0.30 # 4 tiles (2x2)
	self.WEIGHT_GLOBAL = 0.20 # 1 tile (1x1)

	def preprocess_image(self, image: np.ndarray, mode: str = "mobilenet") -> np.ndarray:
	"""
	Normalizes pixel values to match model requirements.
	- mobilenet: [-1.0, 1.0]
	- standard: [0.0, 1.0]
	"""
	if mode == "mobilenet":
	return (image.astype(np.float32) / 127.5) - 1.0
	return image.astype(np.float32) / 255.0

	def extract_rock_tiles(self, full_image: np.ndarray) -> Dict[str, List[np.ndarray]]:
	"""
	Implements the 21-MST strategy.
	Returns a dictionary containing:
	- 'fine': 16 tiles (4x4 grid)
	- 'medium': 4 tiles (2x2 grid)
	- 'global': 1 tile (1x1 full resize)
	"""
	h, w, _ = full_image.shape
	passes = {
	"fine": [],
	"medium": [],
	"global": []
	}

	# 1. Global Pass (1x1)
	passes["global"].append(cv2.resize(full_image, (self.rock_size, self.rock_size), interpolation=cv2.INTER_LINEAR))

	# 2. Medium Pass (2x2)
	for r in range(2):
	for c in range(2):
	left, top = (c * w // 2), (r * h // 2)
	right, bottom = ((c + 1) * w // 2), ((r + 1) * h // 2)
	tile = full_image[top:bottom, left:right]
	passes["medium"].append(cv2.resize(tile, (self.rock_size, self.rock_size), interpolation=cv2.INTER_LINEAR))

	# 3. Fine Pass (4x4)
	for r in range(4):
	for c in range(4):
	left, top = (c * w // 4), (r * h // 4)
	right, bottom = ((c + 1) * w // 4), ((r + 1) * h // 4)
	tile = full_image[top:bottom, left:right]
	passes["fine"].append(cv2.resize(tile, (self.rock_size, self.rock_size), interpolation=cv2.INTER_LINEAR))

	return passes

	def process_scale_detection(self, full_image: np.ndarray) -> np.ndarray:
	"""
	High-Resolution direct metrology resize (1024x1024).
	"""
	return cv2.resize(full_image, (self.scale_size, self.scale_size), interpolation=cv2.INTER_LINEAR)

	def aggregate_21pass_scores(self, fine_scores: List[np.ndarray], medium_scores: List[np.ndarray], global_score: np.ndarray) -> np.ndarray:
	"""
	Aggregates classification scores from 21 passes using weighted averages.
	Expected input: lists/arrays of softmax probabilities.
	"""
	avg_fine = np.mean(fine_scores, axis=0)
	avg_medium = np.mean(medium_scores, axis=0)

	final_score = (
	avg_fine * self.WEIGHT_FINE +
	avg_medium * self.WEIGHT_MEDIUM +
	global_score * self.WEIGHT_GLOBAL
	)
	# Re-normalize to ensure it's a valid distribution
	return final_score / np.sum(final_score)

	if __name__ == "__main__":
	print("Lithotheque Reference Engine Loaded.")
	print("Scientific Weights: Fine=50%, Medium=30%, Global=20%")
	# Example usage:
	# engine = LithoReferenceEngine()
	# tiles = engine.extract_rock_tiles(my_high_res_image)
	# print(f"Extracted {len(tiles['fine'])} fine tiles, {len(tiles['medium'])} medium tiles, and {len(tiles['global'])} global pass.")