""" Core Ink Estimator Engine for Konica Minolta Ink Estimation Tool Orchestrates RIP rasterization, color transforms, histogram math, area scaling, and cost calculations. """ import os import ctypes import numpy as np from typing import Dict, Any, List, Optional from engine.profile_manager import ProfileManager, ProfileData from engine.pdf_rasterizer import DocumentRasterizer, PageRaster from engine.tac_analyzer import TACAnalyzer # Attempt to load the high-performance C library (supports Windows .dll, macOS/Linux .so) C_LIB = None try: so_path = os.path.join(os.path.dirname(__file__), "fast_lut.so") dll_path = os.path.join(os.path.dirname(__file__), "fast_lut.dll") if os.path.exists(dll_path): C_LIB = ctypes.CDLL(dll_path) elif os.path.exists(so_path): C_LIB = ctypes.CDLL(so_path) except Exception as e: print(f"Notice: C acceleration library not loaded, using NumPy fallback: {e}") class InkEstimator: def __init__(self, base_dir: Optional[str] = None): self.profile_manager = ProfileManager(base_dir) def estimate_document( self, filepath: str, mode: int = 1, media: int = 1, quality: int = 1, dpi: float = 300.0, use_icc: bool = True, quantity: int = 1000, duplex: bool = False, cost_per_liter: Optional[Dict[str, float]] = None, currency: str = "EUR", currency_symbol: str = "€", generate_previews: bool = True ) -> Dict[str, Any]: """ Processes an entire PDF or TIFF document and returns full estimation metrics. """ if cost_per_liter is None: # Default industry averages: €120 / Liter cost_per_liter = {"cyan": 120.0, "magenta": 120.0, "yellow": 120.0, "black": 100.0} # Inspect document doc_info = DocumentRasterizer.inspect_document(filepath) page_count = doc_info["page_count"] profile = self.profile_manager.get_profile(mode, media, quality) pages_results = [] total_hist_c = np.zeros(256, dtype=np.int64) total_hist_m = np.zeros(256, dtype=np.int64) total_hist_y = np.zeros(256, dtype=np.int64) total_hist_k = np.zeros(256, dtype=np.int64) total_tac_hist = np.zeros(401, dtype=np.int64) total_pixels = 0 total_area_m2 = 0.0 for page_idx in range(page_count): raster = DocumentRasterizer.rasterize_page(filepath, page_index=page_idx, dpi=dpi) page_area_m2 = (raster.width_mm / 1000.0) * (raster.height_mm / 1000.0) total_area_m2 += page_area_m2 # Evaluate Page page_res = self._process_raster(raster, profile, use_icc=use_icc, generate_previews=generate_previews) pages_results.append(page_res) total_hist_c += page_res["histograms"]["cyan"] total_hist_m += page_res["histograms"]["magenta"] total_hist_y += page_res["histograms"]["yellow"] total_hist_k += page_res["histograms"]["black"] total_tac_hist += page_res["tac_histogram"] total_pixels += raster.pixel_count # Total Document (Single Copy) pixel_area_m2 = (0.0254 / dpi) ** 2 curve = profile.curve doc_c_ml = float(np.sum(total_hist_c * curve[:, 0]) * pixel_area_m2) doc_m_ml = float(np.sum(total_hist_m * curve[:, 1]) * pixel_area_m2) doc_y_ml = float(np.sum(total_hist_y * curve[:, 2]) * pixel_area_m2) doc_k_ml = float(np.sum(total_hist_k * curve[:, 3]) * pixel_area_m2) doc_total_ml = doc_c_ml + doc_m_ml + doc_y_ml + doc_k_ml # Physical Coverage % cov_c = float(np.sum(total_hist_c * np.arange(256)) / (total_pixels * 255.0) * 100.0) if total_pixels > 0 else 0.0 cov_m = float(np.sum(total_hist_m * np.arange(256)) / (total_pixels * 255.0) * 100.0) if total_pixels > 0 else 0.0 cov_y = float(np.sum(total_hist_y * np.arange(256)) / (total_pixels * 255.0) * 100.0) if total_pixels > 0 else 0.0 cov_k = float(np.sum(total_hist_k * np.arange(256)) / (total_pixels * 255.0) * 100.0) if total_pixels > 0 else 0.0 cov_total = cov_c + cov_m + cov_y + cov_k # Total TAC Distribution Breakdown tac_low_pct = float(np.sum(total_tac_hist[:150]) / total_pixels * 100.0) if total_pixels > 0 else 0.0 tac_mid_pct = float(np.sum(total_tac_hist[150:260]) / total_pixels * 100.0) if total_pixels > 0 else 0.0 tac_high_pct = float(np.sum(total_tac_hist[260:320]) / total_pixels * 100.0) if total_pixels > 0 else 0.0 tac_crit_pct = float(np.sum(total_tac_hist[320:]) / total_pixels * 100.0) if total_pixels > 0 else 0.0 # Physical Sheet & Impression Multipliers if page_count == 1: # Single-page document impressions_per_sheet = 2 if duplex else 1 physical_sheets_per_copy = 1 total_impressions = quantity * impressions_per_sheet total_physical_sheets = quantity doc_c_ml_copy = doc_c_ml * impressions_per_sheet doc_m_ml_copy = doc_m_ml * impressions_per_sheet doc_y_ml_copy = doc_y_ml * impressions_per_sheet doc_k_ml_copy = doc_k_ml * impressions_per_sheet doc_total_ml_copy = doc_total_ml * impressions_per_sheet else: # Multi-page document impressions_per_sheet = 2 if duplex else 1 physical_sheets_per_copy = (page_count + 1) // 2 if duplex else page_count total_impressions = quantity * page_count total_physical_sheets = quantity * physical_sheets_per_copy doc_c_ml_copy = doc_c_ml doc_m_ml_copy = doc_m_ml doc_y_ml_copy = doc_y_ml doc_k_ml_copy = doc_k_ml doc_total_ml_copy = doc_total_ml job_c_ml = doc_c_ml_copy * quantity job_m_ml = doc_m_ml_copy * quantity job_y_ml = doc_y_ml_copy * quantity job_k_ml = doc_k_ml_copy * quantity job_total_ml = doc_total_ml_copy * quantity # Costs ($ / ml = $ / 1000) cost_c = job_c_ml * (cost_per_liter["cyan"] / 1000.0) cost_m = job_m_ml * (cost_per_liter["magenta"] / 1000.0) cost_y = job_y_ml * (cost_per_liter["yellow"] / 1000.0) cost_k = job_k_ml * (cost_per_liter["black"] / 1000.0) cost_total = cost_c + cost_m + cost_y + cost_k # Unit Cost Metrics cost_per_copy = (cost_total / quantity) if quantity > 0 else 0.0 cost_per_page = (cost_per_copy / page_count) if page_count > 0 else 0.0 cost_per_physical_sheet = (cost_total / total_physical_sheets) if total_physical_sheets > 0 else 0.0 cost_per_thousand_copies = cost_per_copy * 1000.0 cost_per_thousand_sheets = cost_per_physical_sheet * 1000.0 return { "document": { "filename": os.path.basename(filepath), "format": doc_info["format"], "page_count": page_count, "total_area_m2": round(total_area_m2, 4), "sheets_per_copy": physical_sheets_per_copy, "is_duplex": duplex }, "parameters": { "mode": mode, "mode_label": ["", "RICH", "STD", "ECO"][mode] if 1 <= mode <= 3 else "RICH", "media": media, "media_label": ["", "Coated", "Matte", "Woodfree"][media] if 1 <= media <= 3 else "Coated", "quality": quality, "quality_label": ["", "Normal (Dither)", "HD (Error Diffusion)"][quality] if 1 <= quality <= 2 else "Normal", "dpi": dpi, "use_icc": use_icc, "quantity": quantity, "duplex": duplex, "currency": currency, "currency_symbol": currency_symbol, "cost_per_liter": cost_per_liter }, "single_copy_metrics": { "cyan_ml": round(doc_c_ml_copy, 4), "magenta_ml": round(doc_m_ml_copy, 4), "yellow_ml": round(doc_y_ml_copy, 4), "black_ml": round(doc_k_ml_copy, 4), "total_ml": round(doc_total_ml_copy, 4), # Grams (density ~ 1.05 g/ml for UV inkjet ink) "cyan_grams": round(doc_c_ml_copy * 1.05, 4), "magenta_grams": round(doc_m_ml_copy * 1.05, 4), "yellow_grams": round(doc_y_ml_copy * 1.05, 4), "black_grams": round(doc_k_ml_copy * 1.05, 4), "total_grams": round(doc_total_ml_copy * 1.05, 4) }, "coverage_percentages": { "cyan_pct": round(cov_c, 2), "magenta_pct": round(cov_m, 2), "yellow_pct": round(cov_y, 2), "black_pct": round(cov_k, 2), "total_tac_avg_pct": round(cov_total, 2) }, "tac_distribution": { "safe_0_150_pct": round(tac_low_pct, 2), "moderate_150_260_pct": round(tac_mid_pct, 2), "warning_260_320_pct": round(tac_high_pct, 2), "critical_exceeded_320_pct": round(tac_crit_pct, 2) }, "job_totals": { "quantity": quantity, "total_impressions": total_impressions, "total_physical_sheets": total_physical_sheets, "physical_sheets_per_copy": physical_sheets_per_copy, "cyan_ml": round(job_c_ml, 2), "magenta_ml": round(job_m_ml, 2), "yellow_ml": round(job_y_ml, 2), "black_ml": round(job_k_ml, 2), "total_ml": round(job_total_ml, 2), "total_liters": round(job_total_ml / 1000.0, 3) }, "cost_analysis": { "cyan_cost": round(cost_c, 2), "magenta_cost": round(cost_m, 2), "yellow_cost": round(cost_y, 2), "black_cost": round(cost_k, 2), "total_job_cost": round(cost_total, 2), "cost_per_copy": round(cost_per_copy, 4), "cost_per_page": round(cost_per_page, 4), "cost_per_sheet": round(cost_per_physical_sheet, 4), "cost_per_1000_copies": round(cost_per_thousand_copies, 2), "cost_per_1000_sheets": round(cost_per_thousand_sheets, 2) }, "pages": [ { "page_number": p["page_number"], "width_mm": p["width_mm"], "height_mm": p["height_mm"], "cyan_ml": p["cyan_ml"], "magenta_ml": p["magenta_ml"], "yellow_ml": p["yellow_ml"], "black_ml": p["black_ml"], "total_ml": p["total_ml"], "coverage_c": p["coverage_c"], "coverage_m": p["coverage_m"], "coverage_y": p["coverage_y"], "coverage_k": p["coverage_k"], "previews": p["previews"] } for p in pages_results ] } def _process_raster(self, raster: PageRaster, profile: ProfileData, use_icc: bool, generate_previews: bool): num_pixels = raster.pixel_count pixel_area_m2 = (0.0254 / raster.dpi) ** 2 curve = profile.curve hist_c = np.zeros(256, dtype=np.int64) hist_m = np.zeros(256, dtype=np.int64) hist_y = np.zeros(256, dtype=np.int64) hist_k = np.zeros(256, dtype=np.int64) tac_hist = np.zeros(401, dtype=np.int64) if C_LIB is not None and profile.hm44_bytes is not None: # High-speed C path c_hist_c = (ctypes.c_uint64 * 256)() c_hist_m = (ctypes.c_uint64 * 256)() c_hist_y = (ctypes.c_uint64 * 256)() c_hist_k = (ctypes.c_uint64 * 256)() c_tac_hist = (ctypes.c_uint64 * 401)() C_LIB.process_cmyk_pipeline( raster.cmyk_bytes, num_pixels, 1 if (use_icc and profile.has_icc) else 0, profile.icc_in_tables if profile.has_icc else b"\x00" * 1024, profile.icc_clut if profile.has_icc else b"\x00" * 4, profile.icc_out_tables if profile.has_icc else b"\x00" * 1024, profile.icc_grid_pts if profile.has_icc else 0, profile.hm44_bytes, None, c_hist_c, c_hist_m, c_hist_y, c_hist_k, c_tac_hist ) hist_c = np.array(list(c_hist_c), dtype=np.int64) hist_m = np.array(list(c_hist_m), dtype=np.int64) hist_y = np.array(list(c_hist_y), dtype=np.int64) hist_k = np.array(list(c_hist_k), dtype=np.int64) tac_hist = np.array(list(c_tac_hist), dtype=np.int64) else: raw_arr = np.frombuffer(raster.cmyk_bytes, dtype=np.uint8).reshape(-1, 4) hist_c = np.bincount(raw_arr[:, 0], minlength=256) hist_m = np.bincount(raw_arr[:, 1], minlength=256) hist_y = np.bincount(raw_arr[:, 2], minlength=256) hist_k = np.bincount(raw_arr[:, 3], minlength=256) tac_hist = np.zeros(401, dtype=np.int64) chunk_size = 2_000_000 for i in range(0, num_pixels, chunk_size): chunk = raw_arr[i:i+chunk_size] chunk_tac = np.sum(chunk, axis=1, dtype=np.uint16) * 100 // 255 tac_hist += np.bincount(np.clip(chunk_tac, 0, 400), minlength=401) c_ml = float(np.sum(hist_c * curve[:, 0]) * pixel_area_m2) m_ml = float(np.sum(hist_m * curve[:, 1]) * pixel_area_m2) y_ml = float(np.sum(hist_y * curve[:, 2]) * pixel_area_m2) k_ml = float(np.sum(hist_k * curve[:, 3]) * pixel_area_m2) total_ml = c_ml + m_ml + y_ml + k_ml cov_c = float(np.sum(hist_c * np.arange(256)) / (num_pixels * 255.0) * 100.0) if num_pixels > 0 else 0.0 cov_m = float(np.sum(hist_m * np.arange(256)) / (num_pixels * 255.0) * 100.0) if num_pixels > 0 else 0.0 cov_y = float(np.sum(hist_y * np.arange(256)) / (num_pixels * 255.0) * 100.0) if num_pixels > 0 else 0.0 cov_k = float(np.sum(hist_k * np.arange(256)) / (num_pixels * 255.0) * 100.0) if num_pixels > 0 else 0.0 previews = {} if generate_previews: previews = TACAnalyzer.generate_previews(raster.cmyk_bytes, raster.width, raster.height) return { "page_number": raster.page_num, "width_mm": raster.width_mm, "height_mm": raster.height_mm, "cyan_ml": round(c_ml, 4), "magenta_ml": round(m_ml, 4), "yellow_ml": round(y_ml, 4), "black_ml": round(k_ml, 4), "total_ml": round(total_ml, 4), "coverage_c": round(cov_c, 2), "coverage_m": round(cov_m, 2), "coverage_y": round(cov_y, 2), "coverage_k": round(cov_k, 2), "histograms": { "cyan": hist_c, "magenta": hist_m, "yellow": hist_y, "black": hist_k }, "tac_histogram": tac_hist, "previews": previews }