详细解释total_up_1 = np.array([0, 130, 80]) total_down_1 = np.array([179, 255, 255]) total_mask_1 = cv2.inRange(hsv_roi, total_up_1, total_down_1)

total_word_feature_extractor_zh.dat

total_word_feature_extractor_zh.dat total_word_feature_extractor_zh.dat total_word_feature_extractor_zh.dat total_word_feature_extractor_zh.dat

Aspose.Total_for_.NET_2017.2.28.part1

Total_Control_7_0_0_u10_Install_x64.zip

"Total_Control_7_0_0_u10_Install_x64.zip" 提供的正是这样一款解决方案，它是一款强大的手机与电脑投屏及控制软件。下面，我们将深入探讨这款软件的功能特性及其应用。首先，"Total Control" 是其名称，意指用户...

total_up_1 = np.array([0, 130, 80]) total_down_1 = np.array([179, 255, 255]) total_mask_1 = cv2.inRange(hsv_roi, total_up_1, total_down_1)

具体来说，它使用了一个 HSV 颜色空间的范围（total_up_1 和 total_down_1）来定义目标颜色的范围，并将这个范围内的像素设为白色，其余像素设为黑色。这个掩码可以用于随后的图像处理和分析操作，例如在图像中找到...

if name == "main": # Heurisic: suboptimal, have no notion of balance. env = BipedalWalker() env.reset() steps = 0 total_reward = 0 a = np.array([0.0, 0.0, 0.0, 0.0]) STAY_ON_ONE_LEG, PUT_OTHER_DOWN, PUSH_OFF = 1, 2, 3 SPEED = 0.29 # Will fall forward on higher speed state = STAY_ON_ONE_LEG moving_leg = 0 supporting_leg = 1 - moving_leg SUPPORT_KNEE_ANGLE = +0.1 supporting_knee_angle = SUPPORT_KNEE_ANGLE

6. a = np.array([0.0, 0.0, 0.0, 0.0])：定义小人的行动向量，初始值为0。 7. STAY_ON_ONE_LEG, PUT_OTHER_DOWN, PUSH_OFF = 1, 2, 3：定义小人的行动状态，分别为“保持单腿站立”、“支撑另一条腿”和“向前跳跃...

def read_data(pkl_path, loc): with open(pkl_path, 'rb') as file: data = pickle.load(file) pos_data, neg_data = [], [] for d in data: if d['label'] == 0: pos_data.append(d[loc]) else: neg_data.append(d[loc]) pos_data = torch.stack(pos_data).float() neg_data = torch.stack(neg_data).float() # assert pos_data.shape[0] == neg_data.shape[0] return pos_data, neg_data def read_data_all(pkl_path): with open(pkl_path, 'rb') as file: data = pickle.load(file) pos_data, neg_data = [], [] for d in data: if d['label'] == 0: pos_data.append(d) else: neg_data.append(d) return pos_data, neg_data def train_val_split(pos_data, neg_data, num_shot=10): total_num = pos_data.shape[0] train_num = int(num_shot) if num_shot != -1 else total_num val_num = total_num - train_num train_idx = random.sample(range(total_num), train_num) val_idx = list(set(range(total_num)) - set(train_idx)) x_train = torch.cat([pos_data[train_idx], neg_data[train_idx]], dim=0) y_train = torch.cat([torch.zeros(train_num), torch.ones(train_num)]) x_val = torch.cat([pos_data[val_idx], neg_data[val_idx]], dim=0) y_val = torch.cat([torch.zeros(val_num), torch.ones(val_num)]) return x_train, y_train, x_val, y_val def mix_data_fitting(x_train, y_train, model_name='lr'): model_list = [] for head in range(x_train.shape[1]): # 32 * 32 = 1024 if model_name == 'lda': model = LinearDiscriminantAnalysis() elif model_name == 'lr': model = LogisticRegression(penalty='l2') elif model_name == 'sgd': model = SGDClassifier(loss='log_loss') else: raise NotImplementedError model.fit(x_train[:, head], y_train) model_list.append(model) return model_list # def data_fitting_opt(x_train): # solutions = [] # pos_data = x_train[:len(x_train) // 2] # neg_data = x_train[len(x_train) // 2:] # for head in range(x_train.shape[1]): # def objective(x,u1,u2): # s12 = np.dot(np.transpose(u1-u2),u1-u2) # s1 = np.dot(np.transpose(u1),u1) # s2 = np.dot(np.transpose(u2),u2) # np12 = np.dot(np.dot(x,s12),np.transpose(x)) # np1 = np.sqrt(np.dot(np.dot(x,s1),np.transpose(x))) # np2 = np.sqrt(np.dot(np.dot(x,s2),np.transpose(x))) # down = np12 + (np1 + np2)2 # return -np12/down import numpy as np def objective(x, u1, u2): # Number of samples n = u1.shape[0] u1 = np.array(u1) u2 = np.array(u2) # Compute means mu_x = np.mean(u1, axis=0) # Shape: (128,) mu_y = np.mean(u2, axis=0) # Shape: (128,) # Numerator: |x^T (mu_x - mu_y)|^2 diff_mean = mu_x - mu_y numerator = (np.dot(x, diff_mean))2 #/ (n2) # Scalar # Compute covariance matrices with 1/(n-1) u1_centered = u1 - mu_x # Shape: (17, 128) u2_centered = u2 - mu_y # Shape: (17, 128) Sigma_x = np.dot(u1_centered.T, u1_centered) #/ (n - 1) # Shape: (128, 128) Sigma_y = np.dot(u2_centered.T, u2_centered) #/ (n - 1) # Shape: (128, 128),样本数较少时（1%） # Compute denominator terms s_x = np.dot(np.dot(x, Sigma_x), x) # x^T Sigma_x x, scalar s_y = np.dot(np.dot(x, Sigma_y), x) # x^T Sigma_y x, scalar denominator = (np.sqrt(s_x) + np.sqrt(s_y))2 # Compute alpha, avoiding division by zero alpha = numerator / (numerator + denominator) # if (numerator + denominator) > 0 else 0 return -alpha def model_opt(pos, neg): import scipy.optimize as opt u1 = pos # Shape (n, 128) u2 = neg # Shape (n, 128) x0 = np.random.random(u1.shape[1]) # (128,) result = opt.minimize(objective, x0, args=(u1, u2)) return result.x, -result.fun 这是第二段代码

- 遍历数据项，根据d['label']值将d[loc]存入pos_data(标签0)或neg_data(标签1) - 将列表转换为PyTorch张量 - 示例输入： python pos, neg = read_data("data.pkl", loc='feature') loc='...

class KeyWordSpotter(torch.nn.Module): def init( self, ckpt_path, config_path, token_path, lexicon_path, threshold, min_frames=5, max_frames=250, interval_frames=50, score_beam=3, path_beam=20, gpu=-1, is_jit_model=False, ): super().init() os.environ['CUDA_VISIBLE_DEVICES'] = str(gpu) with open(config_path, 'r') as fin: configs = yaml.load(fin, Loader=yaml.FullLoader) dataset_conf = configs['dataset_conf'] # feature related self.sample_rate = 16000 self.wave_remained = np.array([]) self.num_mel_bins = dataset_conf['feature_extraction_conf'][ 'num_mel_bins'] self.frame_length = dataset_conf['feature_extraction_conf'][ 'frame_length'] # in ms self.frame_shift = dataset_conf['feature_extraction_conf'][ 'frame_shift'] # in ms self.downsampling = dataset_conf.get('frame_skip', 1) self.resolution = self.frame_shift / 1000 # in second # fsmn splice operation self.context_expansion = dataset_conf.get('context_expansion', False) self.left_context = 0 self.right_context = 0 if self.context_expansion: self.left_context = dataset_conf['context_expansion_conf']['left'] self.right_context = dataset_conf['context_expansion_conf'][ 'right'] self.feature_remained = None self.feats_ctx_offset = 0 # after downsample, offset exist. # model related if is_jit_model: model = torch.jit.load(ckpt_path) # For script model, only cpu is supported. device = torch.device('cpu') else: # Init model from configs model = init_model(configs['model']) load_checkpoint(model, ckpt_path) use_cuda = gpu >= 0 and torch.cuda.is_available() device = torch.device('cuda' if use_cuda else 'cpu') self.device = device self.model = model.to(device) self.model.eval() logging.info(f'model {ckpt_path} loaded.') self.token_table = read_token(token_path) logging.info(f'tokens {token_path} with ' f'{len(self.token_table)} units loaded.') self.lexicon_table = read_lexicon(lexicon_path) logging.info(f'lexicons {lexicon_path} with ' f'{len(self.lexicon_table)} units loaded.') self.in_cache = torch.zeros(0, 0, 0, dtype=torch.float) # decoding and detection related self.score_beam = score_beam self.path_beam = path_beam self.threshold = threshold self.min_frames = min_frames self.max_frames = max_frames self.interval_frames = interval_frames self.cur_hyps = [(tuple(), (1.0, 0.0, []))] self.hit_score = 1.0 self.hit_keyword = None self.activated = False self.total_frames = 0 # frame offset, for absolute time self.last_active_pos = -1 # the last frame of being activated self.result = {} def set_keywords(self, keywords): # 4. parse keywords tokens assert keywords is not None, \ 'at least one keyword is needed, ' \ 'multiple keywords should be splitted with comma(,)' keywords_str = keywords keywords_list = keywords_str.strip().replace(' ', '').split(',') keywords_token = {} keywords_idxset = {0} keywords_strset = {'<blk>'} keywords_tokenmap = {'<blk>': 0} for keyword in keywords_list: strs, indexes = query_token_set(keyword, self.token_table, self.lexicon_table) keywords_token[keyword] = {} keywords_token[keyword]['token_id'] = indexes keywords_token[keyword]['token_str'] = ''.join('%s ' % str(i) for i in indexes) [keywords_strset.add(i) for i in strs] [keywords_idxset.add(i) for i in indexes] for txt, idx in zip(strs, indexes): if keywords_tokenmap.get(txt, None) is None: keywords_tokenmap[txt] = idx token_print = '' for txt, idx in keywords_tokenmap.items(): token_print += f'{txt}({idx}) ' logging.info(f'Token set is: {token_print}') self.keywords_idxset = keywords_idxset self.keywords_token = keywords_token def accept_wave(self, wave): assert isinstance(wave, bytes), \ "please make sure the input format is bytes(raw PCM)" # convert bytes into float32 data = [] for i in range(0, len(wave), 2): value = struct.unpack('<h', wave[i:i + 2])[0] data.append(value) # here we don't divide 32768.0, # because kaldi.fbank accept original input wave = np.array(data) wave = np.append(self.wave_remained, wave) if wave.size < (self.frame_length * self.sample_rate / 1000) \ * self.right_context : self.wave_remained = wave return None wave_tensor = torch.from_numpy(wave).float().to(self.device) wave_tensor = wave_tensor.unsqueeze(0) # add a channel dimension feats = kaldi.fbank(wave_tensor, num_mel_bins=self.num_mel_bins, frame_length=self.frame_length, frame_shift=self.frame_shift, dither=0, energy_floor=0.0, sample_frequency=self.sample_rate) # update wave remained feat_len = len(feats) frame_shift = int(self.frame_shift / 1000 * self.sample_rate) self.wave_remained = wave[feat_len * frame_shift:] if self.context_expansion: assert feat_len > self.right_context, \ "make sure each chunk feat length is large than right context." # pad feats with remained feature from last chunk if self.feature_remained is None: # first chunk # pad first frame at the beginning, # replicate just support last dimension, so we do transpose. feats_pad = F.pad(feats.T, (self.left_context, 0), mode='replicate').T else: feats_pad = torch.cat((self.feature_remained, feats)) ctx_frm = feats_pad.shape[0] - (self.right_context + self.right_context) ctx_win = (self.left_context + self.right_context + 1) ctx_dim = feats.shape[1] * ctx_win feats_ctx = torch.zeros(ctx_frm, ctx_dim, dtype=torch.float32) for i in range(ctx_frm): feats_ctx[i] = torch.cat(tuple( feats_pad[i:i + ctx_win])).unsqueeze(0) # update feature remained, and feats self.feature_remained = \ feats[-(self.left_context + self.right_context):] feats = feats_ctx.to(self.device) if self.downsampling > 1: last_remainder = 0 if self.feats_ctx_offset == 0 \ else self.downsampling - self.feats_ctx_offset remainder = (feats.size(0) + last_remainder) % self.downsampling feats = feats[self.feats_ctx_offset::self.downsampling, :] self.feats_ctx_offset = remainder \ if remainder == 0 else self.downsampling - remainder return feats def decode_keywords(self, t, probs): absolute_time = t + self.total_frames # search next_hyps depend on current probs and hyps. next_hyps = ctc_prefix_beam_search(absolute_time, probs, self.cur_hyps, self.keywords_idxset, self.score_beam) # update cur_hyps. note: the hyps is sort by path score(pnb+pb), # not the keywords' probabilities. cur_hyps = next_hyps[:self.path_beam] self.cur_hyps = cur_hyps def execute_detection(self, t): absolute_time = t + self.total_frames hit_keyword = None start = 0 end = 0 # hyps for detection hyps = [(y[0], y[1][0] + y[1][1], y[1][2]) for y in self.cur_hyps] # detect keywords in decoding paths. for one_hyp in hyps: prefix_ids = one_hyp[0] # path_score = one_hyp[1] prefix_nodes = one_hyp[2] assert len(prefix_ids) == len(prefix_nodes) for word in self.keywords_token.keys(): lab = self.keywords_token[word]['token_id'] offset = is_sublist(prefix_ids, lab) if offset != -1: hit_keyword = word start = prefix_nodes[offset]['frame'] end = prefix_nodes[offset + len(lab) - 1]['frame'] for idx in range(offset, offset + len(lab)): self.hit_score = prefix_nodes[idx]['prob'] break if hit_keyword is not None: self.hit_score = math.sqrt(self.hit_score) break duration = end - start if hit_keyword is not None: if self.hit_score >= self.threshold and \ self.min_frames <= duration <= self.max_frames \ and (self.last_active_pos == -1 or end - self.last_active_pos >= self.interval_frames): self.activated = True self.last_active_pos = end logging.info( f"Frame {absolute_time} detect {hit_keyword} " f"from {start} to {end} frame. " f"duration {duration}, score {self.hit_score}, Activated.") elif self.last_active_pos > 0 and \ end - self.last_active_pos < self.interval_frames: logging.info( f"Frame {absolute_time} detect {hit_keyword} " f"from {start} to {end} frame. " f"but interval {end-self.last_active_pos} " f"is lower than {self.interval_frames}, Deactivated. ") elif self.hit_score < self.threshold: logging.info(f"Frame {absolute_time} detect {hit_keyword} " f"from {start} to {end} frame. " f"but {self.hit_score} " f"is lower than {self.threshold}, Deactivated. ") elif self.min_frames > duration or duration > self.max_frames: logging.info( f"Frame {absolute_time} detect {hit_keyword} " f"from {start} to {end} frame. " f"but {duration} beyond range" f"({self.min_frames}~{self.max_frames}), Deactivated. ") self.result = { "state": 1 if self.activated else 0, "keyword": hit_keyword if self.activated else None, "start": start self.resolution if self.activated else None, "end": end * self.resolution if self.activated else None, "score": self.hit_score if self.activated else None } def forward(self, wave_chunk): feature = self.accept_wave(wave_chunk) if feature is None or feature.size(0) < 1: return {} # # the feature is not enough to get result. feature = feature.unsqueeze(0) # add a batch dimension logits, self.in_cache = self.model(feature, self.in_cache) probs = logits.softmax(2) # (batch_size, maxlen, vocab_size) probs = probs[0].cpu() # remove batch dimension for (t, prob) in enumerate(probs): t = self.downsampling self.decode_keywords(t, prob) self.execute_detection(t) if self.activated: self.reset() # since a chunk include about 30 frames, # once activated, we can jump the latter frames. # TODO: there should give another method to update result, # avoiding self.result being cleared. break # update frame offset self.total_frames += len(probs) self.downsampling # For streaming kws, the cur_hyps should be reset if the time of # a possible keyword last over the max_frames value you set. # see this issue:https://github.com/duj12/kws_demo/issues/2 if len(self.cur_hyps) > 0 and len(self.cur_hyps[0][0]) > 0: keyword_may_start = int(self.cur_hyps[0][1][2][0]['frame']) if (self.total_frames - keyword_may_start) > self.max_frames: self.reset() return self.result def reset(self): self.cur_hyps = [(tuple(), (1.0, 0.0, []))] self.activated = False self.hit_score = 1.0 def reset_all(self): self.reset() self.wave_remained = np.array([]) self.feature_remained = None self.feats_ctx_offset = 0 # after downsample, offset exist. self.in_cache = torch.zeros(0, 0, 0, dtype=torch.float) self.total_frames = 0 # frame offset, for absolute time self.last_active_pos = -1 # the last frame of being activated self.result = {}请帮我缕清整个脉络

