17 changed files with 197 additions and 1435 deletions
@ -1,16 +0,0 @@ |
|||||
MIT No Attribution |
|
||||
|
|
||||
Copyright <YEAR> <COPYRIGHT HOLDER> |
|
||||
|
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy of this |
|
||||
software and associated documentation files (the "Software"), to deal in the Software |
|
||||
without restriction, including without limitation the rights to use, copy, modify, |
|
||||
merge, publish, distribute, sublicense, and/or sell copies of the Software, and to |
|
||||
permit persons to whom the Software is furnished to do so. |
|
||||
|
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, |
|
||||
INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A |
|
||||
PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT |
|
||||
HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION |
|
||||
OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE |
|
||||
SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
|
||||
@ -1,2 +0,0 @@ |
|||||
# wuyuanbiaoba |
|
||||
|
|
||||
@ -1,33 +0,0 @@ |
|||||
from time import sleep |
|
||||
|
|
||||
import configSer |
|
||||
import tcp_Ser |
|
||||
import upload.DataReporter |
|
||||
import 标靶识别video |
|
||||
import cv2 |
|
||||
|
|
||||
if __name__ == '__main__': |
|
||||
config_path = "./config.json" |
|
||||
# 读取配置文件 |
|
||||
config_obj= configSer.ConfigOperate(config_path) |
|
||||
json_str = config_obj.config_info.to_json(indent=4) |
|
||||
print(f"当前配置：{json_str}") |
|
||||
|
|
||||
tcp_service = tcp_Ser.TcpSer("0.0.0.0", config_obj.config_info.server.port) |
|
||||
tcp_service.start() |
|
||||
reporter = upload.DataReporter.DataReporter(data_fps=config_obj.config_info.fps.data,video_fps=config_obj.config_info.fps.video) |
|
||||
reporter.register_handler(tcp_service.broadcast_message) |
|
||||
reporter.start() |
|
||||
# 启动video |
|
||||
标靶识别video.configObj=config_obj |
|
||||
processor = 标靶识别video.VideoProcessor(reporter) |
|
||||
# 添加订阅者processor |
|
||||
tcp_service.add_subscribe(processor) |
|
||||
# 启动 |
|
||||
processor.video_mode(config_obj.config_info.capture) |
|
||||
|
|
||||
while True: |
|
||||
sleep(10) |
|
||||
|
|
||||
cv2.waitKey(0) |
|
||||
cv2.destroyAllWindows() |
|
||||
@ -1,114 +0,0 @@ |
|||||
import json |
|
||||
import os |
|
||||
from dataclasses import ( |
|
||||
dataclass, |
|
||||
field |
|
||||
) |
|
||||
from typing import Dict |
|
||||
|
|
||||
import numpy as np |
|
||||
from dataclasses_json import dataclass_json |
|
||||
|
|
||||
import models.target |
|
||||
|
|
||||
_file_path: str |
|
||||
|
|
||||
|
|
||||
@dataclass_json |
|
||||
@dataclass |
|
||||
class Server: |
|
||||
port: int = 0 |
|
||||
|
|
||||
@dataclass_json |
|
||||
@dataclass |
|
||||
class Fps: |
|
||||
data: int = 0 |
|
||||
video: int = 0 |
|
||||
|
|
||||
@dataclass_json |
|
||||
@dataclass |
|
||||
class ConfigInfo: |
|
||||
server:Server |
|
||||
fps:Fps |
|
||||
capture: str = "0" |
|
||||
# 标靶配置 |
|
||||
targets: Dict[int, models.target.CircleTarget] = field(default_factory=dict) |
|
||||
|
|
||||
class ConfigOperate: |
|
||||
_file_path: str |
|
||||
config_info: ConfigInfo |
|
||||
def __init__(self,path:str): |
|
||||
self._file_path = path |
|
||||
self.load2obj_sample() |
|
||||
|
|
||||
|
|
||||
def load2dict(self): |
|
||||
""""读取配置""" |
|
||||
if not os.path.exists(self._file_path): |
|
||||
raise FileNotFoundError(f"配置文件 {self._file_path} 不存在") |
|
||||
|
|
||||
with open(self._file_path) as json_file: |
|
||||
config = json.load(json_file) |
|
||||
return config |
|
||||
|
|
||||
# def load2obj_sample2(self): |
|
||||
# """"读取配置""" |
|
||||
# dic=self.load2dict() |
|
||||
# ts = dic["targets"] |
|
||||
# capture = dic["capture"] |
|
||||
# # 获取矩阵数据 |
|
||||
# matrix_dict = dic.get("perspective", {}) |
|
||||
# # n0=convert_to_ndarray(self.matrix_dict["0"]) |
|
||||
# # 将矩阵转换为字符串 |
|
||||
# # matrix_str = np.array2string(n0, precision=8, separator=', ', suppress_small=True) |
|
||||
# for _,t in ts.items(): |
|
||||
# obj = models.target.TargetInfo(**t) |
|
||||
# area = models.target.RectangleArea.from_dict(obj.rectangle_area) |
|
||||
# thres = models.target.Threshold(**obj.threshold) |
|
||||
# self.targets[obj.id] = models.target.CircleTarget( |
|
||||
# obj.id, |
|
||||
# obj.desc, |
|
||||
# area, |
|
||||
# obj.radius, |
|
||||
# thres, |
|
||||
# obj.base |
|
||||
# ) |
|
||||
# return self.targets |
|
||||
|
|
||||
def load2obj_sample(self): |
|
||||
dic=self.load2dict() |
|
||||
dict_str = json.dumps(dic) |
|
||||
self.config_info=ConfigInfo.from_json(dict_str) |
|
||||
|
|
||||
def save2json_file(self): |
|
||||
json_str = self.config_info.to_json(indent=4) |
|
||||
""""更新配置""" |
|
||||
with open(self._file_path, 'w') as json_file: |
|
||||
json_file.write(json_str) |
|
||||
# json.dump(self, json_file, indent=4) |
|
||||
return None |
|
||||
|
|
||||
|
|
||||
def save_dict_config(self, dict_data:Dict): |
|
||||
""""更新配置""" |
|
||||
with open(self._file_path, 'w') as json_file: |
|
||||
json.dump(dict_data, json_file, indent=4) |
|
||||
return None |
|
||||
|
|
||||
def update_dict_config(self, updates): |
|
||||
""" |
|
||||
更新配置文件中的特定字段。 |
|
||||
:param file_path: 配置文件路径 |
|
||||
:param updates: 包含更新内容的字典 |
|
||||
""" |
|
||||
config_dict = self.load2dict() |
|
||||
config_dict.update(updates) |
|
||||
self.save_dict_config(config_dict) |
|
||||
|
|
||||
def convert_to_ndarray(matrix_data): |
|
||||
""" |
|
||||
将 JSON 中的矩阵数据转换为 numpy ndarray。 |
|
||||
:param matrix_data: JSON 中的矩阵数据（列表形式） |
|
||||
:return: numpy ndarray |
|
||||
""" |
|
||||
return np.array(matrix_data, dtype=np.float64) |
|
||||
@ -1,9 +0,0 @@ |
|||||
def print_hi(name): |
|
||||
# 在下面的代码行中使用断点来调试脚本。 |
|
||||
print(f'Hi, {name}') # 按 Ctrl+F8 切换断点。 |
|
||||
|
|
||||
|
|
||||
# 按装订区域中的绿色按钮以运行脚本。 |
|
||||
if __name__ == '__main__': |
|
||||
print_hi('PyCharm') |
|
||||
|
|
||||
@ -1,122 +0,0 @@ |
|||||
import logging |
|
||||
import socket |
|
||||
import threading |
|
||||
from unittest import case |
|
||||
|
|
||||
from models.msg import Msg |
|
||||
|
|
||||
|
|
||||
class TcpSer(threading.Thread): |
|
||||
# 定义服务器地址和端口 |
|
||||
HOST = '127.0.0.1' |
|
||||
PORT = 2230 |
|
||||
def __init__(self,host,port): |
|
||||
super().__init__() |
|
||||
self.HOST=host |
|
||||
self.PORT=port |
|
||||
self.connected_clients=[] |
|
||||
# 消费者 |
|
||||
self.consumers=[] |
|
||||
self.lock = threading.Lock() |
|
||||
# 处理客户端连接的函数 |
|
||||
def handle_client(self,client_socket): |
|
||||
try: |
|
||||
# 当客户端连接时，将其添加到列表中 |
|
||||
self.connected_clients.append(client_socket) |
|
||||
print(f"新连接: {client_socket.getpeername()}") |
|
||||
|
|
||||
# 保持连接，直到客户端断开 |
|
||||
while True: |
|
||||
# 接收客户端数据（如果需要） |
|
||||
data = client_socket.recv(4096) |
|
||||
msg_str=data.decode('utf-8') |
|
||||
if not data: |
|
||||
break # 如果没有数据，退出循环 |
|
||||
print(f"从 {client_socket.getpeername()} 收到: {msg_str}") |
|
||||
# 反序列化为 实例 |
|
||||
s_cmd = Msg.from_json(msg_str) |
|
||||
valid_msg:bool=True |
|
||||
match s_cmd.cmd: |
|
||||
case "getPoints" | "setPoints": |
|
||||
self.on_data(s_cmd,valid_msg) |
|
||||
case "videoFps"| "dataFps": |
|
||||
self.on_data(s_cmd,valid_msg) |
|
||||
case "setCap": |
|
||||
self.on_data(s_cmd,valid_msg) |
|
||||
# todo 添加处理 |
|
||||
case _: |
|
||||
valid_msg = False |
|
||||
err_msg=f"valid cmd={s_cmd.cmd}" |
|
||||
resp=f"""{{"_from": "dev","cmd": "{s_cmd.cmd}","values": {{"operate": false,"err": "{err_msg}"}}}}""" |
|
||||
client_socket.sendall(resp.encode()) |
|
||||
print("非法命令",resp) |
|
||||
|
|
||||
print("通讯完成") |
|
||||
except Exception as e: |
|
||||
print(f"处理客户端时出错: {e}") |
|
||||
finally: |
|
||||
# 从列表中移除客户端并关闭连接 |
|
||||
if client_socket in self.connected_clients: |
|
||||
self.connected_clients.remove(client_socket) |
|
||||
print(f"连接关闭: {client_socket.getpeername()}") |
|
||||
client_socket.close() |
|
||||
|
|
||||
# 注册的消费者必须携带on_data 方法 |
|
||||
def add_subscribe(self,consumer): |
|
||||
if hasattr(consumer, 'on_data'): |
|
||||
print(f"加入 consumer {consumer} ") |
|
||||
self.consumers.append(consumer) |
|
||||
else: |
|
||||
print("consumer 缺少on_data函数,订阅无效 ") |
|
||||
def on_data(self,msg,valid): |
|
||||
for consumer in self.consumers: |
|
||||
try: |
|
||||
resp=consumer.on_data(msg) |
|
||||
self.broadcast_message(resp) |
|
||||
except Exception as e: |
|
||||
logging.warn("通讯异常",e) |
|
||||
|
|
||||
# 广播消息给所有连接的客户端 |
|
||||
def broadcast_message(self,message:str): |
|
||||
with self.lock: |
|
||||
if len(message)==0: |
|
||||
return |
|
||||
|
|
||||
message+="\n\n" |
|
||||
for client in self.connected_clients: |
|
||||
try: |
|
||||
client.sendall(message.encode()) |
|
||||
except Exception as e: |
|
||||
print(f"向客户端发送消息时出错: {e}") |
|
||||
# 如果发送失败，从列表中移除客户端 |
|
||||
if client in self.connected_clients: |
|
||||
self.connected_clients.remove(client) |
|
||||
client.close() |
|
||||
def run(self): |
|
||||
# 创建服务器套接字 |
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket: |
|
||||
server_socket.bind((self.HOST,self.PORT)) |
|
||||
server_socket.listen() |
|
||||
print(f"服务器监听在 {self.HOST}:{self.PORT}...") |
|
||||
|
|
||||
try: |
|
||||
# 保持服务器运行并接受新的连接 |
|
||||
while True: |
|
||||
client_socket, addr = server_socket.accept() |
|
||||
print(f"连接来自 {addr}") |
|
||||
|
|
||||
# 为每个客户端启动一个线程 |
|
||||
client_thread = threading.Thread(target=self.handle_client, args=(client_socket,)) |
|
||||
client_thread.daemon = True # 守护线程，服务器关闭时自动结束 |
|
||||
client_thread.start() |
|
||||
|
|
||||
except KeyboardInterrupt: |
|
||||
print("服务器关闭...") |
|
||||
finally: |
|
||||
# 关闭所有客户端连接 |
|
||||
for client in self.connected_clients: |
|
||||
client.close() |
|
||||
server_socket.close() |
|
||||
if __name__ == '__main__': |
|
||||
tcp=TcpSer("127.0.0.1",2230) |
|
||||
tcp.run() |
|
||||
@ -1,34 +0,0 @@ |
|||||
from typing import Dict, List |
|
||||
|
|
||||
import cv2 |
|
||||
import base64 |
|
||||
|
|
||||
import numpy as np |
|
||||
|
|
||||
|
|
||||
def frame_to_base64(frame, format="JPEG"): |
|
||||
"""将 OpenCV 读取的图片帧转换为 Base64 编码的字符串""" |
|
||||
# 将图片帧编码为 JPEG 或 PNG 格式 |
|
||||
if format.upper() == "JPEG": |
|
||||
encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 80] # JPEG 压缩质量 |
|
||||
elif format.upper() == "PNG": |
|
||||
encode_param = [int(cv2.IMWRITE_PNG_COMPRESSION), 9] # PNG 压缩级别 |
|
||||
else: |
|
||||
raise ValueError("Unsupported format. Use 'JPEG' or 'PNG'.") |
|
||||
|
|
||||
result, encoded_frame = cv2.imencode(f".{format.lower()}", frame, encode_param) |
|
||||
if not result: |
|
||||
raise ValueError("Failed to encode frame.") |
|
||||
|
|
||||
# 将编码后的字节流转换为 Base64 字符串 |
|
||||
base64_string = base64.b64encode(encoded_frame).decode("utf-8") |
|
||||
return base64_string |
|
||||
|
|
||||
|
|
||||
|
|
||||
# 自定义编码器和解码器 |
|
||||
def encode_perspective(value: np.ndarray) -> List[List[float]]: |
|
||||
return value.tolist() |
|
||||
|
|
||||
def decode_perspective(value: List[List[float]]) -> np.ndarray: |
|
||||
return np.array(value) |
|
||||
@ -0,0 +1,60 @@ |
|||||
|
from time import sleep |
||||
|
|
||||
|
import cv2 |
||||
|
import threading |
||||
|
import time |
||||
|
|
||||
|
import numpy as np |
||||
|
import requests |
||||
|
from flask import Flask, Response, render_template_string, request |
||||
|
|
||||
|
import utils |
||||
|
|
||||
|
app_flask = Flask(__name__) |
||||
|
|
||||
|
# 全局变量，用于存储最新的帧 |
||||
|
latest_frame:np.ndarray = None |
||||
|
lock = threading.Lock() |
||||
|
is_running = True |
||||
|
|
||||
|
def update_latest_frame(n_bytes:np.ndarray): |
||||
|
global latest_frame |
||||
|
latest_frame=n_bytes |
||||
|
|
||||
|
def generate_mjpeg(): |
||||
|
"""生成MJPEG格式的视频流""" |
||||
|
while is_running: |
||||
|
with lock: |
||||
|
if latest_frame is None: |
||||
|
continue |
||||
|
_,latest_frame_bytes = utils.frame_to_img(latest_frame) |
||||
|
frame = latest_frame_bytes.tobytes() |
||||
|
pass |
||||
|
sleep(0.1) |
||||
|
yield (b'--frame\r\n' |
||||
|
b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n') |
||||
|
|
||||
|
|
||||
|
@app_flask.route('/video_flow') |
||||
|
def video_feed(): |
||||
|
"""视频流路由""" |
||||
|
return Response( |
||||
|
generate_mjpeg(), |
||||
|
mimetype='multipart/x-mixed-replace; boundary=frame' |
||||
|
) |
||||
|
|
||||
|
def run(): |
||||
|
port=2240 |
||||
|
print(f"推流服务启动,访问端口 127.0.0.1:{port}/video_flow") |
||||
|
app_flask.run(host='0.0.0.0', port=port, threaded=True) |
||||
|
|
||||
|
def video_server_run(): |
||||
|
thread = threading.Thread(target=run) |
||||
|
thread.daemon = True # 设置为守护线程，主线程退出时自动终止 |
||||
|
thread.start() |
||||
|
def stop(): |
||||
|
global is_running |
||||
|
is_running=False |
||||
|
|
||||
|
if __name__ == '__main__': |
||||
|
run() |
||||
@ -1,164 +0,0 @@ |
|||||
import math |
|
||||
|
|
||||
import cv2 |
|
||||
import numpy as np |
|
||||
import matplotlib.pyplot as plt |
|
||||
import matplotlib |
|
||||
matplotlib.use('TkAgg') |
|
||||
|
|
||||
elevation=0 |
|
||||
azimuth=0 |
|
||||
def circle3d(): |
|
||||
global elevation,azimuth |
|
||||
# 创建一个新的图和一个3D坐标轴 |
|
||||
fig = plt.figure() |
|
||||
ax = fig.add_subplot(111, projection='3d') |
|
||||
|
|
||||
# 定义圆的参数 |
|
||||
radius = 1 # 半径 |
|
||||
theta = np.linspace(0, 2 * np.pi, 100) # 参数角度 |
|
||||
x = radius * np.cos(theta) # x坐标 |
|
||||
y = radius * np.sin(theta) # y坐标 |
|
||||
z = np.zeros_like(theta) # z坐标（这里圆在xy平面上） |
|
||||
|
|
||||
# 绘制圆 |
|
||||
# ax.plot(x, y, z, label='3D Circle', color='b') |
|
||||
|
|
||||
# 绘制圆的正俯视图 |
|
||||
# 正视图（xz平面） |
|
||||
ax.plot(x, np.zeros_like(theta), z, label='z View', color='r') |
|
||||
ax.plot(np.zeros_like(theta), y, z, label='h View', color='b') |
|
||||
# 俯视图（xy平面） |
|
||||
ax.plot(x, y, np.zeros_like(theta), label='Top View', color='g') |
|
||||
|
|
||||
# 设置坐标轴范围 |
|
||||
ax.set_xlim([-2, 2]) |
|
||||
ax.set_ylim([-2, 2]) |
|
||||
ax.set_zlim([-2, 2]) |
|
||||
ax.set_xlabel('X') |
|
||||
ax.set_ylabel('Y') |
|
||||
ax.set_zlabel('Z') |
|
||||
# 添加图例 |
|
||||
ax.legend() |
|
||||
# 设置初始视角 |
|
||||
# ax.view_init(elev=53, azim=-48,roll=-38) # 俯仰角30度，方位角45度 |
|
||||
ax.view_init(elev=90, azim=0, roll=0) # 俯仰角30度，方位角45度 |
|
||||
# 隐藏整个坐标轴 |
|
||||
# ax.axis('off') |
|
||||
# 设置窗口透明度 |
|
||||
fig.canvas.manager.window.attributes('-alpha', 0.6) # 设置窗口透明度为 0.6 |
|
||||
# 显示图形 |
|
||||
plt.show() |
|
||||
# input("请用鼠标转动 调整角度 elevation and azimuth: ") |
|
||||
# 获取当前的 elevation 和 azimuth |
|
||||
elevation = ax.elev |
|
||||
azimuth = ax.azim |
|
||||
|
|
||||
|
|
||||
def perspective_transform_by_angle(img, _elevation, _azimuth): |
|
||||
h, w = img.shape[:2] |
|
||||
|
|
||||
# 根据角度计算目标点位置 |
|
||||
fov = math.radians(45) # 假设视场角45度 |
|
||||
dz = 1 / math.tan(_elevation) |
|
||||
dx = dz * math.tan(_azimuth) |
|
||||
|
|
||||
# 源图像四个角点(原始斜视图) |
|
||||
src_points = np.float32([[0, 0], [w, 0], [w, h], [0, h]]) |
|
||||
|
|
||||
# 计算目标点(俯视图) |
|
||||
dst_points = np.float32([ |
|
||||
[w * 0.5 * (1 - dx), h * 0.5 * (1 - dz)], # 左上 |
|
||||
[w * 0.5 * (1 + dx), h * 0.5 * (1 - dz)], # 右上 |
|
||||
[w * 0.5 * (1 + dx), h * 0.5 * (1 + dz)], # 右下 |
|
||||
[w * 0.5 * (1 - dx), h * 0.5 * (1 + dz)] # 左下 |
|
||||
]) |
|
||||
|
|
||||
# 获取变换矩阵并应用 |
|
||||
M = cv2.getPerspectiveTransform(src_points, dst_points) |
|
||||
transformed_image=cv2.warpPerspective(img, M, (h,w)) |
|
||||
# 显示原始图像和变换后的图像 |
|
||||
fig, ax = plt.subplots(1, 2,figsize= (12, 6)) |
|
||||
ax[0].imshow(img) |
|
||||
ax[0].set_title("Original Image") |
|
||||
ax[1].imshow(transformed_image) |
|
||||
ax[1].set_title("Transformed Image") |
|
||||
plt.show() |
|
||||
|
|
||||
|
|
||||
def trans_img(image_path ,x,y): |
|
||||
global elevation, azimuth |
|
||||
image = cv2.imread(image_path) |
|
||||
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) |
|
||||
|
|
||||
# 获取图像尺寸 |
|
||||
height, width = image.shape[:2] |
|
||||
|
|
||||
center_x, center_y = 532,385 #圆心 |
|
||||
# 定义源图像的四个点（假设为矩形） |
|
||||
sx1=center_x-100 |
|
||||
sx2=center_x+100 |
|
||||
sy1=center_y-100 |
|
||||
sy2=center_y + 100 |
|
||||
src_points = np.float32([ |
|
||||
[sx1, sy1], # 左上 |
|
||||
[sx2, sy1], # 右上 |
|
||||
[sx2, sy2], # 右下 |
|
||||
[sx1, sy2] # 左下 |
|
||||
]) |
|
||||
# src_points = np.float32([ |
|
||||
# [center_x, sy1], # 上 |
|
||||
# [sx2, center_y], # 右 |
|
||||
# [center_x, sy2], # 下 |
|
||||
# [sx1, center_y] # 左 |
|
||||
# ]) |
|
||||
# 根据 elevation 和 azimuth 计算目标点 |
|
||||
# 这里是一个简化的计算方法，可以根据实际需求调整 |
|
||||
# 假设目标点在透视变换后的坐标 |
|
||||
# 将斜视的画面变成俯视的画面 |
|
||||
radius = 100 |
|
||||
|
|
||||
# 根据俯仰角和方位角计算目标点的偏移 |
|
||||
offset_x = radius * np.sin(elevation) * np.cos(azimuth) |
|
||||
offset_y = radius * np.sin(elevation) * np.sin(azimuth) |
|
||||
offset_z = radius * np.cos(elevation) |
|
||||
print(f"offset_x={offset_x},offset_y={offset_y}") |
|
||||
# 计算目标点 |
|
||||
# dst_points = np.float32([ |
|
||||
# [0 - offset_x, 0 - offset_y], # 左上 |
|
||||
# [width + offset_x, 0 - offset_y], # 右上 |
|
||||
# [width + offset_x, height + offset_y], # 右下 |
|
||||
# [0 - offset_x, height + offset_y] # 左下 |
|
||||
# ]) |
|
||||
dst_points = np.float32([ |
|
||||
[sx1- offset_x, sy1- offset_y], # 上 |
|
||||
[sx2 + offset_x, sy1 - offset_y], # 右上 |
|
||||
[sx2 + offset_x, sy2+ offset_y], # 右下 |
|
||||
[sx1 - offset_x, sy2 + offset_y] # 下 |
|
||||
]) |
|
||||
# 计算透视变换矩阵 |
|
||||
matrix = cv2.getPerspectiveTransform(src_points, dst_points) |
|
||||
|
|
||||
# 应用透视变换 |
|
||||
transformed_image = cv2.warpPerspective(image, matrix, (width, height)) |
|
||||
|
|
||||
# 显示原始图像和变换后的图像 |
|
||||
fig, ax = plt.subplots(1, 2,figsize= (12, 6)) |
|
||||
ax[0].imshow(image) |
|
||||
ax[0].set_title("Original Image") |
|
||||
ax[1].imshow(transformed_image) |
|
||||
ax[1].set_title("Transformed Image") |
|
||||
plt.show() |
|
||||
|
|
||||
if __name__ == '__main__': |
|
||||
circle3d() |
|
||||
print("测试============") |
|
||||
print(f"elevation: {elevation}") |
|
||||
print(f" azimuth: {azimuth}") |
|
||||
img_path="images/trans/subRawImg.jpg" |
|
||||
rawimg=cv2.imread(img_path) |
|
||||
cv2.imshow("raw Image", rawimg) |
|
||||
# trans_img(img_path,elevation,azimuth) |
|
||||
elev = math.radians(elevation) |
|
||||
azim = math.radians(azimuth) |
|
||||
perspective_transform_by_angle(rawimg, elev,azim) |
|
||||
@ -1,254 +0,0 @@ |
|||||
import json |
|
||||
from time import sleep |
|
||||
from typing import Dict |
|
||||
from dataclasses import asdict |
|
||||
import cv2 |
|
||||
import numpy as np |
|
||||
import signal |
|
||||
import sys |
|
||||
import threading |
|
||||
|
|
||||
import models.target |
|
||||
import tcp_Ser |
|
||||
# 定义全局变量 |
|
||||
drawing = False # 是否正在绘制 |
|
||||
start_point=models.target.Point |
|
||||
end_point = models.target.Point |
|
||||
target_rectangle_dict:Dict[int,models.target.CircleTarget]={} |
|
||||
sigExit=False # 是否退出 |
|
||||
#数据广播 |
|
||||
sig_broadcast=True |
|
||||
tcp = tcp_Ser.TcpSer("127.0.0.1", 2230) |
|
||||
myb=threading.Thread(target=tcp.run) |
|
||||
myb.start() |
|
||||
def check_exit(sig, frame): |
|
||||
global sigExit |
|
||||
print(f"收到退出信号 sig={sig}") |
|
||||
sigExit=True |
|
||||
sleep(1) |
|
||||
print("程序退出") |
|
||||
sys.exit(0) |
|
||||
|
|
||||
#鼠标回调函数 |
|
||||
def add_rectangle(event, x, y, flags, param): |
|
||||
global start_point, end_point, drawing |
|
||||
|
|
||||
if event == cv2.EVENT_LBUTTONDOWN: # 左键按下 |
|
||||
print("左键按下") |
|
||||
start_point = models.target.Point(x,y) |
|
||||
end_point = start_point |
|
||||
drawing = True |
|
||||
elif event == cv2.EVENT_MOUSEMOVE: # 鼠标移动 |
|
||||
if drawing: |
|
||||
end_point = models.target.Point(x,y) |
|
||||
elif event == cv2.EVENT_LBUTTONUP: # 左键抬起 |
|
||||
print("左键抬起") |
|
||||
drawing = False |
|
||||
end_point = models.target.Point(x,y) |
|
||||
if start_point==end_point: |
|
||||
return |
|
||||
distance = cv2.norm(tuple(start_point), tuple(end_point), cv2.NORM_L2) |
|
||||
if distance<20: |
|
||||
print("距离小于20,无效区域") |
|
||||
return |
|
||||
target_id=len(target_rectangle_dict) |
|
||||
# 圆标靶半径 mm |
|
||||
radius= 20.0 |
|
||||
area=models.target.RectangleArea(start_point.x,start_point.y,end_point.x-start_point.x,end_point.y-start_point.y) |
|
||||
new_target=models.target.CircleTarget(target_id,area,radius) |
|
||||
print(f"新增区域[{target_id}] => {start_point, end_point}") |
|
||||
target_rectangle_dict[target_id] = new_target |
|
||||
|
|
||||
|
|
||||
|
|
||||
|
|
||||
def draw_rectangle(img): |
|
||||
gray_frame = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) |
|
||||
|
|
||||
ret, gray_frame = cv2.threshold(gray_frame, 120, 255, cv2.THRESH_BINARY) |
|
||||
# 高斯滤波 |
|
||||
gray_frame = cv2.GaussianBlur(gray_frame, (5, 5), 1) |
|
||||
cv2.imshow('binaryImg', gray_frame) |
|
||||
|
|
||||
if len(target_rectangle_dict)==0: return |
|
||||
|
|
||||
#上报有新数据的点 |
|
||||
once_upload:Dict[int,models.target.CircleTarget]={} |
|
||||
|
|
||||
# 绘图-历史点 |
|
||||
for i, tr in target_rectangle_dict.items(): |
|
||||
_start_point = tr.rectangle_area[0] |
|
||||
_end_point = tr.rectangle_area[1] |
|
||||
#绘制标靶区域 |
|
||||
cv2.rectangle(img,tuple(_start_point), tuple(_end_point), (255, 0, 0), 2) |
|
||||
#检测 |
|
||||
sub_image = extract_sub_image(gray_frame, _start_point, _end_point) |
|
||||
circles = circle2_detect(sub_image) |
|
||||
if len(circles) == 0: |
|
||||
continue |
|
||||
center,radius=circle_show(img,circles,_start_point) |
|
||||
# 纪录圆心位置 |
|
||||
tr.center_point=center |
|
||||
tr.radius_pix=radius |
|
||||
if tr.is_init: |
|
||||
tr.center_init=tr.center_point |
|
||||
tr.is_init=False |
|
||||
tar = tr.circle_displacement() |
|
||||
msg=f"[{tar.id}]displacement_pix={tar.displacement_pix},displacement_phy={tar.displacement_phy}" |
|
||||
print(msg) |
|
||||
once_upload[tr.id]=tr |
|
||||
|
|
||||
|
|
||||
#过滤无效空数据 |
|
||||
if len(once_upload.items())==0: return |
|
||||
json_str = json.dumps( |
|
||||
{k:asdict(v) for k, v in once_upload.items() if v.is_init==False} |
|
||||
) |
|
||||
print(f"标靶数据={json_str}",json_str) |
|
||||
if sig_broadcast: |
|
||||
tcp.broadcast_message(json_str) |
|
||||
|
|
||||
|
|
||||
def circle_show(img, circles, relative_point:models.target.Point): |
|
||||
font = cv2.FONT_HERSHEY_SIMPLEX |
|
||||
color = (255, 0, 0) # 蓝色 |
|
||||
scale = 0.5 |
|
||||
|
|
||||
circle = max(circles, key=lambda c: c[2]) |
|
||||
# print("画圆", circle) |
|
||||
# 绘制圆心 |
|
||||
center = (circle[0] + relative_point.x, circle[1] + relative_point.y) |
|
||||
center_int = tuple(int(x) for x in center) |
|
||||
cv2.circle(img, center_int, 2, (0, 255, 0), 4) |
|
||||
radius = np.round(circle[2], 3) |
|
||||
radius_int = int(radius) |
|
||||
# 绘制外圆 |
|
||||
cv2.circle(img, center_int, radius_int, (0, 0, 255), 2) |
|
||||
# 打印圆心坐标 |
|
||||
|
|
||||
text1 = f"center:{circle}" |
|
||||
text2 = f"r:{radius}" |
|
||||
txt_location = (center_int[0] + radius_int, center_int[1] + radius_int // 2) |
|
||||
cv2.putText(img, text1, txt_location, font, scale, color, 2) |
|
||||
|
|
||||
cp=models.target.Point(x=center[0], y=center[1]) |
|
||||
return cp,radius |
|
||||
|
|
||||
|
|
||||
def circle2_detect(img): |
|
||||
# 圆心距 canny阈值 最小半径 最大半径 |
|
||||
circles_float = cv2.HoughCircles(img, cv2.HOUGH_GRADIENT_ALT, 1.5, 30, param1=300, param2=0.9, minRadius=15, |
|
||||
maxRadius=0) |
|
||||
# 创建一个0行, 2列的空数组 |
|
||||
if circles_float is not None: |
|
||||
num_circles = circles_float.shape[1] # 获取检测到的圆的数量 |
|
||||
print("圆的数量", num_circles) |
|
||||
# 提取圆心坐标（保留2位小数） |
|
||||
centers = [(round(float(x),2), round(float(y),2), round(float(r),2)) for x, y, r in circles_float[0, :]] |
|
||||
return centers |
|
||||
else: |
|
||||
return [] |
|
||||
|
|
||||
|
|
||||
def extract_sub_image(frame, top_left, bottom_right): |
|
||||
""" |
|
||||
从帧中截取子区域 |
|
||||
:param frame: 输入的视频帧 |
|
||||
:param top_left: 子图片的左上角坐标 (x1, y1) |
|
||||
:param bottom_right: 子图片的右下角坐标 (x2, y2) |
|
||||
:return: 截取的子图片 |
|
||||
""" |
|
||||
x1, y1 = top_left |
|
||||
x2, y2 = bottom_right |
|
||||
return frame[y1:y2, x1:x2] |
|
||||
|
|
||||
|
|
||||
def open_video(video_id): |
|
||||
cap = cv2.VideoCapture(video_id) |
|
||||
# cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1600) # 宽度 |
|
||||
# cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 900) # 高度 |
|
||||
if not cap.isOpened(): |
|
||||
print("无法打开摄像头") |
|
||||
exit() |
|
||||
return cap |
|
||||
|
|
||||
|
|
||||
def show_video(cap): |
|
||||
global sigExit,start_point, end_point, drawing |
|
||||
cv2.namedWindow('Frame') |
|
||||
cv2.setMouseCallback('Frame', add_rectangle) |
|
||||
# 读取一帧图像 |
|
||||
while True: |
|
||||
ret, frame = cap.read() |
|
||||
if ret: |
|
||||
frame_handle(frame) |
|
||||
else: |
|
||||
print("无法读取帧") |
|
||||
if cv2.waitKey(1) & 0xFF == ord('q'): # 按'q'退出循环 |
|
||||
break |
|
||||
if sigExit: |
|
||||
break |
|
||||
def show_image(frame): |
|
||||
global start_point, end_point, drawing |
|
||||
cv2.namedWindow('Frame') |
|
||||
cv2.setMouseCallback('Frame', add_rectangle) |
|
||||
# 读取一帧图像 |
|
||||
while True: |
|
||||
cp_img=frame.copy() |
|
||||
frame_handle(cp_img) |
|
||||
|
|
||||
if cv2.waitKey(1) & 0xFF == ord('q'): # 按'q'退出循环 |
|
||||
break |
|
||||
cv2.destroyAllWindows() |
|
||||
|
|
||||
def frame_handle(frame): |
|
||||
# 绘图-历史点 |
|
||||
draw_rectangle(frame) |
|
||||
# 绘图-实时 |
|
||||
if drawing: |
|
||||
cv2.rectangle(frame, tuple(start_point), tuple(end_point), (0, 200, 200), 4) |
|
||||
# print(f"鼠标位置 {start_point} -> {end_point}") |
|
||||
# 显示图像 |
|
||||
cv2.imshow('Frame', frame) |
|
||||
|
|
||||
# 读取图像 |
|
||||
#img_copy = img.copy() # 复制图像用于还原 |
|
||||
|
|
||||
def video_mode(video_id): |
|
||||
capture = open_video(video_id) |
|
||||
fps = capture.get(cv2.CAP_PROP_FPS) |
|
||||
print(f"fps={fps}") |
|
||||
show_video(capture) |
|
||||
# 释放摄像头资源并关闭所有窗口 |
|
||||
capture.release() |
|
||||
cv2.destroyAllWindows() |
|
||||
|
|
||||
def image_mode(): |
|
||||
img_raw=cv2.imread('images/trans/_4point.jpg')#images/target/rp80max3.jpg |
|
||||
# img_raw = cv2.imread('images/trans/_4point.jpg') # images/target/rp80max3.jpg |
|
||||
# img_raw = cv2.imread('images/target/rp80.jpg') # images/target/rp80max3.jpg |
|
||||
show_image(img_raw) |
|
||||
|
|
||||
def rtsp_mode(): |
|
||||
# rtsp_url ="rtsp://admin:123456abc@192.168.1.64:554" |
|
||||
rtsp_url ="rtsp://localhost:8554/rtsp" |
|
||||
|
|
||||
capture = open_video(rtsp_url) |
|
||||
fps = capture.get(cv2.CAP_PROP_FPS) |
|
||||
print(f"fps={fps}") |
|
||||
show_video(capture) |
|
||||
# 释放摄像头资源并关闭所有窗口 |
|
||||
capture.release() |
|
||||
cv2.destroyAllWindows() |
|
||||
|
|
||||
if __name__ == '__main__': |
|
||||
signal.signal(signal.SIGINT, check_exit) |
|
||||
|
|
||||
rtsp_mode() |
|
||||
# video_mode(0) |
|
||||
# image_mode() |
|
||||
|
|
||||
cv2.waitKey(0) |
|
||||
cv2.destroyAllWindows() |
|
||||
|
|
||||
|
|
||||
@ -1,30 +0,0 @@ |
|||||
|
|
||||
import matplotlib.pyplot as plt |
|
||||
import numpy as np |
|
||||
|
|
||||
|
|
||||
# 存储初始范围和当前比例 |
|
||||
initial_limits = {'x': None, 'y': None} |
|
||||
current_scale = 1.0 |
|
||||
fig, ax = plt.subplots() |
|
||||
x = np.linspace(0, 10, 100) |
|
||||
ax.plot(x, np.sin(x)) |
|
||||
def on_press(event): |
|
||||
if event.button == 3: # 右键按下 |
|
||||
initial_limits['x'] = ax.get_xlim() |
|
||||
initial_limits['y'] = ax.get_ylim() |
|
||||
|
|
||||
def on_release(event): |
|
||||
global current_scale |
|
||||
if event.button == 3 and initial_limits['x'] is not None: |
|
||||
new_xlim = ax.get_xlim() |
|
||||
scale_x = (initial_limits['x'][1] - initial_limits['x'][0]) / \ |
|
||||
(new_xlim[1] - new_xlim[0]) |
|
||||
print(f"X轴缩放比例: {scale_x:.2f}倍") |
|
||||
current_scale *= scale_x |
|
||||
print(f"累计总缩放: {current_scale:.2f}倍") |
|
||||
|
|
||||
if __name__ == '__main__': |
|
||||
fig.canvas.mpl_connect('button_press_event', on_press) |
|
||||
fig.canvas.mpl_connect('button_release_event', on_release) |
|
||||
plt.show() |
|
||||
@ -1,62 +0,0 @@ |
|||||
import cv2 |
|
||||
import numpy as np |
|
||||
import 标靶识别 as ie |
|
||||
|
|
||||
|
|
||||
def perspective_transformation(image): |
|
||||
if image is None: |
|
||||
print("Error: Unable to load image.") |
|
||||
exit() |
|
||||
# 定义源图像中的四个点 |
|
||||
src_points = np.float32([ |
|
||||
[4, 16], # 左上角 |
|
||||
[795, 14], # 右上角 |
|
||||
[1027, 736], # 右下角 |
|
||||
[181, 856] # 左下角 |
|
||||
]) |
|
||||
# 定义目标图像中的四个点 # 1050 * 900 |
|
||||
dst_points = np.float32([ |
|
||||
[4, 16], # 左上角 |
|
||||
[795, 14], # 右上角 |
|
||||
[795, 850], # 右下角 |
|
||||
[4, 850] # 左下角 |
|
||||
]) |
|
||||
|
|
||||
# 计算透视变换矩阵 |
|
||||
M = cv2.getPerspectiveTransform(src_points, dst_points) |
|
||||
print(f"原矩阵={M}") |
|
||||
# 逆矩阵 |
|
||||
inverse_matrix = np.linalg.inv(M) |
|
||||
print(f"逆矩阵={inverse_matrix}") |
|
||||
# 应用透视变换 |
|
||||
transformed_image = cv2.warpPerspective(image, M, (1050, 900)) |
|
||||
# 应用透视变换 |
|
||||
inv_transformed_image = cv2.warpPerspective(transformed_image, inverse_matrix, (1050, 900)) |
|
||||
|
|
||||
# 显示原始图像和变换后的图像 |
|
||||
cv2.imshow("Original Image", image) |
|
||||
cv2.imshow("Trans Image", transformed_image) |
|
||||
cv2.imshow("iTrans Image", inv_transformed_image) |
|
||||
cv2.waitKey(0) |
|
||||
cv2.destroyAllWindows() |
|
||||
|
|
||||
def sub_img(): |
|
||||
# 读取图像 |
|
||||
image = cv2.imread("images/target/need_trans.jpg") |
|
||||
if image is None: |
|
||||
print("Error: Unable to load image.") |
|
||||
exit() |
|
||||
# 1050 * 900 |
|
||||
startPoint = [550, 700] |
|
||||
endPoint = [1600, 1600] |
|
||||
# 绘制标靶区域 |
|
||||
# 检测 |
|
||||
subImg = ie.extract_sub_image(image, startPoint, endPoint) |
|
||||
# cv2.imshow("subImg", subImg) |
|
||||
# cv2.waitKey(0) |
|
||||
# cv2.destroyAllWindows() |
|
||||
return subImg |
|
||||
|
|
||||
if __name__ == '__main__': |
|
||||
img=sub_img() |
|
||||
perspective_transformation(img) |
|
||||
@ -1,149 +0,0 @@ |
|||||
import cv2 |
|
||||
import numpy as np |
|
||||
|
|
||||
# 全局变量 |
|
||||
points = [] # 存储选择的四个点 |
|
||||
img = None # 存储原始图像 |
|
||||
img_copy = None # 用于绘制的图像副本 |
|
||||
|
|
||||
|
|
||||
def mouse_callback(event, x, y, flags, param): |
|
||||
"""鼠标回调函数，用于选择四个点""" |
|
||||
global img_copy, points |
|
||||
|
|
||||
if event == cv2.EVENT_LBUTTONDOWN: |
|
||||
if len(points) < 4: |
|
||||
points.append((x, y)) |
|
||||
print(f"已选择点 {len(points)}: ({x}, {y})") |
|
||||
|
|
||||
# 在图像上绘制点 |
|
||||
cv2.circle(img_copy, (x, y), 5, (0, 255, 0), -1) |
|
||||
|
|
||||
# 如果已经选择了4个点，绘制连线 |
|
||||
if len(points) == 4: |
|
||||
# 按照上、右、下、左的顺序连接点 |
|
||||
for i in range(4): |
|
||||
cv2.line(img_copy, points[i], points[(i + 1) % 4], (0, 255, 0), 2) |
|
||||
|
|
||||
# 标记每个点的位置 |
|
||||
cv2.putText(img_copy, "Top", points[0], cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2) |
|
||||
cv2.putText(img_copy, "Right", points[1], cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2) |
|
||||
cv2.putText(img_copy, "Bottom", points[2], cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2) |
|
||||
cv2.putText(img_copy, "Left", points[3], cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2) |
|
||||
|
|
||||
cv2.imshow("Select Points", img_copy) |
|
||||
|
|
||||
|
|
||||
def perspective_transform(image, src_points): |
|
||||
"""执行透视变换""" |
|
||||
# 将点排序为上、右、下、左 |
|
||||
top, right, bottom, left = src_points |
|
||||
|
|
||||
# 计算新图像的宽度（取左右边的最大值） |
|
||||
width_a = np.linalg.norm(np.array(right) - np.array(left)) |
|
||||
width_b = np.linalg.norm(np.array(top) - np.array(bottom)) |
|
||||
max_width = max(int(width_a), int(width_b)) |
|
||||
|
|
||||
# 计算新图像的高度（取上下边的最大值） |
|
||||
height_a = np.linalg.norm(np.array(bottom) - np.array(top)) |
|
||||
height_b = np.linalg.norm(np.array(right) - np.array(left)) |
|
||||
max_height = max(int(height_a), int(height_b)) |
|
||||
|
|
||||
# 定义目标点 |
|
||||
dst = np.array([ |
|
||||
[0, 0], # 左上角 |
|
||||
[max_width - 1, 0], # 右上角 |
|
||||
[max_width - 1, max_height - 1], # 右下角 |
|
||||
[0, max_height - 1] # 左下角 |
|
||||
], dtype="float32") |
|
||||
|
|
||||
# 转换源点数组为numpy数组 |
|
||||
src = np.array([top, right, bottom, left], dtype="float32") |
|
||||
|
|
||||
# 计算透视变换矩阵 |
|
||||
M = cv2.getPerspectiveTransform(src, dst) |
|
||||
|
|
||||
# 执行透视变换 |
|
||||
warped = cv2.warpPerspective(image, M, (max_width, max_height)) |
|
||||
|
|
||||
return warped |
|
||||
|
|
||||
def k_perspective_transform(image, src_points): |
|
||||
"""执行透视变换""" |
|
||||
# 将点排序为上、右、下、左 |
|
||||
top, right, bottom, left = src_points |
|
||||
|
|
||||
# 计算新图像的宽度（取左右边的最大值） |
|
||||
sub_zy = np.array(right) - np.array(left) |
|
||||
sub_sx = np.array(top) - np.array(bottom) |
|
||||
|
|
||||
sub_x=sub_sx[0]/2 |
|
||||
print("x差值",sub_sx[0]) |
|
||||
|
|
||||
sub_y = sub_zy[1] / 2 |
|
||||
print("y差值", sub_sx[0]) |
|
||||
|
|
||||
# 定义目标点 |
|
||||
dst = np.array([ |
|
||||
[top[0]-sub_x, top[1]], # 左上角 |
|
||||
[right[0], right[1]- sub_y], # 左上角 |
|
||||
[bottom[0]+sub_x, bottom[1]], # 左上角 |
|
||||
[left[0] , left[1]+ sub_y], # 左上角 |
|
||||
], dtype="float32") |
|
||||
|
|
||||
# 转换源点数组为numpy数组 |
|
||||
src = np.array([top, right, bottom, left], dtype="float32") |
|
||||
|
|
||||
# 计算透视变换矩阵 |
|
||||
M = cv2.getPerspectiveTransform(src, dst) |
|
||||
|
|
||||
print(f"矩阵M={M}") |
|
||||
|
|
||||
# 执行透视变换 |
|
||||
warped = cv2.warpPerspective(image, M, (1050, 900)) |
|
||||
|
|
||||
return warped |
|
||||
|
|
||||
def main(): |
|
||||
global img, img_copy, points |
|
||||
|
|
||||
# 读取图像 |
|
||||
img = cv2.imread("images/trans/subRawImg.jpg") |
|
||||
if img is None: |
|
||||
print("无法加载图像，请检查路径是否正确") |
|
||||
return |
|
||||
|
|
||||
img_copy = img.copy() |
|
||||
|
|
||||
# 创建窗口并设置鼠标回调 |
|
||||
cv2.namedWindow("Select Points") |
|
||||
cv2.setMouseCallback("Select Points", mouse_callback) |
|
||||
|
|
||||
print("请按顺序点击选择四个点：上、右、下、左") |
|
||||
print("选择完成后按任意键继续...") |
|
||||
|
|
||||
while True: |
|
||||
cv2.imshow("Select Points", img_copy) |
|
||||
key = cv2.waitKey(1) & 0xFF |
|
||||
|
|
||||
# 如果已经选择了4个点，按任意键继续 |
|
||||
if len(points) == 4: |
|
||||
break |
|
||||
|
|
||||
# 执行透视变换 |
|
||||
warped = k_perspective_transform(img, points) |
|
||||
|
|
||||
# 显示结果 |
|
||||
cv2.imshow("Original Image", img) |
|
||||
cv2.imshow("Transformed", warped) |
|
||||
|
|
||||
output_path="images/trans/_4point.jpg" |
|
||||
cv2.imwrite(output_path, warped) |
|
||||
print(f"图像已保存到 {output_path}") |
|
||||
|
|
||||
cv2.waitKey(0) |
|
||||
cv2.destroyAllWindows() |
|
||||
|
|
||||
|
|
||||
if __name__ == "__main__": |
|
||||
main() |
|
||||
@ -1,126 +0,0 @@ |
|||||
import numpy as np |
|
||||
import matplotlib.pyplot as plt |
|
||||
from mpl_toolkits.mplot3d import Axes3D |
|
||||
import 透视变换 |
|
||||
import cv2 |
|
||||
def build_3d_rotation_matrix(elev, azim, roll): |
|
||||
"""生成基于elev/azim/roll的完整旋转矩阵""" |
|
||||
elev_rad = np.radians(elev) |
|
||||
azim_rad = np.radians(azim) |
|
||||
roll_rad = np.radians(roll) |
|
||||
|
|
||||
# 绕x轴旋转(elevation) |
|
||||
Rx = np.array([ |
|
||||
[1, 0, 0], |
|
||||
[0, np.cos(elev_rad), -np.sin(elev_rad)], |
|
||||
[0, np.sin(elev_rad), np.cos(elev_rad)] |
|
||||
]) |
|
||||
|
|
||||
# 绕y轴旋转(azimuth) |
|
||||
Ry = np.array([ |
|
||||
[np.cos(azim_rad), 0, np.sin(azim_rad)], |
|
||||
[0, 1, 0], |
|
||||
[-np.sin(azim_rad), 0, np.cos(azim_rad)] |
|
||||
]) |
|
||||
|
|
||||
# 绕z轴旋转(roll) |
|
||||
Rz = np.array([ |
|
||||
[np.cos(roll_rad), -np.sin(roll_rad), 0], |
|
||||
[np.sin(roll_rad), np.cos(roll_rad), 0], |
|
||||
[0, 0, 1] |
|
||||
]) |
|
||||
|
|
||||
return Rz @ Ry @ Rx # 组合旋转顺序 |
|
||||
|
|
||||
|
|
||||
|
|
||||
if __name__ == '__main__': |
|
||||
# 参数设置 |
|
||||
elev, azim, roll = 34, 0,-35 |
|
||||
rotation_3d = build_3d_rotation_matrix(elev, azim, roll) |
|
||||
|
|
||||
# 创建单位圆 |
|
||||
theta = np.linspace(0, 2 * np.pi, 100) |
|
||||
x = np.cos(theta) |
|
||||
y = np.sin(theta) |
|
||||
z = np.zeros_like(x) |
|
||||
circle = np.vstack((x, y, z)) # 3xN |
|
||||
|
|
||||
circle_transformed = rotation_3d @ circle |
|
||||
# 逆矩阵 |
|
||||
inverse_matrix = np.linalg.inv(rotation_3d) |
|
||||
# 画图 |
|
||||
fig = plt.figure(figsize=(12, 6)) |
|
||||
# 设置窗口透明度 |
|
||||
fig.canvas.manager.window.attributes('-alpha', 0.6) # 设置窗口透明度为 0.6 |
|
||||
ax1 = fig.add_subplot(121, projection='3d') |
|
||||
ax2 = fig.add_subplot(122, projection='3d') |
|
||||
# 标识圆心 |
|
||||
ax1.scatter(0, 0, 0, color='red', s=20, label='Center') |
|
||||
# 原始单位圆 |
|
||||
ax1.plot(circle[0], circle[1], circle[2], label='Original Circle') |
|
||||
# 原始单位圆 横轴和纵轴 |
|
||||
ax1.plot(circle[0], np.zeros_like(theta), circle[2], label='z View', color='r') |
|
||||
ax1.plot(np.zeros_like(theta), circle[1], circle[2], label='h View', color='b') |
|
||||
ax1.set_title('Original Circle (elev=90°, roll=0°)') |
|
||||
ax1.set_xlim([-1, 1]) |
|
||||
ax1.set_ylim([-1, 1]) |
|
||||
ax1.set_zlim([-1, 1]) |
|
||||
ax1.set_box_aspect([1, 1, 1]) |
|
||||
ax1.set_xlabel('X') |
|
||||
ax1.set_ylabel('Y') |
|
||||
ax1.set_zlabel('Z') |
|
||||
ax1.view_init(elev=90, azim=90) # 从Z轴方向观察 保持opencv方向一致 x->左 y->下 |
|
||||
# 变换后的单位圆 |
|
||||
# 标识圆心 |
|
||||
ax2.scatter(0, 0, 0, color='red', s=20, label='Center') |
|
||||
ax2.plot(circle_transformed[0], circle_transformed[1], circle_transformed[2], color='r', label='Transformed Circle') |
|
||||
ax2.set_title('Transformed (elev=52°, roll=-35°)') |
|
||||
ax2.set_xlim([-1, 1]) |
|
||||
ax2.set_ylim([-1, 1]) |
|
||||
ax2.set_zlim([-1, 1]) |
|
||||
ax2.set_xlabel('X') |
|
||||
ax2.set_ylabel('Y') |
|
||||
ax2.set_zlabel('Z') |
|
||||
ax2.set_box_aspect([1, 1, 1]) |
|
||||
ax2.view_init(elev=90, azim=90) # 从Z轴方向观察 |
|
||||
plt.show() |
|
||||
|
|
||||
image_zhen= cv2.imread("images/trans/transformed_image.jpg") |
|
||||
image_xie = cv2.imread("images/trans/subRawImg.jpg") |
|
||||
# transformed_image_hy = cv2.warpPerspective(transformed_image, inverse_matrix, dsize=(image.shape[1], image.shape[0])) |
|
||||
# # 执行变换（自动计算输出图像尺寸） |
|
||||
# 裁剪像素值到 [0, 255] 范围 |
|
||||
|
|
||||
# 获取图像的大小 |
|
||||
height, width = image_zhen.shape[:2] |
|
||||
# 计算中心点坐标 |
|
||||
center_x = width // 2 |
|
||||
center_y = height // 2 |
|
||||
# 构造一个平移矩阵，将原点移到中心 |
|
||||
M_translate = np.float32([ |
|
||||
[1, 0, -1*center_x], |
|
||||
[0, 1, -1*center_y], |
|
||||
[0, 0, 1] |
|
||||
]) |
|
||||
image_xie_padding = cv2.warpPerspective(image_xie, M_translate, |
|
||||
dsize=(image_xie.shape[1], image_xie.shape[0])) |
|
||||
cv2.imshow("image_xie_padding", image_xie_padding) |
|
||||
# 将平移矩阵与目标变换矩阵结合起来 |
|
||||
# inverse_M_combined = np.dot(M_translate, inverse_matrix) |
|
||||
inverse_matrix[2][2]=1.0 |
|
||||
inverse_M_combined = np.dot(M_translate, inverse_matrix) |
|
||||
print(f"斜矩阵={rotation_3d}") |
|
||||
rotation_3d_int = np.clip(rotation_3d, 0, 255) |
|
||||
print(f"斜矩阵_int={rotation_3d}") |
|
||||
print(f"逆矩阵={inverse_M_combined}") |
|
||||
inverse_M_combined_int = np.clip(inverse_M_combined, 0, 255) |
|
||||
# print(f"逆矩阵_int={inverse_M_combined_int}") |
|
||||
transformed_image_hy = cv2.warpPerspective(image_xie, inverse_M_combined_int, |
|
||||
dsize=(image_xie.shape[1]*2, image_xie.shape[0]*2)) |
|
||||
cv2.imshow("transformed_image_hy", transformed_image_hy) |
|
||||
# transformed_image = cv2.warpPerspective(image_zhen, rotation_3d_int,dsize=(image_zhen.shape[1], image_zhen.shape[0])) |
|
||||
# cv2.imshow("rotated_img", transformed_image) |
|
||||
plt.show() |
|
||||
cv2.waitKey(0) |
|
||||
# cv2.destroyAllWindows() |
|
||||
@ -1,163 +0,0 @@ |
|||||
import cv2 |
|
||||
import numpy as np |
|
||||
from math import cos, sin, radians |
|
||||
|
|
||||
|
|
||||
def get_rotation_matrix(angle_x, angle_y, angle_z): |
|
||||
"""生成3D旋转矩阵""" |
|
||||
# 转换为弧度 |
|
||||
rx = radians(angle_x) |
|
||||
ry = radians(angle_y) |
|
||||
rz = radians(angle_z) |
|
||||
|
|
||||
# X轴旋转矩阵 |
|
||||
mat_x = np.array([ |
|
||||
[1, 0, 0], |
|
||||
[0, cos(rx), -sin(rx)], |
|
||||
[0, sin(rx), cos(rx)] |
|
||||
]) |
|
||||
|
|
||||
# Y轴旋转矩阵 |
|
||||
mat_y = np.array([ |
|
||||
[cos(ry), 0, sin(ry)], |
|
||||
[0, 1, 0], |
|
||||
[-sin(ry), 0, cos(ry)] |
|
||||
]) |
|
||||
|
|
||||
# Z轴旋转矩阵 |
|
||||
mat_z = np.array([ |
|
||||
[cos(rz), -sin(rz), 0], |
|
||||
[sin(rz), cos(rz), 0], |
|
||||
[0, 0, 1] |
|
||||
]) |
|
||||
|
|
||||
# 组合旋转矩阵 |
|
||||
rotation_matrix = np.dot(np.dot(mat_x, mat_y), mat_z) |
|
||||
return rotation_matrix |
|
||||
|
|
||||
|
|
||||
def perspective_transform(image, angle_x=0, angle_y=0, angle_z=0, scale=1.0): |
|
||||
"""应用透视变换""" |
|
||||
h, w = image.shape[:2] |
|
||||
|
|
||||
# 获取旋转矩阵 |
|
||||
rotation_matrix = get_rotation_matrix(angle_x, angle_y, angle_z) |
|
||||
|
|
||||
# 创建3D点到2D点的映射 |
|
||||
# 将2D图像视为3D空间中Z=0平面上的物体 |
|
||||
points_3d = np.array([ |
|
||||
[0, 0, 0], # 左上 |
|
||||
[w, 0, 0], # 右上 |
|
||||
[w, h, 0], # 右下 |
|
||||
[0, h, 0] # 左下 |
|
||||
], dtype=np.float32) |
|
||||
|
|
||||
# 应用旋转 |
|
||||
points_3d_rotated = np.dot(points_3d, rotation_matrix.T) |
|
||||
|
|
||||
# 添加透视效果 - 这里简单地将Z坐标作为深度 |
|
||||
# 可以调整这个值来改变透视强度 |
|
||||
points_2d_homo = points_3d_rotated[:, :2] / (scale - points_3d_rotated[:, 2:3] * 0.001) |
|
||||
|
|
||||
# 计算变换矩阵 |
|
||||
src_points = np.array([[0, 0], [w, 0], [w, h], [0, h]], dtype=np.float32) |
|
||||
dst_points = points_2d_homo.astype(np.float32) |
|
||||
|
|
||||
# 计算中心偏移 |
|
||||
min_xy = dst_points.min(axis=0) |
|
||||
max_xy = dst_points.max(axis=0) |
|
||||
dst_points -= min_xy |
|
||||
|
|
||||
# 计算新图像大小 |
|
||||
new_w = int(max_xy[0] - min_xy[0]) |
|
||||
new_h = int(max_xy[1] - min_xy[1]) |
|
||||
|
|
||||
# 获取透视变换矩阵 |
|
||||
M = cv2.getPerspectiveTransform(src_points, dst_points) |
|
||||
|
|
||||
# 应用变换 |
|
||||
transformed = cv2.warpPerspective(image, M, (new_w, new_h)) |
|
||||
|
|
||||
return transformed |
|
||||
|
|
||||
|
|
||||
def combined_perspective_transform(image, angle_x1, angle_y1, angle_z1, angle_y2, scale1=1.0, scale2=1.0): |
|
||||
"""合并两次变换：第一次任意旋转，第二次Y轴旋转""" |
|
||||
h, w = image.shape[:2] |
|
||||
|
|
||||
# 第一次旋转矩阵 |
|
||||
rot1 = get_rotation_matrix(angle_x1, angle_y1, angle_z1) |
|
||||
|
|
||||
# 第二次旋转矩阵 (Y轴旋转) |
|
||||
rot2 = get_rotation_matrix(0, angle_y2, 0) |
|
||||
|
|
||||
# 合并旋转矩阵 |
|
||||
combined_rot = np.dot(rot2, rot1) |
|
||||
|
|
||||
# 创建3D点到2D点的映射 |
|
||||
points_3d = np.array([ |
|
||||
[0, 0, 0], # 左上 |
|
||||
[w, 0, 0], # 右上 |
|
||||
[w, h, 0], # 右下 |
|
||||
[0, h, 0] # 左下 |
|
||||
], dtype=np.float32) |
|
||||
|
|
||||
# 应用合并后的旋转 |
|
||||
points_3d_rotated = np.dot(points_3d, combined_rot.T) |
|
||||
|
|
||||
# 添加透视效果 |
|
||||
points_2d_homo = points_3d_rotated[:, :2] / ((scale1 * scale2) - points_3d_rotated[:, 2:3] * 0.001) |
|
||||
|
|
||||
# 计算变换矩阵 |
|
||||
src_points = np.array([[0, 0], [w, 0], [w, h], [0, h]], dtype=np.float32) |
|
||||
dst_points = points_2d_homo.astype(np.float32) |
|
||||
|
|
||||
# 计算中心偏移 |
|
||||
min_xy = dst_points.min(axis=0) |
|
||||
max_xy = dst_points.max(axis=0) |
|
||||
dst_points -= min_xy |
|
||||
|
|
||||
# 计算新图像大小 |
|
||||
new_w = int(max_xy[0] - min_xy[0]) |
|
||||
new_h = int(max_xy[1] - min_xy[1]) |
|
||||
|
|
||||
# 获取透视变换矩阵 |
|
||||
M = cv2.getPerspectiveTransform(src_points, dst_points) |
|
||||
|
|
||||
# 应用变换 |
|
||||
transformed = cv2.warpPerspective(image, M, (new_w, new_h)) |
|
||||
|
|
||||
return transformed |
|
||||
|
|
||||
def show_transformed(): |
|
||||
image = cv2.imread('images/trans/transformed_image.jpg') |
|
||||
# 应用透视变换 |
|
||||
# 参数说明：angle_x, angle_y, angle_z 分别为绕X,Y,Z轴的旋转角度 |
|
||||
# scale 控制透视效果的强度 |
|
||||
transformed = perspective_transform(image, angle_x=34, angle_y=43, angle_z=-35, scale=1) |
|
||||
|
|
||||
# 显示结果 |
|
||||
cv2.imshow('Original', image) |
|
||||
cv2.imshow('Transformed', transformed) |
|
||||
cv2.waitKey(0) |
|
||||
cv2.destroyAllWindows() |
|
||||
def show_transformed_combined(): |
|
||||
image = cv2.imread('images/trans/transformed_image.jpg') |
|
||||
if image is None: |
|
||||
print("请替换为您的图片路径") |
|
||||
else: |
|
||||
# 第一次变换：任意角度 |
|
||||
# 第二次变换：Y轴旋转90度 |
|
||||
final_result = combined_perspective_transform( |
|
||||
image, |
|
||||
angle_x1=34, angle_y1=0, angle_z1=-35, # 第一次旋转参数 |
|
||||
angle_y2=43, # 第二次Y轴旋转90度 |
|
||||
scale1=1.2, scale2=1.0 # 透视参数 |
|
||||
) |
|
||||
|
|
||||
cv2.imshow('Original', image) |
|
||||
cv2.imshow('Final Result', final_result) |
|
||||
cv2.waitKey(0) |
|
||||
cv2.destroyAllWindows() |
|
||||
if __name__ == '__main__': |
|
||||
show_transformed_combined() |
|
||||
@ -1,86 +0,0 @@ |
|||||
import cv2 |
|
||||
import numpy as np |
|
||||
from cv2 import waitKey |
|
||||
|
|
||||
|
|
||||
imgRaw=cv2.imread('images/target/bowa_target/min.jpg',cv2.IMREAD_GRAYSCALE)#images/target/rp80max3.jpg |
|
||||
imgColor=cv2.imread('images/target/bowa_target/min.jpg') #images/target/rp80max3.jpg |
|
||||
|
|
||||
ret, binary_img = cv2.threshold(imgRaw, 180, 255, cv2.THRESH_BINARY) |
|
||||
#cv2.imshow('binary_img', binary_img) |
|
||||
def circle_detect(img): |
|
||||
#高斯滤波 |
|
||||
#img = cv2.GaussianBlur(img, (3, 3), 1) |
|
||||
#cv2.imshow('gsmh', gaussianBlur) |
|
||||
# 圆心距 canny阈值 最小半径 最大半径 |
|
||||
circlesFloat = cv2.HoughCircles(img, cv2.HOUGH_GRADIENT_ALT, 2, 10, param1=50, param2=0.9, minRadius=10, maxRadius=0) |
|
||||
print("==========") |
|
||||
# 创建一个0行, 2列的空数组 |
|
||||
if circlesFloat is not None: |
|
||||
num_circles = circlesFloat.shape[1] # 获取检测到的圆的数量 |
|
||||
print("圆的数量",num_circles) |
|
||||
# 提取圆心坐标（保留小数） |
|
||||
centers = [(float(x), float(y),float(r)) for x, y, r in circlesFloat[0, :]] |
|
||||
|
|
||||
font = cv2.FONT_HERSHEY_SIMPLEX |
|
||||
color = (255, 0, 0) # 蓝色 |
|
||||
scale = 1 |
|
||||
# 打印圆心坐标 |
|
||||
for center3d in centers: |
|
||||
center=(center3d[0], center3d[1]) |
|
||||
# center_txt=f"{float(center[0]),float(center[1])}" |
|
||||
text=f"center:{center},r:{center3d[2]}" |
|
||||
print(text) |
|
||||
centerInt=tuple(int(x) for x in center) |
|
||||
cv2.putText(img, text, centerInt, font, scale, color,2) |
|
||||
|
|
||||
circles = np.uint16(np.around(circlesFloat)) # 4舍5入, 然后转为uint16 |
|
||||
for i in circles[0, :]: |
|
||||
print("画圆", i) |
|
||||
# 绘制圆心 |
|
||||
center=(i[0], i[1]) |
|
||||
cv2.circle(img, center, 2, (0, 255, 0), 6) |
|
||||
# 绘制外圆 |
|
||||
cv2.circle(img, center, i[2], (0, 0, 255), 2) |
|
||||
|
|
||||
def circle_detect2(img): |
|
||||
#高斯滤波 |
|
||||
#img = cv2.GaussianBlur(img, (3, 3), 1) |
|
||||
#cv2.imshow('gsmh', gaussianBlur) |
|
||||
# 圆心距 canny阈值 最小半径 最大半径 |
|
||||
circlesFloat = cv2.HoughCircles(img, cv2.HOUGH_GRADIENT_ALT, 2, 10, param1=50, param2=0.9, minRadius=10, maxRadius=0) |
|
||||
print("==========") |
|
||||
# 创建一个0行, 2列的空数组 |
|
||||
if circlesFloat is not None: |
|
||||
num_circles = circlesFloat.shape[1] # 获取检测到的圆的数量 |
|
||||
print("圆的数量",num_circles) |
|
||||
# 提取圆心坐标（保留小数） |
|
||||
centers = [(float(x), float(y),float(r)) for x, y, r in circlesFloat[0, :]] |
|
||||
|
|
||||
font = cv2.FONT_HERSHEY_SIMPLEX |
|
||||
color = (255, 0, 0) # 蓝色 |
|
||||
scale = 1 |
|
||||
# 打印圆心坐标 |
|
||||
for center3d in centers: |
|
||||
center=(center3d[0], center3d[1]) |
|
||||
# center_txt=f"{float(center[0]),float(center[1])}" |
|
||||
text=f"center:{center},r:{center3d[2]}" |
|
||||
print(text) |
|
||||
centerInt=tuple(int(x) for x in center) |
|
||||
cv2.putText(img, text, centerInt, font, scale, color,2) |
|
||||
|
|
||||
circles = np.uint16(np.around(circlesFloat)) # 4舍5入, 然后转为uint16 |
|
||||
for i in circles[0, :]: |
|
||||
print("画圆", i) |
|
||||
# 绘制圆心 |
|
||||
center=(i[0], i[1]) |
|
||||
cv2.circle(img, center, 2, (0, 255, 0), 6) |
|
||||
# 绘制外圆 |
|
||||
cv2.circle(img, center, i[2], (0, 0, 255), 2) |
|
||||
|
|
||||
|
|
||||
if __name__ == '__main__': |
|
||||
circle_detect(binary_img) |
|
||||
cv2.imshow('tagCircle', imgColor) |
|
||||
waitKey(0) |
|
||||
cv2.destroyAllWindows() |
|
||||
Loading…
Reference in new issue