288 lines
9.7 KiB
Python
288 lines
9.7 KiB
Python
from enum import Enum
|
||
from io import BytesIO
|
||
|
||
import cv2
|
||
import numpy as np
|
||
from PIL import Image, ImageDraw, ImageFont
|
||
|
||
from konabot.common.path import ASSETS_PATH, FONTS_PATH
|
||
from konabot.common.path import DATA_PATH
|
||
import json
|
||
|
||
class CrashType(Enum):
|
||
BURNT = 0
|
||
FROZEN = 1
|
||
|
||
class AirConditioner:
|
||
air_conditioners: dict[str, "AirConditioner"] = {}
|
||
|
||
def __init__(self, id: str) -> None:
|
||
self.id = id
|
||
self.on = False
|
||
self.temperature = 24 # 默认温度
|
||
self.burnt = False
|
||
self.frozen = False
|
||
AirConditioner.air_conditioners[id] = self
|
||
|
||
def change_ac(self):
|
||
self.burnt = False
|
||
self.frozen = False
|
||
self.on = False
|
||
self.temperature = 24 # 重置为默认温度
|
||
|
||
def broke_ac(self, crash_type: CrashType):
|
||
'''
|
||
让空调坏掉，并保存数据
|
||
|
||
:param crash_type: CrashType 枚举，表示空调坏掉的类型
|
||
'''
|
||
match crash_type:
|
||
case CrashType.BURNT:
|
||
self.burnt = True
|
||
case CrashType.FROZEN:
|
||
self.frozen = True
|
||
self.save_crash_data(crash_type)
|
||
|
||
def save_crash_data(self, crash_type: CrashType):
|
||
'''
|
||
如果空调爆炸了，就往本地的 ac_crash_data.json 里该 id 的记录加一
|
||
'''
|
||
data_file = DATA_PATH / "ac_crash_data.json"
|
||
crash_data = {}
|
||
if data_file.exists():
|
||
with open(data_file, "r", encoding="utf-8") as f:
|
||
crash_data = json.load(f)
|
||
if self.id not in crash_data:
|
||
crash_data[self.id] = {"burnt": 0, "frozen": 0}
|
||
match crash_type:
|
||
case CrashType.BURNT:
|
||
crash_data[self.id]["burnt"] += 1
|
||
case CrashType.FROZEN:
|
||
crash_data[self.id]["frozen"] += 1
|
||
with open(data_file, "w", encoding="utf-8") as f:
|
||
json.dump(crash_data, f, ensure_ascii=False, indent=4)
|
||
|
||
def get_crashes_and_ranking(self) -> tuple[int, int]:
|
||
'''
|
||
获取该群在全国空调损坏的数量与排行榜的位置
|
||
'''
|
||
data_file = DATA_PATH / "ac_crash_data.json"
|
||
if not data_file.exists():
|
||
return 0, 1
|
||
with open(data_file, "r", encoding="utf-8") as f:
|
||
crash_data = json.load(f)
|
||
ranking_list = []
|
||
for gid, record in crash_data.items():
|
||
total = record.get("burnt", 0) + record.get("frozen", 0)
|
||
ranking_list.append((gid, total))
|
||
ranking_list.sort(key=lambda x: x[1], reverse=True)
|
||
total_crashes = crash_data.get(self.id, {}).get("burnt", 0) + crash_data.get(self.id, {}).get("frozen", 0)
|
||
rank = 1
|
||
for gid, total in ranking_list:
|
||
if gid == self.id:
|
||
break
|
||
rank += 1
|
||
return total_crashes, rank
|
||
|
||
def text_to_transparent_image(text, font_size=40, padding=0, text_color=(0, 0, 0)):
|
||
"""
|
||
将文本转换为带透明背景的图像，图像大小刚好包含文本
|
||
"""
|
||
# 创建临时图像来计算文本尺寸
|
||
temp_image = Image.new('RGB', (1, 1), (255, 255, 255))
|
||
temp_draw = ImageDraw.Draw(temp_image)
|
||
|
||
font = ImageFont.truetype(FONTS_PATH / "montserrat.otf", font_size)
|
||
|
||
# 获取文本边界框
|
||
bbox = temp_draw.textbbox((0, 0), text, font=font)
|
||
text_width = bbox[2] - bbox[0]
|
||
text_height = bbox[3] - bbox[1]
|
||
|
||
# 计算图像大小（文本大小 + 内边距）
|
||
image_width = int(text_width + 2 * padding)
|
||
image_height = int(text_height + 2 * padding)
|
||
|
||
# 创建RGBA模式的空白图像（带透明通道）
|
||
image = Image.new('RGBA', (image_width, image_height), (0, 0, 0, 0))
|
||
draw = ImageDraw.Draw(image)
|
||
|
||
# 绘制文本（考虑内边距）
|
||
x = padding - bbox[0] # 调整起始位置
|
||
y = padding - bbox[1]
|
||
|
||
# 设置文本颜色（带透明度）
|
||
if len(text_color) == 3:
|
||
text_color = text_color + (255,) # 添加完全不透明的alpha值
|
||
|
||
draw.text((x, y), text, fill=text_color, font=font)
|
||
|
||
# 转换为OpenCV格式（BGRA）
|
||
image_cv = cv2.cvtColor(np.array(image), cv2.COLOR_RGBA2BGRA)
|
||
return image_cv
|
||
|
||
def perspective_transform(image, target, corners):
|
||
"""
|
||
对图像进行透视变换（保持透明通道）
|
||
target: 画布
|
||
corners: 四个角点的坐标，顺序为 [左上, 右上, 右下, 左下]
|
||
"""
|
||
height, width = image.shape[:2]
|
||
|
||
# 源点（原始图像的四个角）
|
||
src_points = np.array([
|
||
[0, 0], # 左上
|
||
[width-1, 0], # 右上
|
||
[width-1, height-1], # 右下
|
||
[0, height-1] # 左下
|
||
], dtype=np.float32)
|
||
|
||
# 目标点（变换后的四个角）
|
||
dst_points = np.array(corners, dtype=np.float32)
|
||
|
||
# 计算透视变换矩阵
|
||
matrix = cv2.getPerspectiveTransform(src_points, dst_points)
|
||
|
||
# 获取画布大小
|
||
target_height, target_width = target.shape[:2]
|
||
|
||
# 应用透视变换（保持所有通道，包括alpha）
|
||
transformed = cv2.warpPerspective(image, matrix, (target_width, target_height), flags=cv2.INTER_LINEAR)
|
||
|
||
return transformed, matrix
|
||
|
||
def blend_with_transparency(background, foreground, position):
|
||
"""
|
||
将带透明通道的前景图像合成到背景图像上
|
||
position: 前景图像在背景图像上的位置 (x, y)
|
||
"""
|
||
bg = background.copy()
|
||
|
||
# 如果背景没有alpha通道，添加一个
|
||
if bg.shape[2] == 3:
|
||
bg = cv2.cvtColor(bg, cv2.COLOR_BGR2BGRA)
|
||
bg[:, :, 3] = 255 # 完全不透明
|
||
|
||
x, y = position
|
||
fg_height, fg_width = foreground.shape[:2]
|
||
bg_height, bg_width = bg.shape[:2]
|
||
|
||
# 确保位置在图像范围内
|
||
x = max(0, min(x, bg_width - fg_width))
|
||
y = max(0, min(y, bg_height - fg_height))
|
||
|
||
# 提取前景的alpha通道并归一化
|
||
alpha_foreground = foreground[:, :, 3] / 255.0
|
||
|
||
# 对于每个颜色通道进行合成
|
||
for c in range(3):
|
||
bg_region = bg[y:y+fg_height, x:x+fg_width, c]
|
||
fg_region = foreground[:, :, c]
|
||
|
||
# alpha混合公式
|
||
bg[y:y+fg_height, x:x+fg_width, c] = (
|
||
alpha_foreground * fg_region +
|
||
(1 - alpha_foreground) * bg_region
|
||
)
|
||
|
||
# 更新背景的alpha通道（如果需要）
|
||
bg_alpha_region = bg[y:y+fg_height, x:x+fg_width, 3]
|
||
bg[y:y+fg_height, x:x+fg_width, 3] = np.maximum(bg_alpha_region, foreground[:, :, 3])
|
||
|
||
return bg
|
||
|
||
def precise_blend_with_perspective(background, foreground, corners):
|
||
"""
|
||
精确合成：根据四个角点将前景图像透视合成到背景上
|
||
"""
|
||
# 创建与背景相同大小的空白图像
|
||
bg_height, bg_width = background.shape[:2]
|
||
|
||
# 如果背景没有alpha通道，转换为BGRA
|
||
if background.shape[2] == 3:
|
||
background_bgra = cv2.cvtColor(background, cv2.COLOR_BGR2BGRA)
|
||
else:
|
||
background_bgra = background.copy()
|
||
|
||
# 创建与背景相同大小的前景图层
|
||
foreground_layer = np.zeros((bg_height, bg_width, 4), dtype=np.uint8)
|
||
|
||
# 计算前景图像在背景中的边界框
|
||
min_x = int(min(corners[:, 0]))
|
||
max_x = int(max(corners[:, 0]))
|
||
min_y = int(min(corners[:, 1]))
|
||
max_y = int(max(corners[:, 1]))
|
||
|
||
# 将变换后的前景图像放置到对应位置
|
||
fg_height, fg_width = foreground.shape[:2]
|
||
if min_y + fg_height <= bg_height and min_x + fg_width <= bg_width:
|
||
foreground_layer[min_y:min_y+fg_height, min_x:min_x+fg_width] = foreground
|
||
|
||
# 创建掩码（只在前景有内容的地方合成）
|
||
mask = (foreground_layer[:, :, 3] > 0)
|
||
|
||
# 合成图像
|
||
result = background_bgra.copy()
|
||
for c in range(3):
|
||
result[:, :, c][mask] = foreground_layer[:, :, c][mask]
|
||
result[:, :, 3][mask] = foreground_layer[:, :, 3][mask]
|
||
|
||
return result
|
||
|
||
def wiggle_transform(image, intensity=2) -> list[np.ndarray]:
|
||
'''
|
||
返回一组图像振动的帧组，模拟空调运作时的抖动效果
|
||
'''
|
||
frames = []
|
||
height, width = image.shape[:2]
|
||
shifts = [(-intensity, 0), (intensity, 0), (0, -intensity), (0, intensity), (0, 0)]
|
||
for dx, dy in shifts:
|
||
M = np.float32([[1, 0, dx], [0, 1, dy]])
|
||
shifted = cv2.warpAffine(image, M, (width, height))
|
||
frames.append(shifted)
|
||
return frames
|
||
|
||
async def generate_ac_image(ac: AirConditioner) -> BytesIO:
|
||
# 找到空调底图
|
||
ac_image = cv2.imread(str(ASSETS_PATH / "img" / "ac" / "ac.png"), cv2.IMREAD_UNCHANGED)
|
||
|
||
if not ac.on:
|
||
# 空调关闭状态，直接返回底图
|
||
pil_final = Image.fromarray(ac_image)
|
||
output = BytesIO()
|
||
pil_final.save(output, format="GIF")
|
||
return output
|
||
|
||
# 根据生成温度文本图像
|
||
text = f"{round(ac.temperature, 1)}°C"
|
||
text_image = text_to_transparent_image(
|
||
text,
|
||
font_size=60,
|
||
text_color=(0, 0, 0) # 黑色文字
|
||
)
|
||
|
||
# 获取长宽比
|
||
height, width = text_image.shape[:2]
|
||
aspect_ratio = width / height
|
||
|
||
# 定义3D变换的四个角点（透视效果）
|
||
# 顺序: [左上, 右上, 右下, 左下]
|
||
corners = np.array([
|
||
[123, 45], # 左上
|
||
[284, 101], # 右上
|
||
[290, 140], # 右下
|
||
[119, 100] # 左下
|
||
], dtype=np.float32)
|
||
|
||
# 对文本图像进行3D变换（保持透明通道）
|
||
transformed_text, transform_matrix = perspective_transform(text_image, ac_image, corners)
|
||
|
||
final_image_simple = blend_with_transparency(ac_image, transformed_text, (0, 0))
|
||
|
||
intensity = max(2, abs(int(ac.temperature) - 24) // 2)
|
||
|
||
frames = wiggle_transform(final_image_simple, intensity=intensity)
|
||
pil_frames = [Image.fromarray(frame) for frame in frames]
|
||
output = BytesIO()
|
||
pil_frames[0].save(output, format="GIF", save_all=True, append_images=pil_frames[1:], loop=0, duration=50, disposal=2)
|
||
return output |