x = x.permute(0, 2, 1) # 维度转换: (batch, time, channels) _, h_n = self.gru(x) # 获取最终隐藏状态 # 合并双向GRU输出 last_output = torch.cat((h_n[-2], h_n[-1]), dim=1) return self.fc(last_output)...

def predict_one_img(model, img_dataset, args): # def predict_one_img(model, img_dataset, data, target, args): dataloader = DataLoader(dataset=img_dataset, batch_size=1, num_workers=0, shuffle=False) model.eval() test_dice = DiceAverage(args.n_labels) # target = to_one_hot_3d(label, args.n_labels) with torch.no_grad(): for data ,target in tqdm(dataloader, total=len(dataloader)): # data = data.to(device) data, target = data.float(), target.long() target = to_one_hot_3d(target, args.n_labels) data, target = data.to(device), target.to(device) # print(data.shape) # print(target.shape) output = model(data) # output = nn.functional.interpolate(output, scale_factor=(1//args.slice_down_scale,1//args.xy_down_scale,1//args.xy_down_scale), mode='trilinear', align_corners=False) # 空间分辨率恢复到原始size img_dataset.update_result(output.detach().cpu()) pred = img_dataset.recompone_result() pred = torch.argmax(pred, dim=1) pred_img = to_one_hot_3d(pred, args.n_labels) pred_img=pred_img.to(device) test_dice.update(pred_img, target) test_dice = OrderedDict({'Dice_liver': test_dice.avg[1]}) if args.n_labels == 3: test_dice.update({'Dice_tumor': test_dice.avg[2]}) pred = np.asarray(pred.numpy(), dtype='uint8') if args.postprocess: pass # TO DO pred = sitk.GetImageFromArray(np.squeeze(pred, axis=0)) return test_dice, pred

这是一个用于预测单张图像的函数，接收一个模型、一个图像数据集和一些参数作为输入。函数首先将图像数据集加载到一个Dataloader中，然后将模型设置为评估模式并初始化一个DiceAverage对象，该对象用于计算评估指标...

class ResidualBlock(nn.Module): def init(self, in_channels, out_channels, dilation): super(ResidualBlock, self).init() self.conv = nn.Sequential( nn.Conv1d(in_channels, out_channels, kernel_size=3, padding=dilation, dilation=dilation), nn.BatchNorm1d(out_channels), nn.ReLU(), nn.Conv1d(out_channels, out_channels, kernel_size=3, padding=dilation, dilation=dilation), nn.BatchNorm1d(out_channels), nn.ReLU() ) self.attention = nn.Sequential( nn.Conv1d(out_channels, out_channels, kernel_size=1), nn.Sigmoid() ) self.downsample = nn.Conv1d(in_channels, out_channels, kernel_size=1) if in_channels != out_channels else None def forward(self, x): residual = x out = self.conv(x) attention = self.attention(out) out = out * attention if self.downsample: residual = self.downsample(residual) out += residual return out class VMD_TCN(nn.Module): def init(self, input_size, output_size, n_k=1, num_channels=16, dropout=0.2): super(VMD_TCN, self).init() self.input_size = input_size self.nk = n_k if isinstance(num_channels, int): num_channels = [num_channels(2i) for i in range(4)] self.layers = nn.ModuleList() self.layers.append(nn.utils.weight_norm(nn.Conv1d(input_size, num_channels[0], kernel_size=1))) for i in range(len(num_channels)): dilation_size = 2 i in_channels = num_channels[i-1] if i > 0 else num_channels[0] out_channels = num_channels[i] self.layers.append(ResidualBlock(in_channels, out_channels, dilation_size)) self.pool = nn.AdaptiveMaxPool1d(1) self.fc = nn.Linear(num_channels[-1], output_size) self.w = nn.Sequential(nn.Conv1d(num_channels[-1], num_channels[-1], kernel_size=1), nn.Sigmoid()) # 特征融合门控系统 # self.fc1 = nn.Linear(output_size (n_k + 1), output_size) # 全部融合 self.fc1 = nn.Linear(output_size * 2, output_size) # 只选择其中两个融合 self.dropout = nn.Dropout(dropout) # self.weight_fc = nn.Linear(num_channels[-1] * (n_k + 1), n_k + 1) # 置信度系数，对各个结果加权平均软投票思路 def vmd(self, x): x_imfs = [] signal = np.array(x).flatten() # flatten()必须加上否则最后一个batch报错size不匹配！ u, u_hat, omega = VMD(signal, alpha=512, tau=0, K=self.nk, DC=0, init=1, tol=1e-7) for i in range(u.shape[0]): imf = torch.tensor(u[i], dtype=torch.float32) imf = imf.reshape(-1, 1, self.input_size) x_imfs.append(imf) x_imfs.append(x) return x_imfs def forward(self, x): x_imfs = self.vmd(x) total_out = [] # for data in x_imfs: for data in [x_imfs[0], x_imfs[-1]]: out = data.transpose(1, 2) for layer in self.layers: out = layer(out) out = self.pool(out) # torch.Size([96, 56, 1]) w = self.w(out) out = w * out # torch.Size([96, 56, 1]) out = out.view(out.size(0), -1) out = self.dropout(out) out = self.fc(out) total_out.append(out) total_out = torch.cat(total_out, dim=1) # 考虑w1total_out[0]+ w2total_out[1],在第一维，权重相加得到最终结果，不用cat total_out = self.dropout(total_out) output = self.fc1(total_out) return output优化代码

in_channels = num_channels[i-1] if i > 0 else num_channels[0] out_channels = num_channels[i] self.layers.append(ResidualBlock(in_channels, out_channels, dilation_size)) 4. 不建议在forward()...

# data/dataset.py import os import numpy as np import torch from torch.utils.data import Dataset from plyfile import PlyData import open3d as o3d def read_ply_points(file_path): """ 使用 plyfile 解析 .ply 文件，提取 x, y, z 字段返回一个 numpy 数组 (N, 3) """ ply = PlyData.read(file_path) # 提取 x, y, z 字段 x = ply['vertex']['x'] y = ply['vertex']['y'] z = ply['vertex']['z'] # 拼接为点云数组 points = np.vstack([x, y, z]).T points = points.astype(np.float32) return points class UnifiedPointCloudDataset(Dataset): def init(self, root_dirs, file_exts=['.ply', '.stl', '.obj'], num_points=1024): """ :param root_dirs: 包含所有数据文件夹的列表 :param file_exts: 支持的点云格式 :param num_points: 每个点云采样点数 """ self.file_list = [] self.num_points = num_points # 收集所有点云文件路径 for root_dir in root_dirs: if not os.path.exists(root_dir): raise FileNotFoundError(f"❌ 数据目录不存在: {root_dir}") for root, _, files in os.walk(root_dir): for file in files: if any(file.lower().endswith(ext) for ext in file_exts): full_path = os.path.join(root, file) if os.path.exists(full_path): # 确保文件真实存在 self.file_list.append(full_path) print(f"✅ 共发现 {len(self.file_list)} 个点云文件，用于训练") if len(self.file_list) == 0: raise ValueError("⚠️ 没有发现任何点云文件，请检查路径和文件格式") def len(self): return len(self.file_list) def getitem(self, idx): path = self.file_list[idx] ext = os.path.splitext(path)[1].lower() try: if ext == '.ply': # 使用 plyfile 读取 .ply 文件 points = read_ply_points(path) elif ext == '.stl': # 使用 open3d 读取 STL 文件并采样成点云 mesh = o3d.io.read_triangle_mesh(path) pcd = mesh.sample_points_uniformly(number_of_points=100000) points = np.asarray(pcd.points) elif ext == '.obj': # 使用 open3d 读取 OBJ 文件 pcd = o3d.io.read_point_cloud(path) if not pcd.has_points(): raise ValueError(f"点云为空或损坏: {path}") points = np.asarray(pcd.points) else: raise ValueError(f"不支持的格式: {ext}") # 检查点云是否为空 if len(points) < 10: raise ValueError(f"点云为空或损坏: {path}") # 固定点数采样 if len(points) < self.num_points: indices = np.random.choice(len(points), self.num_points, replace=True) else: indices = np.random.choice(len(points), self.num_points, replace=False) points = points[indices] return torch.tensor(points, dtype=torch.float32) except Exception as e: print(f"❌ 读取失败: {path}，错误: {str(e)}") return self.getitem((idx + 1) % len(self.file_list)) # models/ballmamba.py import torch import torch.nn as nn from torch_geometric.nn import knn_graph, radius_graph from models.mamba_block import MambaBlock from utils.pointcloud_utils import farthest_point_sample class FPS_Causal(nn.Module): def init(self, in_channels, hidden_channels, k=512): super(FPS_Causal, self).init() self.k = k self.downsample = nn.Linear(in_channels, hidden_channels) self.mamba = MambaBlock(hidden_channels) def forward(self, x, pos): batch_size, num_points, _ = pos.shape idxs = [] for i in range(batch_size): idx = farthest_point_sample(pos[i], self.k) # 现在支持 (N, 3) idxs.append(idx) idx = torch.stack(idxs, dim=0) # (B, k) # 使用 gather 正确采样 x_sampled = torch.gather(pos, 1, idx.unsqueeze(-1).expand(-1, -1, 3)) # (B, k, 3) x_sampled = self.downsample(x_sampled) # (B, k, hidden_channels) x_sampled = self.mamba(x_sampled) # (B, k, hidden_channels) return x_sampled class BallQuery_Sort(nn.Module): def init(self, radius=0.3, k=64): super(BallQuery_Sort, self).init() self.radius = radius self.k = k def forward(self, pos, x): edge_index = radius_graph(pos, r=self.radius) row, col = edge_index neighbor_pos = pos[col] neighbor_x = x[col] dist = torch.norm(neighbor_pos - pos[row], dim=1) sorted_indices = torch.argsort(dist.view(-1, self.k), dim=1) neighbors = neighbor_x.view(-1, self.k, neighbor_x.shape[1]) return neighbors.gather(1, sorted_indices.unsqueeze(-1).expand(-1, -1, neighbor_x.shape[1])) class BallMambaModel(nn.Module): def init(self, in_channels=3, num_keypoints=1024): super(BallMambaModel, self).init() self.fps = FPS_Causal(in_channels, 64, k=num_keypoints) self.ball_query = BallQuery_Sort(radius=0.1, k=64) self.mamba = MambaBlock(64) self.decoder = nn.Linear(64, 3) def forward(self, pos): print("pos shape:", pos.shape) # 添加此行，查看 pos 的形状 x = self.fps(None, pos) x = self.ball_query(pos, x) x = x.mean(dim=1) x = self.mamba(x.unsqueeze(0)).squeeze(0) out = self.decoder(x) return out import torch import torch.nn as nn import torch.nn.functional as F class MambaBlock(nn.Module): def init(self, d_model: int, d_state: int = 16, d_conv: int = 4, expand: int = 2, dt_rank: int or str = "auto"): """ 完整的 MambaBlock 实现（无 LSTM，基于状态空间模型 SSM） :param d_model: 输入特征维度（通道数） :param d_state: 状态维度（SSM 中的 N） :param d_conv: 卷积核大小（用于局部依赖建模） :param expand: 扩展因子，中间维度 = d_model * expand :param dt_rank: 离散时间步秩，控制参数量，若为 "auto" 则自动计算 """ super(MambaBlock, self).init() self.d_model = d_model self.d_state = d_state self.d_inner = int(expand * d_model) self.dt_rank = dt_rank if dt_rank != "auto" else max(1, self.d_model // 16) # 输入投影：将输入特征升维 self.in_proj = nn.Linear(d_model, self.d_inner * 2, bias=False) # 卷积分支（局部建模） self.conv = nn.Conv1d( in_channels=self.d_inner, out_channels=self.d_inner, kernel_size=d_conv, bias=True, groups=self.d_inner, padding=d_conv - 1, # 保证 causal ) # x_proj 将 x 映射到 dt、B、C self.x_proj = nn.Linear(self.d_inner, self.dt_rank + 2 * d_state, bias=False) # dt_proj 将 dt 映射到 d_inner self.dt_proj = nn.Linear(self.dt_rank, self.d_inner, bias=True) # A 和 D 参数（状态矩阵和跳跃连接） A = torch.arange(1, d_state + 1, dtype=torch.float32)[None, :].repeat(self.d_inner, 1) self.A_log = nn.Parameter(torch.log(A)) self.D = nn.Parameter(torch.ones(self.d_inner)) # (d_inner, ) # 输出投影 self.out_proj = nn.Linear(self.d_inner, d_model, bias=False) def forward(self, x: torch.Tensor) -> torch.Tensor: """ 前向传播 :param x: 输入张量，形状 (B, L, d_model) :return: 输出张量，形状 (B, L, d_model) """ batch, seqlen, dim = x.shape # 1. 输入投影 + 拆分 xz = self.in_proj(x) # (B, L, 2 * d_inner) x, z = torch.split(xz, [self.d_inner, self.d_inner], dim=-1) # (B, L, d_inner) # 2. 卷积处理 x = x.transpose(1, 2) # (B, d_inner, L) x = self.conv(x)[:, :, :seqlen] # (B, d_inner, L) x = x.transpose(1, 2) # (B, L, d_inner) # 3. x_proj 分割出 dt、B、C x_dbl = self.x_proj(x) # (B, L, dt_rank + 2 * d_state) dt, B, C = torch.split( x_dbl, [self.dt_rank, self.d_state, self.d_state], dim=-1 ) # 4. dt_proj 映射 dt dt = F.softplus(self.dt_proj(dt)) # (B, L, d_inner) # 5. 获取 A A = -torch.exp(self.A_log) # (d_inner, d_state) A = A.unsqueeze(0).unsqueeze(0) # (1, 1, d_inner, d_state) B = B.unsqueeze(dim=2) # (B, L, 1, d_state) C = C.unsqueeze(dim=2) # (B, L, 1, d_state) # 调试信息 print(f"[MambaBlock] A.shape = {A.shape}") print(f"[MambaBlock] B.shape = {B.shape}") print(f"[MambaBlock] C.shape = {C.shape}") print(f"[MambaBlock] dt.shape = {dt.shape}") states = torch.zeros(batch, self.d_inner, self.d_state, device=x.device) outputs = [] for t in range(seqlen): # 更新状态 states = states + x[:, t:t+1, :, None] * B[:, t:t+1, :, :] states = states * torch.exp(A * dt[:, t:t+1, :, None]) # 添加 dt 到状态更新 # 获取当前时间步的 C 并进行 einsum current_C = C[:, t] # (B, 1, d_state) current_C = current_C.squeeze(1) # (B, d_state) # 使用广播机制 y = torch.einsum("binc,bc->bin", states, current_C) # bc 会广播为 binc outputs.append(y) y = torch.stack(outputs, dim=1) # (B, L, d_inner) # ✅ 修复：self.D 扩展为 (1, 1, d_inner) 以便广播 y = y + x * self.D.view(1, 1, -1) # 加上跳跃连接 # 激活 + 输出 y = y * F.silu(z) out = self.out_proj(y) # 调试信息 print(f"[MambaBlock] y.shape = {y.shape}") print(f"[MambaBlock] out.shape = {out.shape}") return out # train.py import os import torch import torch.nn as nn from torch.utils.data import DataLoader from data.dataset import UnifiedPointCloudDataset from models.ballmamba import BallMambaModel from utils.loss import ChamferLoss # 👇 Windows 多进程训练必须放在 if name == 'main' 里面 if name == 'main': # 设置多进程启动方式（Windows 下推荐 spawn） torch.multiprocessing.set_start_method('spawn') # ✅ 修改为你自己的路径 ROOT_DIRS = [ r"D:\桌面\point\data1\part1", r"D:\桌面\point\data1\part2", r"D:\桌面\point\data1\part3", r"D:\桌面\point\data1\part4", r"D:\桌面\point\data1\part5", r"D:\桌面\point\data1\part6", r"D:\桌面\point\data1\part7", r"D:\桌面\point\data1\part8", r"D:\桌面\point\data1\part9", r"D:\桌面\point\data1\part10", r"D:\桌面\point\data1\part11", r"D:\桌面\point\data1\part12", r"D:\桌面\point\data1\part13", r"D:\桌面\point\data1\part14", r"D:\桌面\point\data1\part15", r"D:\桌面\point\data1\part16", r"D:\桌面\point\data1\part17", r"D:\桌面\point\data1\part18", r"D:\桌面\point\data1\part19", r"D:\桌面\point\data1\part20", ] # ✅ 创建 Dataset dataset = UnifiedPointCloudDataset( root_dirs=ROOT_DIRS, file_exts=['.ply', '.stl'], num_points=1024 ) print(f"✅ 共发现 {len(dataset)} 个点云文件，用于训练") if len(dataset) == 0: raise ValueError("⚠️ 没有发现任何点云文件，请检查路径和文件格式") # ✅ 创建 DataLoader（num_workers=0 可临时绕过问题） loader = DataLoader( dataset, batch_size=16, shuffle=True, num_workers=0, # 👈 Windows 下训练先设置为 0，后续再尝试 4 pin_memory=True ) # ✅ 模型初始化 device = torch.device("cpu") model = BallMambaModel(in_channels=3, num_keypoints=512).to(device) optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) criterion = ChamferLoss().to(device) # ✅ 训练循环 for epoch in range(50): model.train() total_loss = 0 for i, points in enumerate(loader): points = points.to(device) # 输入输出一致（重构任务） recon_points = model(points) loss = criterion(recon_points, points) optimizer.zero_grad() loss.backward() optimizer.step() total_loss += loss.item() if i % 10 == 0: print(f"Epoch [{epoch+1}/50], Batch [{i+1}/{len(loader)}], Loss: {loss.item():.4f}") print(f"Epoch [{epoch+1}/50] 完成，平均 Loss: {total_loss / len(loader):.4f}") torch.save(model.state_dict(), f"models/ballmamba_epoch_{epoch+1}.pth") 上述是我的项目模型训练的代码，现在运行后出现问题C:\ProgramData\miniconda3\envs\torch\python.exe D:\桌面\point\scripts\train_model.py ✅ 共发现 907 个点云文件，用于训练 ✅ 共发现 907 个点云文件，用于训练 pos shape: torch.Size([16, 1024, 3]) [MambaBlock] A.shape = torch.Size([1, 1, 128, 16]) [MambaBlock] B.shape = torch.Size([16, 512, 1, 16]) [MambaBlock] C.shape = torch.Size([16, 512, 1, 16]) [MambaBlock] dt.shape = torch.Size([16, 512, 128]) Traceback (most recent call last): File "D:\桌面\point\scripts\train_model.py", line 76, in <module> recon_points = model(points) File "C:\ProgramData\miniconda3\envs\torch\lib\site-packages\torch\nn\modules\module.py", line 727, in _call_impl result = self.forward(*input, **kwargs) File "D:\桌面\point\models\ballmamba.py", line 63, in forward x = self.fps(None, pos) File "C:\ProgramData\miniconda3\envs\torch\lib\site-packages\torch\nn\modules\module.py", line 727, in _call_impl result = self.forward(*input, **kwargs) File "D:\桌面\point\models\ballmamba.py", line 30, in forward x_sampled = self.mamba(x_sampled) # (B, k, hidden_channels) File "C:\ProgramData\miniconda3\envs\torch\lib\site-packages\torch\nn\modules\module.py", line 727, in _call_impl result = self.forward(*input, **kwargs) File "D:\桌面\point\models\mamba_block.py", line 111, in forward y = y + x * self.D.view(1, 1, -1) # 加上跳跃连接 RuntimeError: The size of tensor a (16) must match the size of tensor b (512) at non-singleton dimension 2 进程已结束，退出代码为 1 应该如何修改代码？给我修改后的完整代码

