当前位置：首页 > news >正文

摸鱼笔记[6]-图像压缩存储.md

news 2026/6/17 20:35:43

摘要

使用AVIF图像压缩方式压缩图片后进行长期存储, 适用于工业视觉检测场景的图像长期保存.

声明

本文内容由 AI 辅助生成, 已经人工审核和编辑。

简介

AVIF格式简介

[https://github.com/Tim0x0/avif-imageio]
[https://blog.timxs.com/archives/9AyEmIqR]
[https://zhuanlan.zhihu.com/p/355256489]
[https://github.com/AOMediaCodec/libavif]
适合使用 AVIF 的场景：

CCD/工业相机采集图像的长期归档存储
云端视觉检测平台的大批量图片压缩传输
需要保留 10-bit 量化信息的精密测量场景

AVIF（AV1 Image File Format）是一种基于 AV1 视频编码的静态图像容器格式，由 Alliance for Open Media（AOMedia）于 2019 年发布。完全免专利费（Royalty-free），AOMedia 开放授权。

技术	效果
可变块大小	4×4 至 128×128 自适应分割，提升复杂纹理区域精度
方向性帧内预测	56 种角度模式 + 滤波器帧内预测，减少空间冗余
调色板模式	对色块区域（如图标、UI）压缩率极高
CDEF + Loop Restoration	去块滤波 + 环路恢复，保持边缘锐利
Tile 并行编码	支持多线程编码，工业批量处理友好

JPEG-AI格式简介

[https://jpeg.org/jpegai/]
JPEG-AI（ISO/IEC 6048，也称 JPEG AI）是联合图像专家组（JPEG）于 2022-2024 年间标准化的新一代图像编码格式，核心特点是基于神经网络的学习型压缩。

特性	说明
端到端学习	分析-合成网络联合优化，率失真函数可微分
内容自适应	网络自动分配码率：复杂纹理区域多比特，平滑区域少比特
语义保留	隐空间特征对人眼敏感结构（边缘、纹理）有天然偏好
机器视觉友好	支持"面向机器"（For Machines）编码模式，保留检测特征
渐进解码	支持从低质量到高质量的逐层重建

pillow-avif-plugin简介

pillow-avif-plugin 是一个用于 Python 的 Pillow 扩展插件，让 Pillow 能够直接读写 AVIF 格式图片。
依赖底层库 libavif，Windows 上通常通过预编译 wheel 自动包含，Linux 可能需要系统安装 libavif。

工程

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
图片批量压缩工具 (AVIF格式)
功能：将 D:/CCD图片/202606 所有图片压缩为AVIF，保留目录结构，支持断点续传
依赖：pip install pillow-avif-plugin
"""import os
import sys
import json
import time
from datetime import datetime
from pathlib import Path# ============================================================
# 尝试导入 pillow-avif-plugin
# ============================================================
try:from PIL import Imageimport pillow_avif  # 注册AVIF插件到PillowAVIF_AVAILABLE = True
except ImportError:print("[错误] 缺少依赖: pillow-avif-plugin")print("安装命令: pip install pillow-avif-plugin")sys.exit(1)# ============================================================
# 配置区域
# ============================================================
SOURCE_DIR = Path("D:/CCD图片/202606")
TARGET_DIR = Path("D:/CCD图片/长期存储")
CHECKPOINT_FILE = TARGET_DIR / ".compress_checkpoint.json"# AVIF压缩参数
AVIF_QUALITY = 60          # 质量 (0-100)，越高画质越好文件越大
AVIF_SPEED = 6             # 编码速度 (0-10)，越高编码越快体积略大
MAX_SIZE = (10000, 10000)    # 最大尺寸限制，超出则等比缩放# 支持的源图片格式
SUPPORTED_EXTS = {'.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.tif', '.webp', '.gif'}# ============================================================
# 断点续传管理器
# ============================================================
class CheckpointManager:"""基于文件路径和修改时间的断点续传管理"""def __init__(self, checkpoint_path):self.checkpoint_path = checkpoint_pathself.data = self._load()def _load(self):"""加载断点记录"""if self.checkpoint_path.exists():try:with open(self.checkpoint_path, 'r', encoding='utf-8') as f:return json.load(f)except (json.JSONDecodeError, Exception):return {}return {}def save(self):"""保存断点记录"""try:# 确保目录存在self.checkpoint_path.parent.mkdir(parents=True, exist_ok=True)with open(self.checkpoint_path, 'w', encoding='utf-8') as f:json.dump(self.data, f, ensure_ascii=False, indent=2)except Exception as e:print(f"  [警告] 断点保存失败: {e}")def is_processed(self, src_path):"""检查文件是否已处理且未被修改"""src_str = str(src_path)if src_str not in self.data:return Falserecord = self.data[src_str]# 检查源文件修改时间try:current_mtime = os.path.getmtime(src_path)stored_mtime = record.get('mtime', 0)if abs(current_mtime - stored_mtime) > 1:return False  # 文件已被修改except OSError:return False# 检查目标文件是否存在target_path = record.get('target_path', '')if not target_path or not Path(target_path).exists():return Falsereturn Truedef mark_processed(self, src_path, target_path):"""标记文件处理完成"""try:mtime = os.path.getmtime(src_path)self.data[str(src_path)] = {'mtime': mtime,'target_path': str(target_path),'size_original': os.path.getsize(src_path),'processed_at': datetime.now().isoformat()}self.save()except Exception:passdef get_stats(self):"""获取断点统计"""return len(self.data)# ============================================================
# 图片压缩器
# ============================================================
class ImageCompressor:"""AVIF图片压缩器"""def __init__(self, quality=60, speed=6, max_size=(4096, 4096)):self.quality = qualityself.speed = speedself.max_size = max_sizeself.stats = {'total': 0,'success': 0,'skipped': 0,'failed': 0,'total_original_mb': 0,'total_compressed_mb': 0}def _resize_if_needed(self, img):"""如果图片超出最大尺寸，等比缩放"""if img.width > self.max_size[0] or img.height > self.max_size[1]:# 使用LANCZOS重采样保持质量img.thumbnail(self.max_size, Image.LANCZOS)return imgdef _prepare_mode(self, img):"""准备合适的色彩模式"""# AVIF支持RGBA，保留透明通道if img.mode in ('RGBA', 'LA', 'P'):return img.convert('RGBA') if img.mode == 'P' and 'transparency' in img.info else img.convert('RGBA')elif img.mode in ('RGB', 'L', 'I;16', 'I'):return img.convert('RGB')else:return img.convert('RGB')def compress(self, src_path, target_path):"""压缩单张图片到AVIF格式返回: (success: bool, original_size: int, compressed_size: int)"""try:# 确保目标目录存在target_path.parent.mkdir(parents=True, exist_ok=True)with Image.open(src_path) as img:# 准备图片img = self._prepare_mode(img)img = self._resize_if_needed(img)# 保存为AVIFimg.save(target_path,format='AVIF',quality=self.quality,speed=self.speed)# 获取文件大小original_size = src_path.stat().st_sizecompressed_size = target_path.stat().st_sizereturn True, original_size, compressed_sizeexcept Exception as e:print(f"    [失败] {src_path.name}: {str(e)[:80]}")# 清理失败的目标文件if target_path.exists():try:target_path.unlink()except Exception:passreturn False, 0, 0# ============================================================
# 主程序
# ============================================================
def format_size(size_bytes):"""格式化文件大小显示"""if size_bytes >= 1024 * 1024 * 1024:return f"{size_bytes / 1024 / 1024 / 1024:.2f} GB"elif size_bytes >= 1024 * 1024:return f"{size_bytes / 1024 / 1024:.2f} MB"elif size_bytes >= 1024:return f"{size_bytes / 1024:.1f} KB"else:return f"{size_bytes} B"def main():print("=" * 70)print("  图片批量压缩工具 (AVIF格式) - 支持断点续传")print("=" * 70)print(f"  源目录:     {SOURCE_DIR}")print(f"  目标目录:   {TARGET_DIR}")print(f"  AVIF质量:   {AVIF_QUALITY}")print(f"  编码速度:   {AVIF_SPEED}")print(f"  最大尺寸:   {MAX_SIZE[0]}x{MAX_SIZE[1]}")print("=" * 70)# 检查源目录if not SOURCE_DIR.exists():print(f"[错误] 源目录不存在: {SOURCE_DIR}")return 1# 创建目标目录TARGET_DIR.mkdir(parents=True, exist_ok=True)# 初始化checkpoint = CheckpointManager(CHECKPOINT_FILE)compressor = ImageCompressor(AVIF_QUALITY, AVIF_SPEED, MAX_SIZE)# 扫描所有图片print("\n[1/4] 扫描图片文件...")image_files = []for ext in SUPPORTED_EXTS:# 同时搜索大小写扩展名image_files.extend(SOURCE_DIR.rglob(f"*{ext}"))image_files.extend(SOURCE_DIR.rglob(f"*{ext.upper()}"))# 去重并排序（保证顺序稳定）image_files = sorted(set(image_files))compressor.stats['total'] = len(image_files)print(f"  发现图片: {len(image_files)} 个")print(f"  断点记录: {checkpoint.get_stats()} 条")# 统计已处理already_done = sum(1 for f in image_files if checkpoint.is_processed(f))print(f"  已完成:   {already_done} 个")print(f"  待处理:   {len(image_files) - already_done} 个")if not image_files:print("[提示] 未找到任何图片文件")return 0# 确认开始if len(image_files) - already_done > 0:print(f"\n[2/4] 按 Enter 开始压缩，或 Ctrl+C 取消...")try:input()except KeyboardInterrupt:print("\n[取消] 用户中断")return 0# 开始压缩print(f"\n[3/4] 开始压缩...")start_time = time.time()last_save_time = start_timefor i, src_path in enumerate(image_files, 1):# 计算相对路径，保持目录结构try:rel_path = src_path.relative_to(SOURCE_DIR)except ValueError:rel_path = src_path.name# 目标路径：保持目录结构，扩展名改为.aviftarget_path = TARGET_DIR / rel_path.with_suffix('.avif')# 检查断点if checkpoint.is_processed(src_path):compressor.stats['skipped'] += 1if i % 100 == 0 or i == len(image_files):print(f"  [{i}/{len(image_files)}] ✓ 跳过: {rel_path}")continue# 显示进度print(f"  [{i}/{len(image_files)}] 压缩: {rel_path}")# 执行压缩success, orig_size, comp_size = compressor.compress(src_path, target_path)if success:compressor.stats['success'] += 1compressor.stats['total_original_mb'] += orig_sizecompressor.stats['total_compressed_mb'] += comp_sizeratio = (1 - comp_size / orig_size) * 100 if orig_size > 0 else 0print(f"         原: {format_size(orig_size)} → 新: {format_size(comp_size)} | 节省: {ratio:.1f}%")# 保存断点checkpoint.mark_processed(src_path, target_path)# 定期保存断点（每30秒）current_time = time.time()if current_time - last_save_time > 30:checkpoint.save()last_save_time = current_timeelse:compressor.stats['failed'] += 1# 最终保存断点checkpoint.save()# 输出统计elapsed = time.time() - start_timeprint(f"\n[4/4] 压缩完成!")print("=" * 70)print(f"  总文件:     {compressor.stats['total']}")print(f"  成功:       {compressor.stats['success']}")print(f"  跳过:       {compressor.stats['skipped']}")print(f"  失败:       {compressor.stats['failed']}")print(f"  耗时:       {elapsed:.1f} 秒 ({elapsed/60:.1f} 分钟)")if compressor.stats['total_original_mb'] > 0:orig_total = compressor.stats['total_original_mb']comp_total = compressor.stats['total_compressed_mb']total_ratio = (1 - comp_total / orig_total) * 100print(f"\n  原始总大小: {format_size(orig_total)}")print(f"  压缩后大小: {format_size(comp_total)}")print(f"  空间节省:   {total_ratio:.1f}%")print("=" * 70)print(f"  断点文件: {CHECKPOINT_FILE}")print("  提示: 如需重新压缩某文件，删除断点文件中对应条目后重跑")print("=" * 70)return 0if __name__ == "__main__":sys.exit(main())

java版本架构

这是一个基于 Java Swing 开发的桌面端批量图片压缩工具，核心功能是将常见图片格式（JPG/PNG/BMP/TIFF/WebP/GIF 等）批量转换为 AVIF 格式，并提供以下特性：

批量/单文件压缩为 AVIF
支持断点续传（暂停后可继续）
支持等比缩放、质量与编码速度调节
保持源目录结构输出
基于 FlatLaf 的现代 Swing 界面主题

技术/库	用途
Java 8+	开发语言
Swing + FlatLaf	桌面 GUI 界面
avif-imageio 0.1.2	AVIF 编解码核心（含原生动态库）
Gson	JSON 配置与断点文件序列化
Gradle	构建工具（从 AboutDialog 推断）

com.avif.compressor/
├── Main.java                          # 程序入口
├── config/
│   └── AppConfig.java                 # 配置管理
├── engine/
│   ├── AvifCompressor.java            # AVIF 压缩引擎
│   ├── CheckpointManager.java         # 断点续传管理
│   └── ImageScanner.java              # 图片扫描器
├── model/
│   ├── CompressSettings.java          # 压缩参数模型
│   └── CompressTask.java              # 单个压缩任务模型
├── ui/
│   ├── MainFrame.java                 # 主窗口
│   ├── CompressWorker.java            # 后台压缩线程
│   ├── SettingsDialog.java            # 设置对话框
│   ├── AboutDialog.java               # 关于对话框
└── util/└── FormatUtils.java               # 格式化工具