loss = F.cross_entropy(pred.permute(0, 2, 1), target) # 交换维度 --- #### **3. 点云模型特殊注意事项** - **点云维度**：确认输入数据是否为标准格式 (批大小, 特征数, 点数)，典型值如 (32, 3, ...

Traceback (most recent call last): File "d:\海康\MVS\Development\Samples\Python\MvImport\two.py", line 702, in <module> class MainWindow(QMainWindow): File "d:\海康\MVS\Development\Samples\Python\MvImport\two.py", line 820, in MainWindow brightness = np.mean(gray) NameError: name 'gray' is not defined这个错误是下面这个代码的 # -- coding: utf-8 -- import sys import os import cv2 import numpy as np import math import time import logging import threading from collections import deque from PyQt5.QtWidgets import ( QApplication, QMainWindow, QPushButton, QWidget, QVBoxLayout, QHBoxLayout, QMessageBox, QLabel, QFileDialog, QToolBox, QComboBox, QStatusBar, QGroupBox, QSlider, QDockWidget, QProgressDialog, QLineEdit, QRadioButton, QGridLayout, QSpinBox, QCheckBox, QDialog, QDialogButtonBox, QDoubleSpinBox, QProgressBar, ) from PyQt5.QtCore import QRect, Qt, QSettings, QThread, pyqtSignal, QTimer, QMetaObject, pyqtSlot,Q_ARG from PyQt5.QtGui import QImage, QPixmap from CamOperation_class import CameraOperation from MvCameraControl_class import * import ctypes from ctypes import cast, POINTER from datetime import datetime import skimage import platform from CameraParams_header import ( MV_GIGE_DEVICE, MV_USB_DEVICE, MV_GENTL_CAMERALINK_DEVICE, MV_GENTL_CXP_DEVICE, MV_GENTL_XOF_DEVICE ) # ===== 全局配置 ===== # 模板匹配参数 MATCH_THRESHOLD = 0.75 # 降低匹配置信度阈值以提高灵敏度 MIN_MATCH_COUNT = 10 # 最小匹配特征点数量 MIN_FRAME_INTERVAL = 0.1 # 最小检测间隔(秒) # ===== 全局变量 ===== current_sample_path = "" detection_history = [] isGrabbing = False isOpen = False obj_cam_operation = None frame_monitor_thread = None template_matcher_thread = None MV_OK = 0 MV_E_CALLORDER = -2147483647 # ==================== 优化后的质量检测算法 ==================== def enhanced_check_print_quality(sample_image_path, test_image, threshold=0.05): # 不再使用传感器数据调整阈值 adjusted_threshold = threshold try: sample_img_data = np.fromfile(sample_image_path, dtype=np.uint8) sample_image = cv2.imdecode(sample_img_data, cv2.IMREAD_GRAYSCALE) if sample_image is None: logging.error(f"无法解码样本图像: {sample_image_path}") return None, None, None except Exception as e: logging.exception(f"样本图像读取异常: {str(e)}") return None, None, None if len(test_image.shape) == 3: test_image_gray = cv2.cvtColor(test_image, cv2.COLOR_BGR2GRAY) else: test_image_gray = test_image.copy() sample_image = cv2.GaussianBlur(sample_image, (5, 5), 0) test_image_gray = cv2.GaussianBlur(test_image_gray, (5, 5), 0) try: # 使用更鲁棒的SIFT特征检测器 sift = cv2.SIFT_create() keypoints1, descriptors1 = sift.detectAndCompute(sample_image, None) keypoints2, descriptors2 = sift.detectAndCompute(test_image_gray, None) if descriptors1 is None or descriptors2 is None: logging.warning("无法提取特征描述符，跳过配准") aligned_sample = sample_image else: # 使用FLANN匹配器提高匹配精度 FLANN_INDEX_KDTREE = 1 index_params = dict(algorithm=FLANN_INDEX_KDTREE, trees=5) search_params = dict(checks=50) flann = cv2.FlannBasedMatcher(index_params, search_params) matches = flann.knnMatch(descriptors1, descriptors2, k=2) # 应用Lowe's比率测试筛选优质匹配 good_matches = [] for m, n in matches: if m.distance < 0.7 * n.distance: good_matches.append(m) if len(good_matches) > MIN_MATCH_COUNT: src_pts = np.float32([keypoints1[m.queryIdx].pt for m in good_matches]).reshape(-1, 1, 2) dst_pts = np.float32([keypoints2[m.trainIdx].pt for m in good_matches]).reshape(-1, 1, 2) H, mask = cv2.findHomography(src_pts, dst_pts, cv2.RANSAC, 5.0) if H is not None: aligned_sample = cv2.warpPerspective( sample_image, H, (test_image_gray.shape[1], test_image_gray.shape[0]) ) logging.info("图像配准成功，使用配准后样本") else: aligned_sample = sample_image logging.warning("无法计算单应性矩阵，使用原始样本") else: aligned_sample = sample_image logging.warning(f"特征点匹配不足({len(good_matches)}/{MIN_MATCH_COUNT})，跳过图像配准") except Exception as e: logging.error(f"图像配准失败: {str(e)}") aligned_sample = sample_image try: if aligned_sample.shape != test_image_gray.shape: test_image_gray = cv2.resize(test_image_gray, (aligned_sample.shape[1], aligned_sample.shape[0])) except Exception as e: logging.error(f"图像调整大小失败: {str(e)}") return None, None, None try: from skimage.metrics import structural_similarity as compare_ssim ssim_score, ssim_diff = compare_ssim( aligned_sample, test_image_gray, full=True, gaussian_weights=True, data_range=255 ) except ImportError: from skimage.measure import compare_ssim ssim_score, ssim_diff = compare_ssim( aligned_sample, test_image_gray, full=True, gaussian_weights=True ) except Exception as e: logging.error(f"SSIM计算失败: {str(e)}") abs_diff = cv2.absdiff(aligned_sample, test_image_gray) ssim_diff = abs_diff.astype(np.float32) / 255.0 ssim_score = 1.0 - np.mean(ssim_diff) ssim_diff = (1 - ssim_diff) * 255 abs_diff = cv2.absdiff(aligned_sample, test_image_gray) combined_diff = cv2.addWeighted(ssim_diff.astype(np.uint8), 0.7, abs_diff, 0.3, 0) _, thresholded = cv2.threshold(combined_diff, 30, 255, cv2.THRESH_BINARY) kernel = np.ones((3, 3), np.uint8) thresholded = cv2.morphologyEx(thresholded, cv2.MORPH_OPEN, kernel) thresholded = cv2.morphologyEx(thresholded, cv2.MORPH_CLOSE, kernel) diff_pixels = np.count_nonzero(thresholded) total_pixels = aligned_sample.size diff_ratio = diff_pixels / total_pixels is_qualified = diff_ratio <= adjusted_threshold marked_image = cv2.cvtColor(test_image_gray, cv2.COLOR_GRAY2BGR) marked_image[thresholded == 255] = [0, 0, 255] # 放大缺陷标记 scale_factor = 2.0 # 放大2倍 marked_image = cv2.resize(marked_image, None, fx=scale_factor, fy=scale_factor, interpolation=cv2.INTER_LINEAR) labels = skimage.measure.label(thresholded) properties = skimage.measure.regionprops(labels) for prop in properties: if prop.area > 50: y, x = prop.centroid # 根据放大比例调整坐标 x_scaled = int(x * scale_factor) y_scaled = int(y * scale_factor) cv2.putText(marked_image, f"Defect", (x_scaled, y_scaled), cv2.FONT_HERSHEY_SIMPLEX, 0.5 * scale_factor, (0, 255, 255), int(scale_factor)) return is_qualified, diff_ratio, marked_image # ==================== 视觉触发的质量检测流程 ==================== def vision_controlled_check(capture_image=None, match_score=0.0): """修改为接受图像帧和匹配分数""" global current_sample_path, detection_history logging.info("视觉触发质量检测启动") # 如果没有提供图像，使用当前帧 if capture_image is None: frame = obj_cam_operation.get_current_frame() else: frame = capture_image if frame is None: QMessageBox.warning(mainWindow, "错误", "无法获取当前帧图像！", QMessageBox.Ok) return progress = QProgressDialog("正在检测...", "取消", 0, 100, mainWindow) progress.setWindowModality(Qt.WindowModal) progress.setValue(10) try: diff_threshold = mainWindow.sliderDiffThreshold.value() / 100.0 logging.info(f"使用差异度阈值: {diff_threshold}") progress.setValue(30) is_qualified, diff_ratio, marked_image = enhanced_check_print_quality( current_sample_path, frame, threshold=diff_threshold ) progress.setValue(70) if is_qualified is None: QMessageBox.critical(mainWindow, "检测错误", "检测失败，请检查日志", QMessageBox.Ok) return logging.info(f"检测结果: 合格={is_qualified}, 差异={diff_ratio}") progress.setValue(90) update_diff_display(diff_ratio, is_qualified) result_text = f"印花是否合格: {'合格' if is_qualified else '不合格'}\n差异占比: {diff_ratio100:.2f}%\n阈值: {diff_threshold100:.2f}%" QMessageBox.information(mainWindow, "检测结果", result_text, QMessageBox.Ok) if marked_image is not None: # 创建可调整大小的窗口 cv2.namedWindow("缺陷标记结果", cv2.WINDOW_NORMAL) cv2.resizeWindow("缺陷标记结果", 800, 600) # 初始大小 cv2.imshow("缺陷标记结果", marked_image) cv2.waitKey(0) cv2.destroyAllWindows() detection_result = { 'timestamp': datetime.now(), 'qualified': is_qualified, 'diff_ratio': diff_ratio, 'threshold': diff_threshold, 'trigger_type': 'vision' if capture_image else 'manual' } detection_history.append(detection_result) update_history_display() progress.setValue(100) except Exception as e: logging.exception("印花检测失败") QMessageBox.critical(mainWindow, "检测错误", f"检测过程中发生错误: {str(e)}", QMessageBox.Ok) finally: progress.close() # ==================== 相机操作函数 ==================== def open_device(): global deviceList, nSelCamIndex, obj_cam_operation, isOpen, frame_monitor_thread, mainWindow if isOpen: QMessageBox.warning(mainWindow, "Error", '相机已打开!', QMessageBox.Ok) return MV_E_CALLORDER nSelCamIndex = mainWindow.ComboDevices.currentIndex() if nSelCamIndex < 0: QMessageBox.warning(mainWindow, "Error", '请选择相机!', QMessageBox.Ok) return MV_E_CALLORDER # 创建相机控制对象 cam = MvCamera() # 初始化相机操作对象 obj_cam_operation = CameraOperation(cam, deviceList, nSelCamIndex) ret = obj_cam_operation.open_device() if 0 != ret: strError = "打开设备失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) isOpen = False else: set_continue_mode() get_param() isOpen = True enable_controls() # 创建并启动帧监控线程 frame_monitor_thread = FrameMonitorThread(obj_cam_operation) frame_monitor_thread.frame_status.connect(mainWindow.statusBar().showMessage) frame_monitor_thread.start() def start_grabbing(): global obj_cam_operation, isGrabbing, template_matcher_thread ret = obj_cam_operation.start_grabbing(mainWindow.widgetDisplay.winId()) if ret != 0: strError = "开始取流失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: isGrabbing = True enable_controls() # 等待第一帧到达 QThread.msleep(500) if not obj_cam_operation.is_frame_available(): QMessageBox.warning(mainWindow, "警告", "开始取流后未接收到帧，请检查相机连接！", QMessageBox.Ok) # 如果启用了自动检测，启动检测线程 if mainWindow.chkContinuousMatch.isChecked(): toggle_template_matching(True) def stop_grabbing(): global obj_cam_operation, isGrabbing, template_matcher_thread ret = obj_cam_operation.Stop_grabbing() if ret != 0: strError = "停止取流失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: isGrabbing = False enable_controls() # 停止模板匹配线程 if template_matcher_thread and template_matcher_thread.isRunning(): template_matcher_thread.stop() def close_device(): global isOpen, isGrabbing, obj_cam_operation, frame_monitor_thread, template_matcher_thread if frame_monitor_thread and frame_monitor_thread.isRunning(): frame_monitor_thread.stop() frame_monitor_thread.wait(2000) # 停止模板匹配线程 if template_matcher_thread and template_matcher_thread.isRunning(): template_matcher_thread.stop() template_matcher_thread.wait(2000) template_matcher_thread = None if isOpen and obj_cam_operation: obj_cam_operation.close_device() isOpen = False isGrabbing = False enable_controls() # ==================== 连续帧匹配检测器 ==================== class ContinuousFrameMatcher(QThread): frame_processed = pyqtSignal(np.ndarray, float, bool) # 处理后的帧, 匹配分数, 是否匹配 match_score_updated = pyqtSignal(float) # 匹配分数更新信号 match_success = pyqtSignal(np.ndarray, float) # 匹配成功信号 (帧, 匹配分数) def init(self, cam_operation, parent=None): super().init(parent) self.cam_operation = cam_operation self.running = True self.sample_template = None self.min_match_count = MIN_MATCH_COUNT self.match_threshold = MATCH_THRESHOLD self.sample_kp = None self.sample_des = None self.current_match_score = 0.0 self.last_match_time = 0 self.frame_counter = 0 self.consecutive_fail_count = 0 self.last_trigger_time = 0 # 上次触发时间 self.cool_down = 0.2 # 冷却时间(秒) # 特征检测器 - 使用SIFT self.sift = cv2.SIFT_create() # 特征匹配器 - 使用FLANN提高匹配精度 FLANN_INDEX_KDTREE = 1 index_params = dict(algorithm=FLANN_INDEX_KDTREE, trees=5) search_params = dict(checks=50) self.flann = cv2.FlannBasedMatcher(index_params, search_params) # 性能监控 self.processing_times = deque(maxlen=100) self.frame_rates = deque(maxlen=100) # 黑白相机优化 self.clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8)) def preprocess_image(self, image): """增强黑白图像特征提取""" # 如果是单通道图像，转换为三通道 if len(image.shape) == 2: image = cv2.cvtColor(image, cv2.COLOR_GRAY2BGR) # 对比度增强 (LAB空间) lab = cv2.cvtColor(image, cv2.COLOR_BGR2LAB) l, a, b = cv2.split(lab) cl = self.clahe.apply(l) limg = cv2.merge((cl, a, b)) enhanced = cv2.cvtColor(limg, cv2.COLOR_LAB2BGR) # 边缘增强 gray = cv2.cvtColor(enhanced, cv2.COLOR_BGR2GRAY) edges = cv2.Canny(gray, 50, 150) # 组合特征 return cv2.addWeighted(enhanced, 0.7, cv2.cvtColor(edges, cv2.COLOR_GRAY2BGR), 0.3, 0) def set_sample(self, sample_img): """设置标准样本并提取特征""" # 保存样本图像 self.sample_img = sample_img # 预处理增强特征 processed_sample = self.preprocess_image(sample_img) # 提取样本特征点 self.sample_kp, self.sample_des = self.sift.detectAndCompute(processed_sample, None) if self.sample_des is None or len(self.sample_kp) < self.min_match_count: logging.warning("样本特征点不足") return False logging.info(f"样本特征提取成功: {len(self.sample_kp)}个关键点") return True def process_frame(self, frame): """处理帧：特征提取、匹配和可视化""" is_matched = False match_score = 0.0 processed_frame = frame.copy() # 检查是否已设置样本 if self.sample_kp is None or self.sample_des is None: return processed_frame, match_score, is_matched # 预处理当前帧 processed_frame = self.preprocess_image(frame) # 转换为灰度图像用于特征提取 gray_frame = cv2.cvtColor(processed_frame, cv2.COLOR_BGR2GRAY) try: # 提取当前帧的特征点 kp, des = self.sift.detectAndCompute(gray_frame, None) if des is None or len(kp) < 10: # 特征点不足 return processed_frame, match_score, is_matched # 匹配特征点 matches = self.flann.knnMatch(self.sample_des, des, k=2) # 应用Lowe's比率测试 good_matches = [] for m, n in matches: if m.distance < 0.7 * n.distance: good_matches.append(m) # 计算匹配分数（匹配点数量占样本特征点数量的比例） if len(self.sample_kp) > 0: match_score = len(good_matches) / len(self.sample_kp) match_score = min(1.0, max(0.0, match_score)) # 确保在0-1范围内 else: match_score = 0.0 # 判断是否匹配成功 if len(good_matches) >= self.min_match_count and match_score >= self.match_threshold: is_matched = True # 在图像上绘制匹配结果 if len(gray_frame.shape) == 2: processed_frame = cv2.cvtColor(gray_frame, cv2.COLOR_GRAY2BGR) # 绘制匹配点 processed_frame = cv2.drawMatches( self.sample_img, self.sample_kp, processed_frame, kp, good_matches, None, flags=cv2.DrawMatchesFlags_NOT_DRAW_SINGLE_POINTS ) # 在图像上显示匹配分数 cv2.putText(processed_frame, f"Match Score: {match_score:.2f}", (20, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2) # 更新当前匹配分数 self.current_match_score = match_score self.match_score_updated.emit(match_score) # 检查是否匹配成功且超过冷却时间 current_time = time.time() if is_matched and (current_time - self.last_trigger_time) > self.cool_down: self.last_trigger_time = current_time logging.info(f"匹配成功! 分数: {match_score:.2f}, 触发质量检测") # 发出匹配成功信号 (传递当前帧) self.match_success.emit(frame.copy(), match_score) except Exception as e: logging.error(f"帧处理错误: {str(e)}") return processed_frame, match_score, is_matched def set_threshold(self, threshold): """更新匹配阈值""" self.match_threshold = max(0.0, min(1.0, threshold)) logging.info(f"更新匹配阈值: {self.match_threshold:.2f}") def run(self): """主处理循环 - 连续处理每一帧""" logging.info("连续帧匹配线程启动") self.last_match_time = time.time() self.consecutive_fail_count = 0 while self.running: start_time = time.time() # 检查相机状态 if not self.cam_operation or not self.cam_operation.is_grabbing: if self.consecutive_fail_count % 10 == 0: logging.debug("相机未取流，等待...") time.sleep(0.1) self.consecutive_fail_count += 1 continue # 获取当前帧 frame = self.cam_operation.get_current_frame() if frame is None: self.consecutive_fail_count += 1 if self.consecutive_fail_count % 10 == 0: logging.warning(f"连续{self.consecutive_fail_count}次获取帧失败") time.sleep(0.05) continue self.consecutive_fail_count = 0 try: # 处理帧 processed_frame, match_score, is_matched = self.process_frame(frame) # 发送处理结果 self.frame_processed.emit(processed_frame, match_score, is_matched) except Exception as e: logging.error(f"帧处理错误: {str(e)}") # 控制处理频率 processing_time = time.time() - start_time sleep_time = max(0.01, MIN_FRAME_INTERVAL - processing_time) time.sleep(sleep_time) logging.info("连续帧匹配线程退出") # ==================== 模板匹配控制函数 ==================== def toggle_template_matching(state): global template_matcher_thread, current_sample_path logging.debug(f"切换连续匹配状态: {state}") if state == Qt.Checked and isGrabbing: # 确保已设置样本 if not current_sample_path: logging.warning("尝试启动连续匹配但未设置样本") QMessageBox.warning(mainWindow, "错误", "请先设置标准样本", QMessageBox.Ok) mainWindow.chkContinuousMatch.setChecked(False) return if template_matcher_thread is None: logging.info("创建新的连续帧匹配线程") template_matcher_thread = ContinuousFrameMatcher(obj_cam_operation) template_matcher_thread.frame_processed.connect(update_frame_display) template_matcher_thread.match_score_updated.connect(update_match_score_display) # 正确连接匹配成功信号到质量检测函数 template_matcher_thread.match_success.connect( lambda frame, score: vision_controlled_check(frame, score) ) # 加载样本图像 sample_img = cv2.imread(current_sample_path) if sample_img is None: logging.error("无法加载标准样本图像") QMessageBox.warning(mainWindow, "错误", "无法加载标准样本图像", QMessageBox.Ok) mainWindow.chkContinuousMatch.setChecked(False) return if not template_matcher_thread.set_sample(sample_img): logging.warning("标准样本特征不足") QMessageBox.warning(mainWindow, "错误", "标准样本特征不足", QMessageBox.Ok) mainWindow.chkContinuousMatch.setChecked(False) return if not template_matcher_thread.isRunning(): logging.info("启动连续帧匹配线程") template_matcher_thread.start() elif template_matcher_thread and template_matcher_thread.isRunning(): logging.info("停止连续帧匹配线程") template_matcher_thread.stop() # 重置匹配分数显示 update_match_score_display(0.0) # 重置帧显示 if obj_cam_operation and obj_cam_operation.is_frame_available(): frame = obj_cam_operation.get_current_frame() if frame is not None: display_frame = frame.copy() # 添加状态信息 cv2.putText(display_frame, "Continuous Matching Disabled", (20, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2) update_frame_display(display_frame, 0.0, False) # 添加类型转换函数 def numpy_to_qimage(np_array): """将 numpy 数组转换为 QImage""" if np_array is None: return QImage() height, width, channel = np_array.shape bytes_per_line = 3 * width # 确保数据是连续的 if not np_array.flags['C_CONTIGUOUS']: np_array = np.ascontiguousarray(np_array) # 转换 BGR 到 RGB rgb_image = cv2.cvtColor(np_array, cv2.COLOR_BGR2RGB) # 创建 QImage qimg = QImage( rgb_image.data, width, height, bytes_per_line, QImage.Format_RGB888 ) # 复制数据以避免内存问题 return qimg.copy() # 修改 update_frame_display 函数 def update_frame_display(frame, match_score, is_matched): """更新主显示窗口（线程安全）""" # 确保在GUI线程中执行 if QThread.currentThread() != QApplication.instance().thread(): # 转换为 QImage 再传递 qimg = numpy_to_qimage(frame) QMetaObject.invokeMethod( mainWindow, "updateDisplay", Qt.QueuedConnection, Q_ARG(QImage, qimg), Q_ARG(float, match_score), Q_ARG(bool, is_matched) ) return # 如果已经在主线程，直接调用主窗口的更新方法 mainWindow.updateDisplay(frame, match_score, is_matched) def update_match_score_display(score): """更新匹配分数显示""" # 将分数转换为百分比显示 score_percent = score * 100 mainWindow.lblMatchScoreValue.setText(f"{score_percent:.1f}%") # 根据分数设置颜色 if score > 0.8: # 高于80%显示绿色 color = "green" elif score > 0.6: # 60%-80%显示黄色 color = "orange" else: # 低于60%显示红色 color = "red" mainWindow.lblMatchScoreValue.setStyleSheet(f"color: {color}; font-weight: bold;") def update_diff_display(diff_ratio, is_qualified): mainWindow.lblCurrentDiff.setText(f"当前差异度: {diff_ratio100:.2f}%") if is_qualified: mainWindow.lblDiffStatus.setText("状态: 合格") mainWindow.lblDiffStatus.setStyleSheet("color: green; font-size: 12px;") else: mainWindow.lblDiffStatus.setText("状态: 不合格") mainWindow.lblDiffStatus.setStyleSheet("color: red; font-size: 12px;") def update_diff_threshold(value): mainWindow.lblDiffValue.setText(f"{value}%") def update_sample_display(): global current_sample_path if current_sample_path: mainWindow.lblSamplePath.setText(f"当前样本: {os.path.basename(current_sample_path)}") mainWindow.lblSamplePath.setToolTip(current_sample_path) mainWindow.bnPreviewSample.setEnabled(True) else: mainWindow.lblSamplePath.setText("当前样本: 未设置样本") mainWindow.bnPreviewSample.setEnabled(False) def update_history_display(): global detection_history mainWindow.cbHistory.clear() for i, result in enumerate(detection_history[-10:]): timestamp = result['timestamp'].strftime("%H:%M:%S") status = "合格" if result['qualified'] else "不合格" ratio = f"{result['diff_ratio']100:.2f}%" trigger = "视觉" if result['trigger_type'] == 'vision' else "手动" mainWindow.cbHistory.addItem(f"[{trigger} {timestamp}] {status} - 差异: {ratio}") def update_match_threshold(value): """更新匹配阈值显示并应用到匹配器""" global template_matcher_thread # 更新UI显示 if mainWindow: mainWindow.lblThresholdValue.setText(f"{value}%") # 如果匹配线程存在，更新其匹配阈值 if template_matcher_thread: # 转换为0-1范围的浮点数 threshold = value / 100.0 template_matcher_thread.set_threshold(threshold) logging.debug(f"更新匹配阈值: {threshold:.2f}") # ==================== 主窗口类 ==================== class MainWindow(QMainWindow): def init(self): super().init() self.setWindowTitle("布料印花检测系统 - 连续匹配版") self.resize(1200, 800) central_widget = QWidget() self.setCentralWidget(central_widget) main_layout = QVBoxLayout(central_widget) # 设备枚举区域 device_layout = QHBoxLayout() self.ComboDevices = QComboBox() self.bnEnum = QPushButton("枚举设备") self.bnOpen = QPushButton("打开设备") self.bnClose = QPushButton("关闭设备") device_layout.addWidget(self.ComboDevices) device_layout.addWidget(self.bnEnum) device_layout.addWidget(self.bnOpen) device_layout.addWidget(self.bnClose) main_layout.addLayout(device_layout) # 取流控制组 self.groupGrab = QGroupBox("取流控制") grab_layout = QHBoxLayout(self.groupGrab) self.bnStart = QPushButton("开始取流") self.bnStop = QPushButton("停止取流") self.radioContinueMode = QRadioButton("连续模式") self.radioTriggerMode = QRadioButton("触发模式") self.bnSoftwareTrigger = QPushButton("软触发") grab_layout.addWidget(self.bnStart) grab_layout.addWidget(self.bnStop) grab_layout.addWidget(self.radioContinueMode) grab_layout.addWidget(self.radioTriggerMode) grab_layout.addWidget(self.bnSoftwareTrigger) main_layout.addWidget(self.groupGrab) # 参数设置组 self.paramgroup = QGroupBox("相机参数") param_layout = QGridLayout(self.paramgroup) self.edtExposureTime = QLineEdit() self.edtGain = QLineEdit() self.edtFrameRate = QLineEdit() self.bnGetParam = QPushButton("获取参数") self.bnSetParam = QPushButton("设置参数") self.bnSaveImage = QPushButton("保存图像") param_layout.addWidget(QLabel("曝光时间:"), 0, 0) param_layout.addWidget(self.edtExposureTime, 0, 1) param_layout.addWidget(self.bnGetParam, 0, 2) param_layout.addWidget(QLabel("增益:"), 1, 0) param_layout.addWidget(self.edtGain, 1, 1) param_layout.addWidget(self.bnSetParam, 1, 2) param_layout.addWidget(QLabel("帧率:"), 2, 0) param_layout.addWidget(self.edtFrameRate, 2, 1) param_layout.addWidget(self.bnSaveImage, 2, 2) main_layout.addWidget(self.paramgroup) # 图像显示区域 self.widgetDisplay = QLabel() self.widgetDisplay.setMinimumSize(640, 480) self.widgetDisplay.setStyleSheet("background-color: black;") self.widgetDisplay.setAlignment(Qt.AlignCenter) self.widgetDisplay.setText("相机预览区域") main_layout.addWidget(self.widgetDisplay, 1) # 创建自定义UI组件 self.setup_custom_ui() # 添加阈值自适应定时器 self.threshold_timer = QTimer() self.threshold_timer.timeout.connect(self.auto_adjust_threshold) self.threshold_timer.start(2000) # 每2秒调整一次 def auto_adjust_threshold(self): """根据环境亮度自动调整匹配阈值""" if not obj_cam_operation or not isGrabbing: return # 获取当前帧并计算平均亮度 frame = obj_cam_operation.get_current_frame() if frame is None: return gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) brightness = np.mean(gray) # 根据亮度动态调整阈值 (亮度低时降低阈值要求) if brightness < 50: # 暗环境 new_threshold = 40 # 40% elif brightness > 200: # 亮环境 new_threshold = 65 # 65% else: # 正常环境 new_threshold = 55 # 55% # 更新UI self.sliderThreshold.setValue(new_threshold) self.lblThresholdValue.setText(f"{new_threshold}%") # 更新匹配器阈值 update_match_threshold(new_threshold) # 状态栏显示调整信息 self.statusBar().showMessage(f"亮度: {brightness:.1f}, 自动调整阈值至: {new_threshold}%", 3000) # 处理不同通道数的图像 if len(frame.shape) == 2 or frame.shape[2] == 1: # 已经是灰度图 gray = frame elif frame.shape[2] == 3: # 三通道彩色图 gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) elif frame.shape[2] == 4: # 四通道图（带alpha） gray = cv2.cvtColor(frame, cv2.COLOR_BGRA2GRAY) else: # 其他通道数，无法处理 logging.warning(f"无法处理的图像格式: {frame.shape}") return # 计算平均亮度 brightness = np.mean(gray) def setup_custom_ui(self): # 工具栏 toolbar = self.addToolBar("检测工具") self.bnCheckPrint = QPushButton("手动检测") self.bnSaveSample = QPushButton("保存标准样本") self.bnPreviewSample = QPushButton("预览样本") self.cbHistory = QComboBox() self.cbHistory.setMinimumWidth(300) toolbar.addWidget(self.bnCheckPrint) toolbar.addWidget(self.bnSaveSample) toolbar.addWidget(self.bnPreviewSample) toolbar.addWidget(QLabel("历史记录:")) toolbar.addWidget(self.cbHistory) # 状态栏样本路径 self.lblSamplePath = QLabel("当前样本: 未设置样本") self.statusBar().addPermanentWidget(self.lblSamplePath) # 右侧面板 right_panel = QWidget() right_layout = QVBoxLayout(right_panel) right_layout.setContentsMargins(10, 10, 10, 10) # 差异度调整组 diff_group = QGroupBox("差异度调整") diff_layout = QVBoxLayout(diff_group) self.lblDiffThreshold = QLabel("差异度阈值 (0-100%):") self.sliderDiffThreshold = QSlider(Qt.Horizontal) self.sliderDiffThreshold.setRange(0, 100) self.sliderDiffThreshold.setValue(5) self.lblDiffValue = QLabel("5%") self.lblCurrentDiff = QLabel("当前差异度: -") self.lblCurrentDiff.setStyleSheet("font-size: 14px; font-weight: bold;") self.lblDiffStatus = QLabel("状态: 未检测") self.lblDiffStatus.setStyleSheet("font-size: 12px;") diff_layout.addWidget(self.lblDiffThreshold) diff_layout.addWidget(self.sliderDiffThreshold) diff_layout.addWidget(self.lblDiffValue) diff_layout.addWidget(self.lblCurrentDiff) diff_layout.addWidget(self.lblDiffStatus) right_layout.addWidget(diff_group) # ===== 连续匹配面板 ===== match_group = QGroupBox("连续帧匹配") match_layout = QVBoxLayout(match_group) # 样本设置 sample_layout = QHBoxLayout() self.bnSetSample = QPushButton("设置标准样本") self.bnPreviewSample = QPushButton("预览样本") self.lblSampleStatus = QLabel("状态: 未设置样本") sample_layout.addWidget(self.bnSetSample) sample_layout.addWidget(self.bnPreviewSample) sample_layout.addWidget(self.lblSampleStatus) match_layout.addLayout(sample_layout) # 匹配参数 param_layout = QHBoxLayout() self.lblMatchThreshold = QLabel("匹配阈值:") self.sliderThreshold = QSlider(Qt.Horizontal) self.sliderThreshold.setRange(50, 100) self.sliderThreshold.setValue(75) # 降低默认阈值 self.lblThresholdValue = QLabel("75%") param_layout.addWidget(self.lblMatchThreshold) param_layout.addWidget(self.sliderThreshold) param_layout.addWidget(self.lblThresholdValue) match_layout.addLayout(param_layout) # 匹配分数显示 match_score_layout = QHBoxLayout() self.lblMatchScore = QLabel("实时匹配分数:") self.lblMatchScoreValue = QLabel("0.0%") self.lblMatchScoreValue.setStyleSheet("font-weight: bold;") match_score_layout.addWidget(self.lblMatchScore) match_score_layout.addWidget(self.lblMatchScoreValue) match_layout.addLayout(match_score_layout) # 连续匹配开关 self.chkContinuousMatch = QCheckBox("启用连续帧匹配") self.chkContinuousMatch.setChecked(False) match_layout.addWidget(self.chkContinuousMatch) right_layout.addWidget(match_group) right_layout.addStretch(1) # 停靠窗口 dock = QDockWidget("检测控制面板", self) dock.setWidget(right_panel) dock.setFeatures(QDockWidget.DockWidgetMovable | QDockWidget.DockWidgetFloatable) self.addDockWidget(Qt.RightDockWidgetArea, dock) @pyqtSlot(QImage, float, bool) def updateDisplay(self, qimg, match_score, is_matched): """线程安全的显示更新方法（只接收 QImage）""" if qimg.isNull(): return # 创建QPixmap并缩放 pixmap = QPixmap.fromImage(qimg) scaled_pixmap = pixmap.scaled( self.widgetDisplay.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation ) # 更新显示 self.widgetDisplay.setPixmap(scaled_pixmap) self.widgetDisplay.setAlignment(Qt.AlignCenter) # 更新显示 self.widgetDisplay.setPixmap(scaled_pixmap) self.widgetDisplay.setAlignment(Qt.AlignCenter) def closeEvent(self, event): logging.info("主窗口关闭，执行清理...") close_device() event.accept() # ===== 辅助函数 ===== def ToHexStr(num): if not isinstance(num, int): try: num = int(num) except: return f"<非整数:{type(num)}>" chaDic = {10: 'a', 11: 'b', 12: 'c', 13: 'd', 14: 'e', 15: 'f'} hexStr = "" if num < 0: num = num + 2 ** 32 while num >= 16: digit = num % 16 hexStr = chaDic.get(digit, str(digit)) + hexStr num //= 16 hexStr = chaDic.get(num, str(num)) + hexStr return "0x" + hexStr def enum_devices(): global deviceList, obj_cam_operation n_layer_type = ( MV_GIGE_DEVICE | MV_USB_DEVICE | MV_GENTL_CAMERALINK_DEVICE | MV_GENTL_CXP_DEVICE | MV_GENTL_XOF_DEVICE ) # 创建设备列表 deviceList = MV_CC_DEVICE_INFO_LIST() # 枚举设备 ret = MvCamera.MV_CC_EnumDevices(n_layer_type, deviceList) if ret != MV_OK: error_msg = f"枚举设备失败! 错误码: 0x{ret:x}" logging.error(error_msg) QMessageBox.warning(mainWindow, "错误", error_msg, QMessageBox.Ok) return ret if deviceList.nDeviceNum == 0: QMessageBox.warning(mainWindow, "提示", "未找到任何设备", QMessageBox.Ok) return MV_OK logging.info(f"找到 {deviceList.nDeviceNum} 个设备") # 处理设备信息 devList = [] for i in range(deviceList.nDeviceNum): # 获取设备信息 mvcc_dev_info = ctypes.cast( deviceList.pDeviceInfo[i], ctypes.POINTER(MV_CC_DEVICE_INFO) ).contents # 根据设备类型提取信息 if mvcc_dev_info.nTLayerType == MV_GIGE_DEVICE: st_gige_info = mvcc_dev_info.SpecialInfo.stGigEInfo ip_addr = ( f"{(st_gige_info.nCurrentIp >> 24) & 0xFF}." f"{(st_gige_info.nCurrentIp >> 16) & 0xFF}." f"{(st_gige_info.nCurrentIp >> 8) & 0xFF}." f"{st_gige_info.nCurrentIp & 0xFF}" ) # 修复：将c_ubyte_Array_16转换为字节串再解码 user_defined_bytes = bytes(st_gige_info.chUserDefinedName) dev_name = f"GigE: {user_defined_bytes.decode('gbk', 'ignore')}" devList.append(f"[{i}] {dev_name} ({ip_addr})") elif mvcc_dev_info.nTLayerType == MV_USB_DEVICE: st_usb_info = mvcc_dev_info.SpecialInfo.stUsb3VInfo serial = bytes(st_usb_info.chSerialNumber).decode('ascii', 'ignore').rstrip('\x00') # 修复：同样处理用户自定义名称 user_defined_bytes = bytes(st_usb_info.chUserDefinedName) dev_name = f"USB: {user_defined_bytes.decode('gbk', 'ignore')}" devList.append(f"[{i}] {dev_name} (SN: {serial})") else: devList.append(f"[{i}] 未知设备类型: {mvcc_dev_info.nTLayerType}") # 更新UI mainWindow.ComboDevices.clear() mainWindow.ComboDevices.addItems(devList) if devList: mainWindow.ComboDevices.setCurrentIndex(0) mainWindow.statusBar().showMessage(f"找到 {deviceList.nDeviceNum} 个设备", 3000) return MV_OK def set_continue_mode(): ret = obj_cam_operation.set_trigger_mode(False) if ret != 0: strError = "设置连续模式失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: mainWindow.radioContinueMode.setChecked(True) mainWindow.radioTriggerMode.setChecked(False) mainWindow.bnSoftwareTrigger.setEnabled(False) def set_software_trigger_mode(): ret = obj_cam_operation.set_trigger_mode(True) if ret != 0: strError = "设置触发模式失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: mainWindow.radioContinueMode.setChecked(False) mainWindow.radioTriggerMode.setChecked(True) mainWindow.bnSoftwareTrigger.setEnabled(isGrabbing) def trigger_once(): ret = obj_cam_operation.trigger_once() if ret != 0: strError = "软触发失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) def save_sample_image(): global isGrabbing, obj_cam_operation, current_sample_path if not isGrabbing: QMessageBox.warning(mainWindow, "错误", "请先开始取流并捕获图像！", QMessageBox.Ok) return # 尝试捕获当前帧 frame = obj_cam_operation.capture_frame() if frame is None: QMessageBox.warning(mainWindow, "无有效图像", "未捕获到有效图像，请检查相机状态！", QMessageBox.Ok) return # 确保图像有效 if frame.size == 0 or frame.shape[0] == 0 or frame.shape[1] == 0: QMessageBox.warning(mainWindow, "无效图像", "捕获的图像无效，请检查相机设置！", QMessageBox.Ok) return settings = QSettings("ClothInspection", "CameraApp") last_dir = settings.value("last_save_dir", os.path.join(os.getcwd(), "captures")) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") default_filename = f"sample_{timestamp}" file_path, selected_filter = QFileDialog.getSaveFileName( mainWindow, "保存标准样本图像", os.path.join(last_dir, default_filename), "BMP Files (.bmp);;PNG Files (.png);;JPEG Files (.jpg);;所有文件 ()", options=QFileDialog.DontUseNativeDialog ) if not file_path: return # 确保文件扩展名正确 file_extension = os.path.splitext(file_path)[1].lower() if not file_extension: if "BMP" in selected_filter: file_path += ".bmp" elif "PNG" in selected_filter: file_path += ".png" elif "JPEG" in selected_filter or "JPG" in selected_filter: file_path += ".jpg" else: file_path += ".bmp" file_extension = os.path.splitext(file_path)[1].lower() # 创建目录（如果不存在） directory = os.path.dirname(file_path) if directory and not os.path.exists(directory): try: os.makedirs(directory, exist_ok=True) except OSError as e: QMessageBox.critical(mainWindow, "目录创建错误", f"无法创建目录 {directory}: {str(e)}", QMessageBox.Ok) return # 保存图像 try: # 使用OpenCV保存图像 if not cv2.imwrite(file_path, frame): raise Exception("OpenCV保存失败") # 更新状态 current_sample_path = file_path update_sample_display() settings.setValue("last_save_dir", os.path.dirname(file_path)) # 显示成功消息 QMessageBox.information(mainWindow, "成功", f"标准样本已保存至:\n{file_path}", QMessageBox.Ok) # 更新样本状态 mainWindow.lblSampleStatus.setText("状态: 样本已设置") mainWindow.lblSampleStatus.setStyleSheet("color: green;") except Exception as e: logging.error(f"保存图像失败: {str(e)}") QMessageBox.critical(mainWindow, "保存错误", f"保存图像时发生错误:\n{str(e)}", QMessageBox.Ok) def preview_sample(): global current_sample_path if not current_sample_path or not os.path.exists(current_sample_path): QMessageBox.warning(mainWindow, "错误", "请先设置有效的标准样本图像！", QMessageBox.Ok) return try: # 直接使用OpenCV加载图像 sample_img = cv2.imread(current_sample_path) if sample_img is None: raise Exception("无法加载图像") # 显示图像 cv2.namedWindow("标准样本预览", cv2.WINDOW_NORMAL) cv2.resizeWindow("标准样本预览", 800, 600) cv2.imshow("标准样本预览", sample_img) cv2.waitKey(0) cv2.destroyAllWindows() except Exception as e: QMessageBox.warning(mainWindow, "错误", f"预览样本失败: {str(e)}", QMessageBox.Ok) def is_float(str): try: float(str) return True except ValueError: return False def get_param(): try: ret = obj_cam_operation.get_parameters() if ret != MV_OK: strError = "获取参数失败，错误码: " + ToHexStr(ret) QMessageBox.warning(mainWindow, "错误", strError, QMessageBox.Ok) else: mainWindow.edtExposureTime.setText("{0:.2f}".format(obj_cam_operation.exposure_time)) mainWindow.edtGain.setText("{0:.2f}".format(obj_cam_operation.gain)) mainWindow.edtFrameRate.setText("{0:.2f}".format(obj_cam_operation.frame_rate)) except Exception as e: error_msg = f"获取参数时发生错误: {str(e)}" QMessageBox.critical(mainWindow, "严重错误", error_msg, QMessageBox.Ok) def set_param(): frame_rate = mainWindow.edtFrameRate.text() exposure = mainWindow.edtExposureTime.text() gain = mainWindow.edtGain.text() if not (is_float(frame_rate) and is_float(exposure) and is_float(gain)): strError = "设置参数失败: 参数必须是有效的浮点数" QMessageBox.warning(mainWindow, "错误", strError, QMessageBox.Ok) return MV_E_PARAMETER try: ret = obj_cam_operation.set_param( frame_rate=float(frame_rate), exposure_time=float(exposure), gain=float(gain) ) if ret != MV_OK: strError = "设置参数失败，错误码: " + ToHexStr(ret) QMessageBox.warning(mainWindow, "错误", strError, QMessageBox.Ok) except Exception as e: error_msg = f"设置参数时发生错误: {str(e)}" QMessageBox.critical(mainWindow, "严重错误", error_msg, QMessageBox.Ok) def enable_controls(): global isGrabbing, isOpen mainWindow.groupGrab.setEnabled(isOpen) mainWindow.paramgroup.setEnabled(isOpen) mainWindow.bnOpen.setEnabled(not isOpen) mainWindow.bnClose.setEnabled(isOpen) mainWindow.bnStart.setEnabled(isOpen and (not isGrabbing)) mainWindow.bnStop.setEnabled(isOpen and isGrabbing) mainWindow.bnSoftwareTrigger.setEnabled(isGrabbing and mainWindow.radioTriggerMode.isChecked()) mainWindow.bnSaveImage.setEnabled(isOpen and isGrabbing) mainWindow.bnCheckPrint.setEnabled(isOpen and isGrabbing) mainWindow.bnSaveSample.setEnabled(isOpen and isGrabbing) mainWindow.bnPreviewSample.setEnabled(bool(current_sample_path)) # 连续匹配控制 mainWindow.chkContinuousMatch.setEnabled(bool(current_sample_path) and isGrabbing) # ===== 相机帧监控线程 ===== class FrameMonitorThread(QThread): frame_status = pyqtSignal(str) # 用于发送状态消息的信号 def init(self, cam_operation): super().init() self.cam_operation = cam_operation self.running = True self.frame_count = 0 self.last_time = time.time() def run(self): """监控相机帧状态的主循环""" while self.running: try: if self.cam_operation and self.cam_operation.is_grabbing: # 获取帧统计信息 frame_info = self.get_frame_info() if frame_info: fps = frame_info.get('fps', 0) dropped = frame_info.get('dropped', 0) status = f"FPS: {fps:.1f} | 丢帧: {dropped}" self.frame_status.emit(status) else: self.frame_status.emit("取流中...") else: self.frame_status.emit("相机未取流") except Exception as e: self.frame_status.emit(f"监控错误: {str(e)}") # 每500ms检查一次 QThread.msleep(500) def stop(self): """停止监控线程""" self.running = False self.wait(1000) # 等待线程结束 def calculate_fps(self): """计算当前帧率""" current_time = time.time() elapsed = current_time - self.last_time if elapsed > 0: fps = self.frame_count / elapsed self.frame_count = 0 self.last_time = current_time return fps return 0 def get_frame_info(self): """获取帧信息""" try: # 更新帧计数 self.frame_count += 1 # 返回帧信息 return { 'fps': self.calculate_fps(), 'dropped': 0 # 实际应用中需要从相机获取真实丢帧数 } except Exception as e: logging.error(f"获取帧信息失败: {str(e)}") return None # ===== 主程序入口 ===== if name == "main": # 配置日志系统 logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("cloth_inspection_continuous.log"), logging.StreamHandler() ] ) logging.info("布料印花检测系统(连续匹配版)启动") app = QApplication(sys.argv) mainWindow = MainWindow() mainWindow.sliderThreshold.valueChanged.connect( lambda value: update_match_threshold(value) ) # 信号连接 mainWindow.sliderDiffThreshold.valueChanged.connect(update_diff_threshold) mainWindow.bnCheckPrint.clicked.connect(lambda: vision_controlled_check(None)) mainWindow.bnSaveSample.clicked.connect(save_sample_image) mainWindow.bnPreviewSample.clicked.connect(preview_sample) mainWindow.bnEnum.clicked.connect(enum_devices) mainWindow.bnOpen.clicked.connect(open_device) mainWindow.bnClose.clicked.connect(close_device) mainWindow.bnStart.clicked.connect(start_grabbing) mainWindow.bnStop.clicked.connect(stop_grabbing) mainWindow.bnSoftwareTrigger.clicked.connect(trigger_once) mainWindow.radioTriggerMode.clicked.connect(set_software_trigger_mode) mainWindow.radioContinueMode.clicked.connect(set_continue_mode) mainWindow.bnGetParam.clicked.connect(get_param) mainWindow.bnSetParam.clicked.connect(set_param) mainWindow.bnSaveImage.clicked.connect(save_sample_image) # 连续匹配信号连接 mainWindow.sliderThreshold.valueChanged.connect(update_match_score_display) mainWindow.chkContinuousMatch.stateChanged.connect(toggle_template_matching) mainWindow.show() app.exec_() close_device() sys.exit()

brightness = np.mean(gray) 错误原因是在计算平均亮度时，变量gray没有定义。查看代码，我们发现gray是在条件分支中定义的，但并非所有分支都定义了gray。特别是当图像通道数不是2（灰度）或3（彩色）或4...

这个程序运行后也保存过样本后还是没有实现每一帧都与标准样本进行匹配比对（我是已经点了自动连续帧匹配后），是这个程序里的每一帧比对里有什么错误和问题吗？在运行过程中每一次画面变化界面里的匹配度数值也一直没有变化。 -- coding: utf-8 -- import sys import os import cv2 import numpy as np import math import time import logging from collections import deque from PyQt5.QtWidgets import ( QApplication, QMainWindow, QPushButton, QWidget, QVBoxLayout, QHBoxLayout, QMessageBox, QLabel, QFileDialog, QToolBox, QComboBox, QStatusBar, QGroupBox, QSlider, QDockWidget, QProgressDialog, QLineEdit, QRadioButton, QGridLayout, QSpinBox, QCheckBox, QDialog, QDialogButtonBox, QDoubleSpinBox, QProgressBar ) from PyQt5.QtCore import QRect, Qt, QSettings, QThread, pyqtSignal, QTimer,QMetaObject,pyqtSlot,QObject from PyQt5.QtGui import QImage, QPixmap from CamOperation_class import CameraOperation from MvCameraControl_class import * import ctypes from ctypes import cast, POINTER from datetime import datetime import skimage import platform from CameraParams_header import ( MV_GIGE_DEVICE, MV_USB_DEVICE, MV_GENTL_CAMERALINK_DEVICE, MV_GENTL_CXP_DEVICE, MV_GENTL_XOF_DEVICE ) ===== 全局配置 ===== 模板匹配参数 MATCH_THRESHOLD = 0.75 # 降低匹配置信度阈值以提高灵敏度 MIN_MATCH_COUNT = 10 # 最小匹配特征点数量 MIN_FRAME_INTERVAL = 0.1 # 最小检测间隔(秒) ===== 全局变量 ===== current_sample_path = “” detection_history = [] isGrabbing = False isOpen = False obj_cam_operation = None frame_monitor_thread = None template_matcher_thread = None MV_OK = 0 MV_E_CALLORDER = -2147483647 ==================== 优化后的质量检测算法 ==================== def enhanced_check_print_quality(sample_image_path, test_image, threshold=0.05): # 不再使用传感器数据调整阈值 adjusted_threshold = threshold try: sample_img_data = np.fromfile(sample_image_path, dtype=np.uint8) sample_image = cv2.imdecode(sample_img_data, cv2.IMREAD_GRAYSCALE) if sample_image is None: logging.error(f"无法解码样本图像: {sample_image_path}") return None, None, None except Exception as e: logging.exception(f"样本图像读取异常: {str(e)}") return None, None, None if len(test_image.shape) == 3: test_image_gray = cv2.cvtColor(test_image, cv2.COLOR_BGR2GRAY) else: test_image_gray = test_image.copy() sample_image = cv2.GaussianBlur(sample_image, (5, 5), 0) test_image_gray = cv2.GaussianBlur(test_image_gray, (5, 5), 0) try: # 使用更鲁棒的SIFT特征检测器 sift = cv2.SIFT_create() keypoints1, descriptors1 = sift.detectAndCompute(sample_image, None) keypoints2, descriptors2 = sift.detectAndCompute(test_image_gray, None) if descriptors1 is None or descriptors2 is None: logging.warning("无法提取特征描述符，跳过配准") aligned_sample = sample_image else: # 使用FLANN匹配器提高匹配精度 FLANN_INDEX_KDTREE = 1 index_params = dict(algorithm=FLANN_INDEX_KDTREE, trees=5) search_params = dict(checks=50) flann = cv2.FlannBasedMatcher(index_params, search_params) matches = flann.knnMatch(descriptors1, descriptors2, k=2) # 应用Lowe's比率测试筛选优质匹配 good_matches = [] for m, n in matches: if m.distance < 0.7 * n.distance: good_matches.append(m) if len(good_matches) > MIN_MATCH_COUNT: src_pts = np.float32([keypoints1[m.queryIdx].pt for m in good_matches]).reshape(-1, 1, 2) dst_pts = np.float32([keypoints2[m.trainIdx].pt for m in good_matches]).reshape(-1, 1, 2) H, mask = cv2.findHomography(src_pts, dst_pts, cv2.RANSAC, 5.0) if H is not None: aligned_sample = cv2.warpPerspective( sample_image, H, (test_image_gray.shape[1], test_image_gray.shape[0]) ) logging.info("图像配准成功，使用配准后样本") else: aligned_sample = sample_image logging.warning("无法计算单应性矩阵，使用原始样本") else: aligned_sample = sample_image logging.warning(f"特征点匹配不足({len(good_matches)}/{MIN_MATCH_COUNT})，跳过图像配准") except Exception as e: logging.error(f"图像配准失败: {str(e)}") aligned_sample = sample_image try: if aligned_sample.shape != test_image_gray.shape: test_image_gray = cv2.resize(test_image_gray, (aligned_sample.shape[1], aligned_sample.shape[0])) except Exception as e: logging.error(f"图像调整大小失败: {str(e)}") return None, None, None try: from skimage.metrics import structural_similarity as compare_ssim ssim_score, ssim_diff = compare_ssim( aligned_sample, test_image_gray, full=True, gaussian_weights=True, data_range=255 ) except ImportError: from skimage.measure import compare_ssim ssim_score, ssim_diff = compare_ssim( aligned_sample, test_image_gray, full=True, gaussian_weights=True ) except Exception as e: logging.error(f"SSIM计算失败: {str(e)}") abs_diff = cv2.absdiff(aligned_sample, test_image_gray) ssim_diff = abs_diff.astype(np.float32) / 255.0 ssim_score = 1.0 - np.mean(ssim_diff) ssim_diff = (1 - ssim_diff) * 255 abs_diff = cv2.absdiff(aligned_sample, test_image_gray) combined_diff = cv2.addWeighted(ssim_diff.astype(np.uint8), 0.7, abs_diff, 0.3, 0) _, thresholded = cv2.threshold(combined_diff, 30, 255, cv2.THRESH_BINARY) kernel = np.ones((3, 3), np.uint8) thresholded = cv2.morphologyEx(thresholded, cv2.MORPH_OPEN, kernel) thresholded = cv2.morphologyEx(thresholded, cv2.MORPH_CLOSE, kernel) diff_pixels = np.count_nonzero(thresholded) total_pixels = aligned_sample.size diff_ratio = diff_pixels / total_pixels is_qualified = diff_ratio <= adjusted_threshold marked_image = cv2.cvtColor(test_image_gray, cv2.COLOR_GRAY2BGR) marked_image[thresholded == 255] = [0, 0, 255] # 放大缺陷标记 scale_factor = 2.0 # 放大2倍 marked_image = cv2.resize(marked_image, None, fx=scale_factor, fy=scale_factor, interpolation=cv2.INTER_LINEAR) labels = skimage.measure.label(thresholded) properties = skimage.measure.regionprops(labels) for prop in properties: if prop.area > 50: y, x = prop.centroid # 根据放大比例调整坐标 x_scaled = int(x * scale_factor) y_scaled = int(y * scale_factor) cv2.putText(marked_image, f"Defect", (x_scaled, y_scaled), cv2.FONT_HERSHEY_SIMPLEX, 0.5 * scale_factor, (0, 255, 255), int(scale_factor)) return is_qualified, diff_ratio, marked_image ==================== 视觉触发的质量检测流程 ==================== def vision_controlled_check(capture_path=None): global current_sample_path, detection_history logging.info(“视觉触发质量检测启动”) # 如果没有提供抓拍路径，使用当前帧 if capture_path is None: frame = obj_cam_operation.get_current_frame() else: # 从文件加载抓拍的图像 frame = cv2.imread(capture_path) if frame is None: logging.error(f"无法加载抓拍图像: {capture_path}") frame = obj_cam_operation.get_current_frame() if frame is None: QMessageBox.warning(mainWindow, "错误", "无法获取当前帧图像！", QMessageBox.Ok) return progress = QProgressDialog("正在检测...", "取消", 0, 100, mainWindow) progress.setWindowModality(Qt.WindowModal) progress.setValue(10) try: diff_threshold = mainWindow.sliderDiffThreshold.value() / 100.0 logging.info(f"使用差异度阈值: {diff_threshold}") progress.setValue(30) is_qualified, diff_ratio, marked_image = enhanced_check_print_quality( current_sample_path, frame, threshold=diff_threshold ) progress.setValue(70) if is_qualified is None: QMessageBox.critical(mainWindow, "检测错误", "检测失败，请检查日志", QMessageBox.Ok) return logging.info(f"检测结果: 合格={is_qualified}, 差异={diff_ratio}") progress.setValue(90) update_diff_display(diff_ratio, is_qualified) result_text = f"印花是否合格: {'合格' if is_qualified else '不合格'}\n差异占比: {diff_ratio100:.2f}%\n阈值: {diff_threshold100:.2f}%" QMessageBox.information(mainWindow, "检测结果", result_text, QMessageBox.Ok) if marked_image is not None: # 创建可调整大小的窗口 cv2.namedWindow("缺陷标记结果", cv2.WINDOW_NORMAL) cv2.resizeWindow("缺陷标记结果", 800, 600) # 初始大小 cv2.imshow("缺陷标记结果", marked_image) cv2.waitKey(0) cv2.destroyAllWindows() detection_result = { 'timestamp': datetime.now(), 'qualified': is_qualified, 'diff_ratio': diff_ratio, 'threshold': diff_threshold, 'trigger_type': 'vision' if capture_path else 'manual' } detection_history.append(detection_result) update_history_display() progress.setValue(100) except Exception as e: logging.exception("印花检测失败") QMessageBox.critical(mainWindow, "检测错误", f"检测过程中发生错误: {str(e)}", QMessageBox.Ok) finally: progress.close() ==================== 相机操作函数 ==================== def open_device(): global deviceList, nSelCamIndex, obj_cam_operation, isOpen, frame_monitor_thread, mainWindow if isOpen: QMessageBox.warning(mainWindow, “Error”, ‘相机已打开!’, QMessageBox.Ok) return MV_E_CALLORDER nSelCamIndex = mainWindow.ComboDevices.currentIndex() if nSelCamIndex < 0: QMessageBox.warning(mainWindow, "Error", '请选择相机!', QMessageBox.Ok) return MV_E_CALLORDER # 创建相机控制对象 cam = MvCamera() # 初始化相机操作对象 obj_cam_operation = CameraOperation(cam, deviceList, nSelCamIndex) ret = obj_cam_operation.open_device() if 0 != ret: strError = "打开设备失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) isOpen = False else: set_continue_mode() get_param() isOpen = True enable_controls() # 创建并启动帧监控线程 frame_monitor_thread = FrameMonitorThread(obj_cam_operation) frame_monitor_thread.frame_status.connect(mainWindow.statusBar().showMessage) frame_monitor_thread.start() def start_grabbing(): global obj_cam_operation, isGrabbing, template_matcher_thread ret = obj_cam_operation.start_grabbing(mainWindow.widgetDisplay.winId()) if ret != 0: strError = "开始取流失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: isGrabbing = True enable_controls() # 等待第一帧到达 QThread.msleep(500) if not obj_cam_operation.is_frame_available(): QMessageBox.warning(mainWindow, "警告", "开始取流后未接收到帧，请检查相机连接！", QMessageBox.Ok) # 如果启用了自动检测，启动检测线程 if mainWindow.chkContinuousMatch.isChecked(): toggle_template_matching(True) def stop_grabbing(): global obj_cam_operation, isGrabbing, template_matcher_thread ret = obj_cam_operation.Stop_grabbing() if ret != 0: strError = "停止取流失败 ret:" + ToHexStr(ret) QMessageBox.warning(mainWindow, "Error", strError, QMessageBox.Ok) else: isGrabbing = False enable_controls() # 停止模板匹配线程 if template_matcher_thread and template_matcher_thread.isRunning(): template_matcher_thread.stop() def close_device(): global isOpen, isGrabbing, obj_cam_operation, frame_monitor_thread, template_matcher_thread if frame_monitor_thread and frame_monitor_thread.isRunning(): frame_monitor_thread.stop() frame_monitor_thread.wait(2000) # 停止模板匹配线程 if template_matcher_thread and template_matcher_thread.isRunning(): template_matcher_thread.stop() template_matcher_thread.wait(2000) template_matcher_thread = None if isOpen and obj_cam_operation: obj_cam_operation.close_device() isOpen = False isGrabbing = False enable_controls() ==================== 连续帧匹配检测器 ==================== class ContinuousFrameMatcher(QThread): frame_processed = pyqtSignal(np.ndarray, float, bool) # 处理后的帧, 匹配分数, 是否匹配 match_score_updated = pyqtSignal(float) # 匹配分数更新信号 match_success = pyqtSignal(np.ndarray, float) # 新增: 匹配成功信号 (帧, 匹配分数) def init(self, cam_operation, parent=None): super().init(parent) self.cam_operation = cam_operation self.running = True self.sample_template = None self.min_match_count = MIN_MATCH_COUNT self.match_threshold = MATCH_THRESHOLD self.sample_kp = None self.sample_des = None self.current_match_score = 0.0 self.last_match_time = 0 self.frame_counter = 0 self.consecutive_fail_count = 0 self.last_trigger_time = 0 # 上次触发时间 self.cool_down = 0.2 # 冷却时间(秒) # 特征检测器 - 使用SIFT self.sift = cv2.SIFT_create() # 特征匹配器 - 使用FLANN提高匹配精度 FLANN_INDEX_KDTREE = 1 index_params = dict(algorithm=FLANN_INDEX_KDTREE, trees=5) search_params = dict(checks=50) self.flann = cv2.FlannBasedMatcher(index_params, search_params) # 性能监控 self.processing_times = deque(maxlen=100) self.frame_rates = deque(maxlen=100) def set_sample(self, sample_img): """设置标准样本""" # ...现有代码不变... def process_frame(self, frame): is_matched=False match_score=0.0 processed_frame = frame.copy() # 默认返回原始帧 """处理帧：特征提取、匹配和可视化""" # 检查是否匹配成功且超过冷却时间 current_time = time.time() if is_matched and (current_time - self.last_trigger_time) > self.cool_down: self.last_trigger_time = current_time logging.info(f"匹配成功! 分数: {match_score:.2f}, 触发质量检测") # 发出匹配成功信号 self.match_success.emit(frame.copy(), match_score) return processed_frame, match_score, is_matched def run(self): """主处理循环 - 连续处理每一帧""" logging.info("连续帧匹配线程启动") self.last_match_time = time.time() self.consecutive_fail_count = 0 while self.running: start_time = time.time() # 检查相机状态 if not self.cam_operation or not self.cam_operation.is_grabbing: if self.consecutive_fail_count % 10 == 0: logging.debug("相机未取流，等待...") time.sleep(0.1) self.consecutive_fail_count += 1 continue # 获取当前帧 frame = self.cam_operation.get_current_frame() if frame is None: self.consecutive_fail_count += 1 if self.consecutive_fail_count % 10 == 0: logging.warning(f"连续{self.consecutive_fail_count}次获取帧失败") time.sleep(0.05) continue self.consecutive_fail_count = 0 try: # 处理帧 processed_frame, match_score, is_matched = self.process_frame(frame) # 发送处理结果 self.frame_processed.emit(processed_frame, match_score, is_matched) except Exception as e: logging.error(f"帧处理错误: {str(e)}") # 控制处理频率 processing_time = time.time() - start_time sleep_time = max(0.01, MIN_FRAME_INTERVAL - processing_time) time.sleep(sleep_time) logging.info("连续帧匹配线程退出") ==================== 模板匹配控制函数 ==================== def toggle_template_matching(state): global template_matcher_thread, current_sample_path logging.debug(f"切换连续匹配状态: {state}") if state == Qt.Checked and isGrabbing: # 确保已设置样本 if not current_sample_path: logging.warning("尝试启动连续匹配但未设置样本") QMessageBox.warning(mainWindow, "错误", "请先设置标准样本", QMessageBox.Ok) mainWindow.chkContinuousMatch.setChecked(False) return if template_matcher_thread is None: logging.info("创建新的连续帧匹配线程") template_matcher_thread = ContinuousFrameMatcher(obj_cam_operation) template_matcher_thread.frame_processed.connect(update_frame_display) template_matcher_thread.match_score_updated.connect(update_match_score_display) # 连接匹配成功信号 template_matcher_thread.match_success.connect( lambda frame, score: vision_controlled_check(frame, score) ) # 加载样本图像 sample_img = cv2.imread(current_sample_path) if sample_img is None: logging.error("无法加载标准样本图像") QMessageBox.warning(mainWindow, "错误", "无法加载标准样本图像", QMessageBox.Ok) mainWindow.chkContinuousMatch.setChecked(False) return if not template_matcher_thread.set_sample(sample_img): logging.warning("标准样本特征不足") QMessageBox.warning(mainWindow, "错误", "标准样本特征不足", QMessageBox.Ok) mainWindow.chkContinuousMatch.setChecked(False) return if not template_matcher_thread.isRunning(): logging.info("启动连续帧匹配线程") template_matcher_thread.start() elif template_matcher_thread and template_matcher_thread.isRunning(): logging.info("停止连续帧匹配线程") template_matcher_thread.stop() # 重置匹配分数显示 update_match_score_display(0.0) # 重置帧显示 if obj_cam_operation and obj_cam_operation.is_frame_available(): frame = obj_cam_operation.get_current_frame() if frame is not None: display_frame = frame.copy() # 添加状态信息 cv2.putText(display_frame, "Continuous Matching Disabled", (20, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2) update_frame_display(display_frame, 0.0, False) ==================== UI更新函数 ==================== def update_frame_display(frame, match_score, is_matched): “”“更新主显示窗口（线程安全）”“” # 确保在GUI线程中执行 if QThread.currentThread() != QApplication.instance().thread(): QMetaObject.invokeMethod(mainWindow, "updateDisplay", Qt.QueuedConnection, Q_ARG(np.ndarray, frame), Q_ARG(float, match_score), Q_ARG(bool, is_matched)) return # 原始显示逻辑 if len(frame.shape) == 3: # 彩色图像 h, w, ch = frame.shape bytes_per_line = ch * w q_img = QImage(frame.data, w, h, bytes_per_line, QImage.Format_RGB888).rgbSwapped() else: # 灰度图像 h, w = frame.shape bytes_per_line = w q_img = QImage(frame.data, w, h, bytes_per_line, QImage.Format_Grayscale8) # 创建QPixmap并缩放以适应显示区域 pixmap = QPixmap.fromImage(q_img) scaled_pixmap = pixmap.scaled( mainWindow.widgetDisplay.width(), mainWindow.widgetDisplay.height(), Qt.KeepAspectRatio, Qt.SmoothTransformation ) # 更新显示 mainWindow.widgetDisplay.setPixmap(scaled_pixmap) # 更新匹配分数显示 update_match_score_display(match_score) 在MainWindow类中添加 class MainWindow(QMainWindow): # …现有代码… @pyqtSlot(np.ndarray, float, bool) def updateDisplay(self, frame, match_score, is_matched): """线程安全的显示更新方法""" # 将OpenCV图像转换为Qt图像 if len(frame.shape) == 3: # 彩色图像 h, w, ch = frame.shape bytes_per_line = ch * w q_img = QImage(frame.data, w, h, bytes_per_line, QImage.Format_RGB888).rgbSwapped() else: # 灰度图像 h, w = frame.shape bytes_per_line = w q_img = QImage(frame.data, w, h, bytes_per_line, QImage.Format_Grayscale8) # 创建QPixmap并缩放 pixmap = QPixmap.fromImage(q_img) scaled_pixmap = pixmap.scaled( self.widgetDisplay.width(), self.widgetDisplay.height(), Qt.KeepAspectRatio, Qt.SmoothTransformation ) # 更新显示 self.widgetDisplay.setPixmap(scaled_pixmap) update_match_score_display(match_score) ==================== UI更新函数 ==================== def update_frame_display(frame, match_score, is_matched): “”“更新主显示窗口”“” # 将OpenCV图像转换为Qt图像 if len(frame.shape) == 3: # 彩色图像 h, w, ch = frame.shape bytes_per_line = ch * w q_img = QImage(frame.data, w, h, bytes_per_line, QImage.Format_RGB888).rgbSwapped() else: # 灰度图像 h, w = frame.shape bytes_per_line = w q_img = QImage(frame.data, w, h, bytes_per_line, QImage.Format_Grayscale8) # 创建QPixmap并缩放以适应显示区域 pixmap = QPixmap.fromImage(q_img) scaled_pixmap = pixmap.scaled( mainWindow.widgetDisplay.width(), mainWindow.widgetDisplay.height(), Qt.KeepAspectRatio, Qt.SmoothTransformation ) # 更新显示 mainWindow.widgetDisplay.setPixmap(scaled_pixmap) # 更新匹配分数显示 update_match_score_display(match_score) def update_match_score_display(score): “”“更新匹配分数显示”“” # 将分数转换为百分比显示 score_percent = score * 100 mainWindow.lblMatchScoreValue.setText(f"{score_percent:.1f}%") # 根据分数设置颜色 if score > 0.8: # 高于80%显示绿色 color = "green" elif score > 0.6: # 60%-80%显示黄色 color = "orange" else: # 低于60%显示红色 color = "red" mainWindow.lblMatchScoreValue.setStyleSheet(f"color: {color}; font-weight: bold;") def update_diff_display(diff_ratio, is_qualified): mainWindow.lblCurrentDiff.setText(f"当前差异度: {diff_ratio100:.2f}%") if is_qualified: mainWindow.lblDiffStatus.setText(“状态: 合格”) mainWindow.lblDiffStatus.setStyleSheet(“color: green; font-size: 12px;”) else: mainWindow.lblDiffStatus.setText(“状态: 不合格”) mainWindow.lblDiffStatus.setStyleSheet(“color: red; font-size: 12px;”) def update_diff_threshold(value): mainWindow.lblDiffValue.setText(f"{value}%") def update_sample_display(): global current_sample_path if current_sample_path: mainWindow.lblSamplePath.setText(f"当前样本: {os.path.basename(current_sample_path)}") mainWindow.lblSamplePath.setToolTip(current_sample_path) mainWindow.bnPreviewSample.setEnabled(True) else: mainWindow.lblSamplePath.setText(“当前样本: 未设置样本”) mainWindow.bnPreviewSample.setEnabled(False) def update_history_display(): global detection_history mainWindow.cbHistory.clear() for i, result in enumerate(detection_history[-10:]): timestamp = result[‘timestamp’].strftime(“%H:%M:%S”) status = “合格” if result[‘qualified’] else “不合格” ratio = f"{result[‘diff_ratio’]100:.2f}%" trigger = “视觉” if result[‘trigger_type’] == ‘vision’ else “手动” mainWindow.cbHistory.addItem(f"[{trigger} {timestamp}] {status} - 差异: {ratio}") ==================== 主窗口类 ==================== class MainWindow(QMainWindow): def init(self): super().init() self.setWindowTitle(“布料印花检测系统 - 连续匹配版”) self.resize(1200, 800) central_widget = QWidget() self.setCentralWidget(central_widget) main_layout = QVBoxLayout(central_widget) # 设备枚举区域 device_layout = QHBoxLayout() self.ComboDevices = QComboBox() self.bnEnum = QPushButton("枚举设备") self.bnOpen = QPushButton("打开设备") self.bnClose = QPushButton("关闭设备") device_layout.addWidget(self.ComboDevices) device_layout.addWidget(self.bnEnum) device_layout.addWidget(self.bnOpen) device_layout.addWidget(self.bnClose) main_layout.addLayout(device_layout) # 取流控制组 self.groupGrab = QGroupBox("取流控制") grab_layout = QHBoxLayout(self.groupGrab) self.bnStart = QPushButton("开始取流") self.bnStop = QPushButton("停止取流") self.radioContinueMode = QRadioButton("连续模式") self.radioTriggerMode = QRadioButton("触发模式") self.bnSoftwareTrigger = QPushButton("软触发") grab_layout.addWidget(self.bnStart) grab_layout.addWidget(self.bnStop) grab_layout.addWidget(self.radioContinueMode) grab_layout.addWidget(self.radioTriggerMode) grab_layout.addWidget(self.bnSoftwareTrigger) main_layout.addWidget(self.groupGrab) # 参数设置组 self.paramgroup = QGroupBox("相机参数") param_layout = QGridLayout(self.paramgroup) self.edtExposureTime = QLineEdit() self.edtGain = QLineEdit() self.edtFrameRate = QLineEdit() self.bnGetParam = QPushButton("获取参数") self.bnSetParam = QPushButton("设置参数") self.bnSaveImage = QPushButton("保存图像") param_layout.addWidget(QLabel("曝光时间:"), 0, 0) param_layout.addWidget(self.edtExposureTime, 0, 1) param_layout.addWidget(self.bnGetParam, 0, 2) param_layout.addWidget(QLabel("增益:"), 1, 0) param_layout.addWidget(self.edtGain, 1, 1) param_layout.addWidget(self.bnSetParam, 1, 2) param_layout.addWidget(QLabel("帧率:"), 2, 0) param_layout.addWidget(self.edtFrameRate, 2, 1) param_layout.addWidget(self.bnSaveImage, 2, 2) main_layout.addWidget(self.paramgroup) # 图像显示区域 self.widgetDisplay = QLabel() self.widgetDisplay.setMinimumSize(640, 480) self.widgetDisplay.setStyleSheet("background-color: black;") self.widgetDisplay.setAlignment(Qt.AlignCenter) self.widgetDisplay.setText("相机预览区域") main_layout.addWidget(self.widgetDisplay, 1) # 创建自定义UI组件 self.setup_custom_ui() def setup_custom_ui(self): # 工具栏 toolbar = self.addToolBar("检测工具") self.bnCheckPrint = QPushButton("手动检测") self.bnSaveSample = QPushButton("保存标准样本") self.bnPreviewSample = QPushButton("预览样本") self.cbHistory = QComboBox() self.cbHistory.setMinimumWidth(300) toolbar.addWidget(self.bnCheckPrint) toolbar.addWidget(self.bnSaveSample) toolbar.addWidget(self.bnPreviewSample) toolbar.addWidget(QLabel("历史记录:")) toolbar.addWidget(self.cbHistory) # 状态栏样本路径 self.lblSamplePath = QLabel("当前样本: 未设置样本") self.statusBar().addPermanentWidget(self.lblSamplePath) # 右侧面板 right_panel = QWidget() right_layout = QVBoxLayout(right_panel) right_layout.setContentsMargins(10, 10, 10, 10) # 差异度调整组 diff_group = QGroupBox("差异度调整") diff_layout = QVBoxLayout(diff_group) self.lblDiffThreshold = QLabel("差异度阈值 (0-100%):") self.sliderDiffThreshold = QSlider(Qt.Horizontal) self.sliderDiffThreshold.setRange(0, 100) self.sliderDiffThreshold.setValue(5) self.lblDiffValue = QLabel("5%") self.lblCurrentDiff = QLabel("当前差异度: -") self.lblCurrentDiff.setStyleSheet("font-size: 14px; font-weight: bold;") self.lblDiffStatus = QLabel("状态: 未检测") self.lblDiffStatus.setStyleSheet("font-size: 12px;") diff_layout.addWidget(self.lblDiffThreshold) diff_layout.addWidget(self.sliderDiffThreshold) diff_layout.addWidget(self.lblDiffValue) diff_layout.addWidget(self.lblCurrentDiff) diff_layout.addWidget(self.lblDiffStatus) right_layout.addWidget(diff_group) # ===== 连续匹配面板 ===== match_group = QGroupBox("连续帧匹配") match_layout = QVBoxLayout(match_group) # 样本设置 sample_layout = QHBoxLayout() self.bnSetSample = QPushButton("设置标准样本") self.bnPreviewSample = QPushButton("预览样本") self.lblSampleStatus = QLabel("状态: 未设置样本") sample_layout.addWidget(self.bnSetSample) sample_layout.addWidget(self.bnPreviewSample) sample_layout.addWidget(self.lblSampleStatus) match_layout.addLayout(sample_layout) # 匹配参数 param_layout = QHBoxLayout() self.lblMatchThreshold = QLabel("匹配阈值:") self.sliderThreshold = QSlider(Qt.Horizontal) self.sliderThreshold.setRange(50, 100) self.sliderThreshold.setValue(75) # 降低默认阈值 self.lblThresholdValue = QLabel("75%") param_layout.addWidget(self.lblMatchThreshold) param_layout.addWidget(self.sliderThreshold) param_layout.addWidget(self.lblThresholdValue) match_layout.addLayout(param_layout) # 匹配分数显示 match_score_layout = QHBoxLayout() self.lblMatchScore = QLabel("实时匹配分数:") self.lblMatchScoreValue = QLabel("0.0%") self.lblMatchScoreValue.setStyleSheet("font-weight: bold;") match_score_layout.addWidget(self.lblMatchScore) match_score_layout.addWidget(self.lblMatchScoreValue) match_layout.addLayout(match_score_layout) # 连续匹配开关 self.chkContinuousMatch = QCheckBox("启用连续帧匹配") self.chkContinuousMatch.setChecked(False) match_layout.addWidget(self.chkContinuousMatch) right_layout.addWidget(match_group) right_layout.addStretch(1) # 停靠窗口 dock = QDockWidget("检测控制面板", self) dock.setWidget(right_panel) dock.setFeatures(QDockWidget.DockWidgetMovable | QDockWidget.DockWidgetFloatable) self.addDockWidget(Qt.RightDockWidgetArea, dock) def setup_connections(self): # 连接匹配成功信号到质量检测 if template_matcher_thread: template_matcher_thread.match_success.connect( lambda frame, score: vision_controlled_check(frame, score) ) def closeEvent(self, event): logging.info("主窗口关闭，执行清理...") close_device() event.accept() ===== 辅助函数 ===== def ToHexStr(num): if not isinstance(num, int): try: num = int(num) except: return f"<非整数:{type(num)}>" chaDic = {10: ‘a’, 11: ‘b’, 12: ‘c’, 13: ‘d’, 14: ‘e’, 15: ‘f’} hexStr = “” if num < 0: num = num + 2 ** 32 while num >= 16: digit = num % 16 hexStr = chaDic.get(digit, str(digit)) + hexStr num //= 16 hexStr = chaDic.get(num, str(num)) + hexStr return “0x” + hexStr def enum_devices(): global deviceList, obj_cam_operation n_layer_type = ( MV_GIGE_DEVICE | MV_USB_DEVICE | MV_GENTL_CAMERALINK_DEVICE | MV_GENTL_CXP_DEVICE | MV_GENTL_XOF_DEVICE ) # 创建设备列表 deviceList = MV_CC_DEVICE_INFO_LIST() # 枚举设备 ret = MvCamera.MV_CC_EnumDevices(n_layer_type, deviceList) if ret != MV_OK: error_msg = f"枚举设备失败! 错误码: 0x{ret:x}" logging.error(error_msg) QMessageBox.warning(mainWindow, "错误", error_msg, QMessageBox.Ok) return ret if deviceList.nDeviceNum == 0: QMessageBox.warning(mainWindow, "提示", "未找到任何设备", QMessageBox.Ok) return MV_OK logging.info(f"找到 {deviceList.nDeviceNum} 个设备") # 处理设备信息 devList = [] for i in range(deviceList.nDeviceNum): # 获取设备信息 mvcc_dev_info = ctypes.cast( deviceList.pDeviceInfo[i], ctypes.POINTER(MV_CC_DEVICE_INFO) ).contents # 根据设备类型提取信息 if mvcc_dev_info.nTLayerType == MV_GIGE_DEVICE: st_gige_info = mvcc_dev_info.SpecialInfo.stGigEInfo ip_addr = ( f"{(st_gige_info.nCurrentIp >> 24) & 0xFF}." f"{(st_gige_info.nCurrentIp >> 16) & 0xFF}." f"{(st_gige_info.nCurrentIp >> 8) & 0xFF}." f"{st_gige_info.nCurrentIp & 0xFF}" ) # 修复：将c_ubyte_Array_16转换为字节串再解码 user_defined_bytes = bytes(st_gige_info.chUserDefinedName) dev_name = f"GigE: {user_defined_bytes.decode('gbk', 'ignore')}" devList.append(f"[{i}] {dev_name} ({ip_addr})") elif mvcc_dev_info.nTLayerType == MV_USB_DEVICE: st_usb_info = mvcc_dev_info.SpecialInfo.stUsb3VInfo serial = bytes(st_usb_info.chSerialNumber).decode('ascii', 'ignore').rstrip('\x00') # 修复：同样处理用户自定义名称 user_defined_bytes = bytes(st_usb_info.chUserDefinedName) dev_name = f"USB: {user_defined_bytes.decode('gbk', 'ignore')}" devList.append(f"[{i}] {dev_name} (SN: {serial})") else: devList.append(f"[{i}] 未知设备类型: {mvcc_dev_info.nTLayerType}") # 更新UI mainWindow.ComboDevices.clear() mainWindow.ComboDevices.addItems(devList) if devList: mainWindow.ComboDevices.setCurrentIndex(0) mainWindow.statusBar().showMessage(f"找到 {deviceList.nDeviceNum} 个设备", 3000) return MV_OK def set_continue_mode(): ret = obj_cam_operation.set_trigger_mode(False) if ret != 0: strError = “设置连续模式失败 ret:” + ToHexStr(ret) QMessageBox.warning(mainWindow, “Error”, strError, QMessageBox.Ok) else: mainWindow.radioContinueMode.setChecked(True) mainWindow.radioTriggerMode.setChecked(False) mainWindow.bnSoftwareTrigger.setEnabled(False) def set_software_trigger_mode(): ret = obj_cam_operation.set_trigger_mode(True) if ret != 0: strError = “设置触发模式失败 ret:” + ToHexStr(ret) QMessageBox.warning(mainWindow, “Error”, strError, QMessageBox.Ok) else: mainWindow.radioContinueMode.setChecked(False) mainWindow.radioTriggerMode.setChecked(True) mainWindow.bnSoftwareTrigger.setEnabled(isGrabbing) def trigger_once(): ret = obj_cam_operation.trigger_once() if ret != 0: strError = “软触发失败 ret:” + ToHexStr(ret) QMessageBox.warning(mainWindow, “Error”, strError, QMessageBox.Ok) def save_sample_image(): global isGrabbing, obj_cam_operation, current_sample_path if not isGrabbing: QMessageBox.warning(mainWindow, "错误", "请先开始取流并捕获图像！", QMessageBox.Ok) return # 尝试捕获当前帧 frame = obj_cam_operation.capture_frame() if frame is None: QMessageBox.warning(mainWindow, "无有效图像", "未捕获到有效图像，请检查相机状态！", QMessageBox.Ok) return # 确保图像有效 if frame.size == 0 or frame.shape[0] == 0 or frame.shape[1] == 0: QMessageBox.warning(mainWindow, "无效图像", "捕获的图像无效，请检查相机设置！", QMessageBox.Ok) return settings = QSettings("ClothInspection", "CameraApp") last_dir = settings.value("last_save_dir", os.path.join(os.getcwd(), "captures")) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") default_filename = f"sample_{timestamp}" file_path, selected_filter = QFileDialog.getSaveFileName( mainWindow, "保存标准样本图像", os.path.join(last_dir, default_filename), "BMP Files (.bmp);;PNG Files (.png);;JPEG Files (.jpg);;所有文件 ()", options=QFileDialog.DontUseNativeDialog ) if not file_path: return # 确保文件扩展名正确 file_extension = os.path.splitext(file_path)[1].lower() if not file_extension: if "BMP" in selected_filter: file_path += ".bmp" elif "PNG" in selected_filter: file_path += ".png" elif "JPEG" in selected_filter or "JPG" in selected_filter: file_path += ".jpg" else: file_path += ".bmp" file_extension = os.path.splitext(file_path)[1].lower() # 创建目录（如果不存在） directory = os.path.dirname(file_path) if directory and not os.path.exists(directory): try: os.makedirs(directory, exist_ok=True) except OSError as e: QMessageBox.critical(mainWindow, "目录创建错误", f"无法创建目录 {directory}: {str(e)}", QMessageBox.Ok) return # 保存图像 try: # 使用OpenCV保存图像 if not cv2.imwrite(file_path, frame): raise Exception("OpenCV保存失败") # 更新状态 current_sample_path = file_path update_sample_display() settings.setValue("last_save_dir", os.path.dirname(file_path)) # 显示成功消息 QMessageBox.information(mainWindow, "成功", f"标准样本已保存至:\n{file_path}", QMessageBox.Ok) # 更新样本状态 mainWindow.lblSampleStatus.setText("状态: 样本已设置") mainWindow.lblSampleStatus.setStyleSheet("color: green;") except Exception as e: logging.error(f"保存图像失败: {str(e)}") QMessageBox.critical(mainWindow, "保存错误", f"保存图像时发生错误:\n{str(e)}", QMessageBox.Ok) def preview_sample(): global current_sample_path if not current_sample_path or not os.path.exists(current_sample_path): QMessageBox.warning(mainWindow, “错误”, “请先设置有效的标准样本图像！”, QMessageBox.Ok) return try: # 直接使用OpenCV加载图像 sample_img = cv2.imread(current_sample_path) if sample_img is None: raise Exception("无法加载图像") # 显示图像 cv2.namedWindow("标准样本预览", cv2.WINDOW_NORMAL) cv2.resizeWindow("标准样本预览", 800, 600) cv2.imshow("标准样本预览", sample_img) cv2.waitKey(0) cv2.destroyAllWindows() except Exception as e: QMessageBox.warning(mainWindow, "错误", f"预览样本失败: {str(e)}", QMessageBox.Ok) def is_float(str): try: float(str) return True except ValueError: return False def get_param(): try: ret = obj_cam_operation.get_parameters() if ret != MV_OK: strError = “获取参数失败，错误码: " + ToHexStr(ret) QMessageBox.warning(mainWindow, “错误”, strError, QMessageBox.Ok) else: mainWindow.edtExposureTime.setText(”{0:.2f}“.format(obj_cam_operation.exposure_time)) mainWindow.edtGain.setText(”{0:.2f}“.format(obj_cam_operation.gain)) mainWindow.edtFrameRate.setText(”{0:.2f}“.format(obj_cam_operation.frame_rate)) except Exception as e: error_msg = f"获取参数时发生错误: {str(e)}” QMessageBox.critical(mainWindow, “严重错误”, error_msg, QMessageBox.Ok) def set_param(): frame_rate = mainWindow.edtFrameRate.text() exposure = mainWindow.edtExposureTime.text() gain = mainWindow.edtGain.text() if not (is_float(frame_rate) and is_float(exposure) and is_float(gain)): strError = “设置参数失败: 参数必须是有效的浮点数” QMessageBox.warning(mainWindow, “错误”, strError, QMessageBox.Ok) return MV_E_PARAMETER try: ret = obj_cam_operation.set_param( frame_rate=float(frame_rate), exposure_time=float(exposure), gain=float(gain) ) if ret != MV_OK: strError = “设置参数失败，错误码: " + ToHexStr(ret) QMessageBox.warning(mainWindow, “错误”, strError, QMessageBox.Ok) except Exception as e: error_msg = f"设置参数时发生错误: {str(e)}” QMessageBox.critical(mainWindow, “严重错误”, error_msg, QMessageBox.Ok) def enable_controls(): global isGrabbing, isOpen mainWindow.groupGrab.setEnabled(isOpen) mainWindow.paramgroup.setEnabled(isOpen) mainWindow.bnOpen.setEnabled(not isOpen) mainWindow.bnClose.setEnabled(isOpen) mainWindow.bnStart.setEnabled(isOpen and (not isGrabbing)) mainWindow.bnStop.setEnabled(isOpen and isGrabbing) mainWindow.bnSoftwareTrigger.setEnabled(isGrabbing and mainWindow.radioTriggerMode.isChecked()) mainWindow.bnSaveImage.setEnabled(isOpen and isGrabbing) mainWindow.bnCheckPrint.setEnabled(isOpen and isGrabbing) mainWindow.bnSaveSample.setEnabled(isOpen and isGrabbing) mainWindow.bnPreviewSample.setEnabled(bool(current_sample_path)) # 连续匹配控制 mainWindow.chkContinuousMatch.setEnabled(bool(current_sample_path) and isGrabbing) ===== 相机帧监控线程 ===== class FrameMonitorThread(QThread): frame_status = pyqtSignal(str) # 用于发送状态消息的信号 def init(self, cam_operation): super().init() self.cam_operation = cam_operation self.running = True self.frame_count = 0 self.last_time = time.time() def run(self): """监控相机帧状态的主循环""" while self.running: try: if self.cam_operation and self.cam_operation.is_grabbing: # 获取帧统计信息 frame_info = self.get_frame_info() if frame_info: fps = frame_info.get('fps', 0) dropped = frame_info.get('dropped', 0) status = f"FPS: {fps:.1f} | 丢帧: {dropped}" self.frame_status.emit(status) else: self.frame_status.emit("取流中...") else: self.frame_status.emit("相机未取流") except Exception as e: self.frame_status.emit(f"监控错误: {str(e)}") # 每500ms检查一次 QThread.msleep(500) def stop(self): """停止监控线程""" self.running = False self.wait(1000) # 等待线程结束 def calculate_fps(self): """计算当前帧率""" current_time = time.time() elapsed = current_time - self.last_time if elapsed > 0: fps = self.frame_count / elapsed self.frame_count = 0 self.last_time = current_time return fps return 0 def get_frame_info(self): """获取帧信息""" try: # 更新帧计数 self.frame_count += 1 # 返回帧信息 return { 'fps': self.calculate_fps(), 'dropped': 0 # 实际应用中需要从相机获取真实丢帧数 } except Exception as e: logging.error(f"获取帧信息失败: {str(e)}") return None ===== 主程序入口 ===== if name == “main”: # 配置日志系统 logging.basicConfig( level=logging.DEBUG, format=‘%(asctime)s - %(name)s - %(levelname)s - %(message)s’, handlers=[ logging.FileHandler(“cloth_inspection_continuous.log”), logging.StreamHandler() ] ) logging.info(“布料印花检测系统(连续匹配版)启动”) app = QApplication(sys.argv) mainWindow = MainWindow() # 信号连接 mainWindow.sliderDiffThreshold.valueChanged.connect(update_diff_threshold) mainWindow.bnCheckPrint.clicked.connect(lambda: vision_controlled_check(None)) mainWindow.bnSaveSample.clicked.connect(save_sample_image) mainWindow.bnPreviewSample.clicked.connect(preview_sample) mainWindow.bnEnum.clicked.connect(enum_devices) mainWindow.bnOpen.clicked.connect(open_device) mainWindow.bnClose.clicked.connect(close_device) mainWindow.bnStart.clicked.connect(start_grabbing) mainWindow.bnStop.clicked.connect(stop_grabbing) mainWindow.bnSoftwareTrigger.clicked.connect(trigger_once) mainWindow.radioTriggerMode.clicked.connect(set_software_trigger_mode) mainWindow.radioContinueMode.clicked.connect(set_continue_mode) mainWindow.bnGetParam.clicked.connect(get_param) mainWindow.bnSetParam.clicked.connect(set_param) mainWindow.bnSaveImage.clicked.connect(save_sample_image) # 连续匹配信号连接 mainWindow.sliderThreshold.valueChanged.connect(update_match_score_display) mainWindow.chkContinuousMatch.stateChanged.connect(toggle_template_matching) mainWindow.show() app.exec_() close_device() sys.exit()

pts = np.float32([[0, 0], [0, h-1], [w-1, h-1], [w-1, 0]]).reshape(-1, 1, 2) dst = cv2.perspectiveTransform(pts, M) processed_frame = cv2.polylines(processed_frame, [np.int32(dst)], True, (0, 255, ...

Traceback (most recent call last): File "D:\桌面\pythonProject9\均匀采样.py", line 44, in <module> downsample_point_cloud(input_path, output_path, voxel_size=0.2) File "D:\桌面\pythonProject9\均匀采样.py", line 15, in downsample_point_cloud points = data.reshape((-1, 4)) # 假设每个点包含x,y,z,intensity ValueError: cannot reshape array of size 632979821 into shape (4)

valid_dims = [d for d in [3,4,6] if total_elements % d == 0] # 常见维度 if not valid_dims: raise ValueError(f"无法确定数据维度，总元素数{total_elements}不符合常见维度") dim = valid_dims[0] # ...

import os import pandas as pd import matplotlib.pyplot as plt from mpl_toolkits.mplot3d import Axes3D import numpy as np # ----------------------------- # 参数设定 # ----------------------------- # 文件夹路径 data_dir = 'D:/MyPythonProjects/ev_supplychain_sim/data/遗忘率创新性实验' # 根据你的路径调整 # 策略组合标签（与文件名中的 ll, lm 等对应） strategies = ['ll', 'lm', 'lh', 'ml', 'mm', 'mh', 'hl', 'hm', 'hh'] # 生成文件路径字典 file_map = {s: os.path.join(data_dir, f'knowledge_evolution{s}.csv') for s in strategies} # ----------------------------- # 读取数据并组织成表格 # ----------------------------- all_data = [] for strat in strategies: df = pd.read_csv(file_map[strat], sep='\t' if '.tsv' in file_map[strat] else ',') for _, row in df.iterrows(): all_data.append({ 'Step': row['Step'], 'Strategy': strat, 'Total_Average': row['Total_Average'] }) # 转为 DataFrame df_all = pd.DataFrame(all_data) # ----------------------------- # 创建网格数据用于3D Surface绘图 # ----------------------------- # 将策略标签映射为数字以用于坐标轴 strategy_to_num = {s: i for i, s in enumerate(strategies)} df_all['Strategy_Num'] = df_all['Strategy'].map(strategy_to_num) # 构建 meshgrid step_vals = sorted(df_all['Step'].unique()) strategy_vals = list(range(len(strategies))) # 创建Z值矩阵 Z = np.zeros((len(step_vals), len(strategy_vals))) for i, step in enumerate(step_vals): for j, s in enumerate(strategies): val = df_all[(df_all['Step'] == step) & (df_all['Strategy'] == s)]['Total_Average'].values Z[i, j] = val[0] if len(val) > 0 else np.nan X, Y = np.meshgrid(strategy_vals, step_vals) # ----------------------------- # 绘制3D图 # ----------------------------- fig = plt.figure(figsize=(12, 8)) ax = fig.add_subplot(111, projection='3d') surf = ax.plot_surface(X, Y, Z, cmap='viridis', edgecolor='k', linewidth=0.3, antialiased=True) # 设置坐标轴标签 ax.set_xlabel('Strategy') ax.set_ylabel('Step') ax.set_zlabel('Total_Average') # 设置策略标签顺序 ax.set_xticks(range(len(strategies))) # 用 0 到 8 的索引 ax.set_xticklabels(strategies) # 对应标签按指定顺序排列 # # 设置策略标签 # ax.set_xticks(strategy_vals) # ax.set_xticklabels(strategies) # 标题和颜色条 ax.set_title('Total_Average over Steps and Strategies') fig.colorbar(surf, shrink=0.5, aspect=10) plt.tight_layout() plt.show()

Z[i, j] = val[0] if len(val) > 0 else np.nan X, Y = np.meshgrid(strategy_vals, step_vals) # 插值填充缺失值 X_flat = np.array([x for x in range(len(strategies)) for _ in step_vals]) Y_flat = np....

详细解释total_up_1 = np.array([0, 130, 80]) total_down_1 = np.array([179, 255, 255]) total_mask_1 = cv2.inRange(hsv_roi, total_up_1, total_down_1)

解释# HSV范围1 total_up_1 = np.array([0, 130, 80]) total_down_1 = np.array([179, 255, 255]) total_mask_1 = cv2.inRange(hsv_roi, total_up_1, total_down_1)

解释 total_up_1 = np.array([0, 130, 80]) total_down_1 = np.array([179, 255, 255]) total_mask_1 = cv2.inRange(hsv_roi, total_up_1, total_down_1)中的三个数值

相关推荐

详细解释total_up_1 = np.array([0, 130, 80]) total_down_1 = np.array([179, 255, 255]) total_mask_1 = cv2.inRange(hsv_roi, total_up_1, total_down_1)

解释# HSV范围1 total_up_1 = np.array([0, 130, 80]) total_down_1 = np.array([179, 255, 255]) total_mask_1 = cv2.inRange(hsv_roi, total_up_1, total_down_1)

解释 total_up_1 = np.array([0, 130, 80]) total_down_1 = np.array([179, 255, 255]) total_mask_1 = cv2.inRange(hsv_roi, total_up_1, total_down_1)中的三个数值

相关推荐

total_word_feature_extractor_zh.dat

Aspose.Total_for_.NET_2017.2.28.part1

Total_Control_7_0_0_u10_Install_x64.zip

total_up_1 = np.array([0, 130, 80]) total_down_1 = np.array([179, 255, 255]) total_mask_1 = cv2.inRange(hsv_roi, total_up_1, total_down_1)

langchain4j-anthropic-spring-boot-starter-0.31.0.jar中文文档.zip

TMS320F28335电机控制程序详解：BLDC、PMSM无感有感及异步VF源代码与开发资料

基于爬山搜索法的风力发电MPPT控制Simulink仿真：定步长与变步长算法性能对比 - 爬山搜索法 最新版

基于MatlabSimulink的风电场调频策略研究：虚拟惯性、超速减载与下垂控制的协调优化

大家在看

机械臂建模+MATLAB代码+六自由度.zip

易语言WinSock模块应用

VxWorks和RTlinux的性能测试分析

波特率任意设 串口调试助手

十几种水下图像增强算法源代码

最新推荐

langchain4j-anthropic-spring-boot-starter-0.31.0.jar中文文档.zip

TMS320F28335电机控制程序详解：BLDC、PMSM无感有感及异步VF源代码与开发资料

基于爬山搜索法的风力发电MPPT控制Simulink仿真：定步长与变步长算法性能对比 - 爬山搜索法 最新版

基于MatlabSimulink的风电场调频策略研究：虚拟惯性、超速减载与下垂控制的协调优化

三菱QL系列PLC在3C-FPC组装机中的定位与伺服控制及触摸屏应用解析

Visual C++.NET编程技术实战指南

HarmonyOS内核深度探秘：优化自由行旅游系统的策略

tkinter模块所有控件

局域网五子棋游戏：娱乐与聊天的完美结合

自由行旅游新篇章：HarmonyOS技术融合与系统架构深度解析

基于爬山搜索法的风力发电MPPT控制Simulink仿真：定步长与变步长算法性能对比 - 爬山搜索法最新版

波特率任意设串口调试助手

基于爬山搜索法的风力发电MPPT控制Simulink仿真：定步长与变步长算法性能对比 - 爬山搜索法最新版