替代cmd的匿名管道双向通讯技术资源-CSDN下载

共6个文件

h：2个

cpp：2个

vcproj：1个

匿名管道

5星 · 超过95%的资源需积分: 50 30 浏览量 2013-02-01 18:20:50 上传评论 5 收藏 4KB RAR 举报

匿名管道是一种在进程间通信（IPC，Inter-Process Communication）技术中常见的方法，尤其在Windows操作系统环境下。它允许两个相关联的进程之间进行数据交换，这些进程通常是由同一个父进程创建的子进程。在"匿名管道双向通讯"的场景中，我们可以理解为一个进程可以既是数据的发送方也是接收方，这与单向管道（只能从一端写入，从另一端读出）有所不同。匿名管道的特点包括： 1. **非持久性**：匿名管道是临时的，它们在创建进程时建立，并在进程结束时消失，没有在系统中的持久标识。 2. **半双工通信**：虽然标题提到的是双向通信，但标准的匿名管道通常是半双工的，即数据流只能在一个方向上流动。不过，通过创建两个管道，一个用于读，一个用于写，可以实现双向通信的效果。 3. **亲缘关系**：通常只允许有亲缘关系的进程（如父子进程或兄弟进程）之间使用匿名管道，这是因为权限和安全性的限制。 4. **缓冲区限制**：管道有一个内部缓冲区，如果写入数据的速度超过读取速度，可能会导致数据丢失。 5. **同步机制**：由于数据流的单向性，通常需要使用同步机制（如互斥量或信号量）来确保数据的一致性和完整性。在"匿名管道双向通讯"的应用场景中，例如替代Windows CMD接管ADB（Android Debug Bridge）的输入输出，我们可以想象这样的流程： 1. 创建两个匿名管道，一个用于从CMD到ADB的数据传输，另一个用于从ADB到CMD的反馈。 2. 子进程模拟CMD的行为，将用户在命令行输入的内容通过一个管道传递给ADB。 3. ADB接收到数据后，执行相应的操作并返回结果。 4. 结果通过另一个管道返回给子进程，再由子进程显示在CMD界面上，形成一种交互式的假象。在`Test.sln`和`Test`这两个文件中，可能包含了一个实现这种功能的C#项目解决方案。`Test.sln`是Visual Studio的解决方案文件，包含了项目设置和依赖信息。而`Test`可能是实际的代码文件，可能用C#语言编写，实现了匿名管道的创建、数据读写以及与ADB的交互逻辑。为了实现这个功能，开发者需要对Windows API有深入的理解，包括`CreatePipe`、`SetHandleInheritable`、`DuplicateHandle`等函数的使用，以及如何正确处理进程间的同步问题。同时，还需要熟悉ADB的命令行接口，以便正确地解析和构造命令。这种技术在自动化测试、设备调试和脚本编写等场景中具有很高的实用价值。

资源推荐

资源详情

资源评论

import tkinter as tk import ttkbootstrap as tb import psutil import random import threading import time from ttkbootstrap.constants import * from PIL import Image, ImageTk, ImageDraw class EnhancedCircularProgressBar: def init(self, parent, size=200, thickness=20, bg_color="#1a1a1a", text_color="#ffffff", font_size=16, title="", unit="", max_value=100, glow_effect=True): self.parent = parent self.size = size self.thickness = thickness self.bg_color = bg_color self.text_color = text_color self.font_size = font_size self.title = title self.unit = unit self.max_value = max_value self.glow_effect = glow_effect self.current_value = 0 self.target_value = 0 self.animation_speed = 2.0 # 动画速度因子 # 创建画布 self.canvas = tk.Canvas( parent, width=size, height=size, bg="black", highlightthickness=0, bd=0 ) # 计算圆心和半径 self.center_x = size / 2 self.center_y = size / 2 self.radius = (size - thickness) / 2 - 5 # 绘制背景圆环 self.draw_background() # 绘制文本 self.text_id = self.canvas.create_text( self.center_x, self.center_y, text="0%", fill=text_color, font=("Arial", font_size, "bold") ) # 绘制标题 self.title_id = self.canvas.create_text( self.center_x, self.center_y + 40, text=title, fill=text_color, font=("Arial", int(font_size0.8)) ) # 初始绘制 self.set_value(0) def draw_background(self): """绘制背景圆环""" # 计算圆环的外接矩形坐标 x0 = self.center_x - self.radius y0 = self.center_y - self.radius x1 = self.center_x + self.radius y1 = self.center_y + self.radius # 绘制背景圆环 self.bg_arc = self.canvas.create_arc( x0, y0, x1, y1, start=0, # 起始角度 extent=360, # 360度完整圆环 width=self.thickness, outline=self.bg_color, style=tk.ARC ) def interpolate_color(self, progress): """根据进度值插值计算颜色（绿-黄-橙-红过渡）""" # 定义颜色过渡点 color_points = [ (0.00, (0, 255, 0)), # 绿色 (0.25, (170, 255, 0)), # 黄绿色 (0.50, (255, 255, 0)), # 黄色 (0.75, (255, 170, 0)), # 橙色 (1.00, (255, 0, 0)) # 红色 ] # 找到当前进度所在的区间 for i in range(1, len(color_points)): if progress <= color_points[i][0]: # 计算区间内的比例 t = (progress - color_points[i-1][0]) / (color_points[i][0] - color_points[i-1][0]) # 线性插值计算RGB值（不使用numpy） r1, g1, b1 = color_points[i-1][1] r2, g2, b2 = color_points[i][1] r = int(r1 + t (r2 - r1)) g = int(g1 + t * (g2 - g1)) b = int(b1 + t * (b2 - b1)) return f"#{r:02x}{g:02x}{b:02x}" # 默认返回红色 return "#ff0000" def set_value(self, value, animate=True): """设置当前值并更新进度条""" # 确保值在合理范围内 value = max(0, min(value, self.max_value)) if animate: self.target_value = value # 启动动画（如果尚未运行） if abs(self.current_value - self.target_value) > 0.1: self.animate_step() else: self.target_value = value self.current_value = value self.update_display() def update_display(self): """更新进度条显示""" # 计算进度比例 progress = self.current_value / self.max_value # 计算角度（从-90度开始，即12点钟方向） angle = progress * 360 # 清除旧的前景圆环 if hasattr(self, 'fg_arc'): self.canvas.delete(self.fg_arc) # 计算圆环的外接矩形坐标 x0 = self.center_x - self.radius y0 = self.center_y - self.radius x1 = self.center_x + self.radius y1 = self.center_y + self.radius # 获取插值颜色 color = self.interpolate_color(progress) # 绘制前景圆环（进度指示） self.fg_arc = self.canvas.create_arc( x0, y0, x1, y1, start=-90, # 从12点钟方向开始 extent=angle, # 根据值计算角度 width=self.thickness, outline=color, style=tk.ARC ) # 提升前景圆环的显示层级 self.canvas.tag_raise(self.fg_arc) # 更新文本显示 display_text = f"{self.current_value:.1f}" if self.unit == "%": display_text = f"{progress100:.1f}%" self.canvas.itemconfig(self.text_id, text=display_text) def animate_step(self): """执行一步动画（不使用numpy）""" if abs(self.current_value - self.target_value) < 0.1: self.current_value = self.target_value else: # 使用缓动函数实现平滑动画 diff = self.target_value - self.current_value self.current_value += diff 0.2 * self.animation_speed self.update_display() # 如果未达到目标值，继续动画 if abs(self.current_value - self.target_value) > 0.1: self.parent.after(16, self.animate_step) # 约60fps class SystemMonitorApp: def init(self): # 创建主窗口 self.root = tb.Window(themename="darkly", title="系统资源监控", size=(1300, 800)) self.root.minsize(1000, 700) # 设置样式 style = tb.Style() style.configure("TFrame", background="#121212") style.configure("Title.TLabel", background="#121212", foreground="#e0e0e0", font=("Arial", 16, "bold")) style.configure("Section.TLabelframe", font=("Arial", 10, "bold")) # 创建主框架 - 使用网格布局 main_frame = tb.Frame(self.root, padding=10) main_frame.pack(fill=tk.BOTH, expand=True) # ========== 顶部控制栏 ========== control_bar = tb.Frame(main_frame, padding=(10, 5)) control_bar.grid(row=0, column=0, columnspan=3, sticky="ew", pady=(0, 10)) # 设备选择下拉框 tb.Label(control_bar, text="监控设备:", bootstyle=PRIMARY).pack(side=tk.LEFT, padx=(0, 5)) self.device_var = tk.StringVar() device_options = ["设备A", "设备B", "设备C", "设备D", "设备E"] device_combo = tb.Combobox( control_bar, textvariable=self.device_var, values=device_options, width=15, state="readonly", bootstyle=PRIMARY ) device_combo.pack(side=tk.LEFT, padx=5) device_combo.current(0) # 设置默认选项 # 采样频率滑块 tb.Label(control_bar, text="采样频率:", bootstyle=PRIMARY).pack(side=tk.LEFT, padx=(20, 5)) self.sampling_rate = tk.IntVar(value=1) tb.Scale( control_bar, from_=0.5, to=5, length=120, orient=tk.HORIZONTAL, variable=self.sampling_rate, bootstyle=PRIMARY ).pack(side=tk.LEFT, padx=5) # 动画速度滑块 (修复resolution问题) tb.Label(control_bar, text="动画速度:", bootstyle=PRIMARY).pack(side=tk.LEFT, padx=(20, 5)) self.animation_speed = tk.DoubleVar(value=1.0) tb.Scale( control_bar, from_=0.5, to=3.0, length=120, orient=tk.HORIZONTAL, variable=self.animation_speed, bootstyle=PRIMARY ).pack(side=tk.LEFT, padx=5) # 控制按钮 tb.Button( control_bar, text="启动", bootstyle=SUCCESS, width=8, command=self.start_monitoring ).pack(side=tk.LEFT, padx=5) tb.Button( control_bar, text="暂停", bootstyle=DANGER, width=8, command=self.stop_monitoring ).pack(side=tk.LEFT, padx=5) tb.Button( control_bar, text="重置", bootstyle=WARNING, width=8, command=self.reset_all ).pack(side=tk.LEFT, padx=5) # 状态指示器 self.status_indicator = tb.Label( control_bar, text="● 已连接", bootstyle=(SUCCESS, INVERSE), font=("Arial", 10), padding=(10, 0) ) self.status_indicator.pack(side=tk.RIGHT, padx=(10, 0)) # ========== 左侧设置面板 ========== settings_frame = tb.Labelframe( main_frame, text="监控设置", bootstyle="info", padding=10, style="Section.TLabelframe" ) settings_frame.grid(row=1, column=0, sticky="nsew", padx=(0, 10), pady=(0, 10)) # 温度报警设置 temp_frame = tb.Frame(settings_frame) temp_frame.pack(fill=tk.X, pady=5) tb.Label(temp_frame, text="温度报警阈值(°C):").pack(side=tk.LEFT) self.temp_threshold = tk.IntVar(value=80) tb.Entry( temp_frame, textvariable=self.temp_threshold, width=8, bootstyle=PRIMARY ).pack(side=tk.RIGHT, padx=(5, 0)) # 内存报警设置 mem_frame = tb.Frame(settings_frame) mem_frame.pack(fill=tk.X, pady=5) tb.Label(mem_frame, text="内存报警阈值(%):").pack(side=tk.LEFT) self.mem_threshold = tk.IntVar(value=90) tb.Entry( mem_frame, textvariable=self.mem_threshold, width=8, bootstyle=PRIMARY ).pack(side=tk.RIGHT, padx=(5, 0)) # 网络报警设置 net_frame = tb.Frame(settings_frame) net_frame.pack(fill=tk.X, pady=5) tb.Label(net_frame, text="网络报警阈值(Mbps):").pack(side=tk.LEFT) self.net_threshold = tk.IntVar(value=800) tb.Entry( net_frame, textvariable=self.net_threshold, width=8, bootstyle=PRIMARY ).pack(side=tk.RIGHT, padx=(5, 0)) # 分隔线 tb.Separator(settings_frame, bootstyle=SECONDARY).pack(fill=tk.X, pady=10) # 主题选择 tb.Label(settings_frame, text="界面主题:", bootstyle=PRIMARY).pack(anchor=tk.W, pady=(0, 5)) self.theme_var = tk.StringVar(value="darkly") themes = ["darkly", "solar", "superhero", "cyborg", "vapor"] tb.Combobox( settings_frame, textvariable=self.theme_var, values=themes, state="readonly", bootstyle=PRIMARY ).pack(fill=tk.X, pady=(0, 10)) # 历史数据按钮 tb.Button( settings_frame, text="查看历史数据", bootstyle=(OUTLINE, INFO), command=self.show_history ).pack(fill=tk.X, pady=5) # ========== 中间监控面板 ========== monitor_frame = tb.Frame(main_frame) monitor_frame.grid(row=1, column=1, sticky="nsew", pady=(0, 10)) # 标题 tb.Label( monitor_frame, text="系统资源实时监控", style="Title.TLabel" ).pack(pady=(0, 15)) # 创建进度条容器框架 progress_container = tb.Frame(monitor_frame) progress_container.pack(fill=tk.BOTH, expand=True, pady=5) # 创建监控指标配置 monitor_config = [ {"title": "CPU温度", "unit": "°C", "max_value": 100, "thickness": 18}, {"title": "功耗", "unit": "W", "max_value": 200, "thickness": 18}, {"title": "内存使用", "unit": "%", "max_value": 100, "thickness": 18}, {"title": "网络带宽", "unit": "Mbps", "max_value": 1000, "thickness": 18}, {"title": "MLU利用率", "unit": "%", "max_value": 100, "thickness": 18}, {"title": "CPU利用率", "unit": "%", "max_value": 100, "thickness": 18} ] # 使用网格布局排列进度条 self.progress_bars = [] for i, config in enumerate(monitor_config): frame = tb.Frame(progress_container) frame.grid(row=i//3, column=i%3, padx=15, pady=15, sticky="nsew") # 创建增强型进度条 progress_bar = EnhancedCircularProgressBar( frame, size=220, thickness=config["thickness"], title=config["title"], unit=config["unit"], max_value=config["max_value"], glow_effect=True ) self.progress_bars.append(progress_bar) progress_bar.canvas.pack(fill=tk.BOTH, expand=True) # 设置网格列权重 for i in range(3): progress_container.columnconfigure(i, weight=1) for i in range(2): progress_container.rowconfigure(i, weight=1) # ========== 右侧信息面板 ========== info_frame = tb.Labelframe( main_frame, text="系统信息", bootstyle="info", padding=10, style="Section.TLabelframe" ) info_frame.grid(row=1, column=2, sticky="nsew", padx=(10, 0), pady=(0, 10)) # 系统信息标签 info_labels = [ ("设备型号:", "MLU"), ("操作系统:", "Ubuntu 20.04 LTS"), ("处理器:", "16核 ARMv8"), ("内存总量:", "32 GB"), ("MLU数量:", "4"), ("网络接口:", "eth0 (10Gbps)") ] for label, value in info_labels: frame = tb.Frame(info_frame) frame.pack(fill=tk.X, pady=3) tb.Label(frame, text=label, width=10, anchor=tk.W).pack(side=tk.LEFT) tb.Label(frame, text=value, bootstyle=INFO).pack(side=tk.LEFT) # 分隔线 tb.Separator(info_frame, bootstyle=SECONDARY).pack(fill=tk.X, pady=10) # 实时状态 tb.Label(info_frame, text="实时状态", bootstyle=PRIMARY).pack(anchor=tk.W, pady=(0, 5)) self.realtime_labels = {} status_items = [ ("CPU温度", "cpu_temp", "°C"), ("功耗", "power", "W"), ("内存使用", "memory", "%"), ("网络带宽", "bandwidth", "Mbps"), ("MLU利用率", "mlu_usage", "%"), ("CPU利用率", "cpu_usage", "%") ] for name, key, unit in status_items: frame = tb.Frame(info_frame) frame.pack(fill=tk.X, pady=2) tb.Label(frame, text=name, width=12, anchor=tk.W).pack(side=tk.LEFT) value_label = tb.Label(frame, text="0.0", width=8, anchor=tk.W) value_label.pack(side=tk.LEFT) tb.Label(frame, text=unit).pack(side=tk.LEFT) self.realtime_labels[key] = value_label # ========== 状态栏 ========== self.status = tb.Label( self.root, text="系统准备就绪 | 当前设备: 设备A", bootstyle=(SECONDARY, INVERSE), anchor=tk.CENTER ) self.status.pack(side=tk.BOTTOM, fill=tk.X) # 配置网格权重 main_frame.columnconfigure(0, weight=1) # 左侧设置面板 main_frame.columnconfigure(1, weight=3) # 中间监控面板主要区域 main_frame.columnconfigure(2, weight=1) # 右侧信息面板 main_frame.rowconfigure(1, weight=1) # 监控控制变量 self.monitoring_active = False self.monitoring_thread = None # 启动初始监控 self.start_monitoring() # 绑定主题切换事件 self.theme_var.trace_add("write", lambda *args: self.change_theme()) self.root.mainloop() def start_monitoring(self): """启动资源监控""" if self.monitoring_active: return self.status.config(text="启动系统资源监控...") self.monitoring_active = True # 使用线程运行监控，避免阻塞UI self.monitoring_thread = threading.Thread(target=self.monitor_resources, daemon=True) self.monitoring_thread.start() def stop_monitoring(self): """停止资源监控""" self.monitoring_active = False self.status.config(text="监控已暂停") def reset_all(self): """重置所有监控指标""" for bar in self.progress_bars: bar.set_value(0) self.status.config(text="所有监控指标已重置") def change_theme(self): """更改应用主题""" theme = self.theme_var.get() # 正确切换主题方法 self.root.style.theme_use(theme) self.status.config(text=f"主题已切换为: {theme.capitalize()}") def show_history(self): """显示历史数据""" self.status.config(text="历史数据查看功能开发中...") def monitor_resources(self): """监控系统资源（线程安全实现）""" while self.monitoring_active: # 获取系统资源数据 data = { 'cpu_temp': self.get_cpu_temperature(), 'power': self.get_power_usage(), 'mem_usage': self.get_memory_usage(), 'network': self.get_network_usage(), 'mlu_usage': self.get_mlu_usage(), 'cpu_usage': psutil.cpu_percent() } # 使用after安全更新UI self.root.after(0, self.update_ui, data) time.sleep(max(0.5, self.sampling_rate.get())) def update_ui(self, data): """安全更新UI组件（在主线程执行）""" # 更新进度条的动画速度 speed = self.animation_speed.get() for bar in self.progress_bars: bar.animation_speed = speed # 更新进度条 self.progress_bars[0].set_value(data['cpu_temp']) self.progress_bars[1].set_value(data['power']) self.progress_bars[2].set_value(data['mem_usage']) self.progress_bars[3].set_value(data['network']) self.progress_bars[4].set_value(data['mlu_usage']) self.progress_bars[5].set_value(data['cpu_usage']) # 更新实时状态标签 self.realtime_labels["cpu_temp"].config(text=f"{data['cpu_temp']:.1f}") self.realtime_labels["power"].config(text=f"{data['power']:.1f}") self.realtime_labels["memory"].config(text=f"{data['mem_usage']:.1f}") self.realtime_labels["bandwidth"].config(text=f"{data['network']:.1f}") self.realtime_labels["mlu_usage"].config(text=f"{data['mlu_usage']:.1f}") self.realtime_labels["cpu_usage"].config(text=f"{data['cpu_usage']:.1f}") # 更新状态栏 status_text = ( f"当前设备: {self.device_var.get()} | " f"CPU: {data['cpu_usage']:.1f}% | " f"温度: {data['cpu_temp']:.1f}°C | " f"内存: {data['mem_usage']:.1f}% | " f"MLU: {data['mlu_usage']:.1f}%" ) self.status.config(text=status_text) # 资源获取方法 def get_cpu_temperature(self): """获取CPU温度（模拟数据）""" return random.uniform(40, 90) def get_power_usage(self): """获取功耗（模拟数据）""" return random.uniform(50, 180) def get_memory_usage(self): """获取内存使用率""" return psutil.virtual_memory().percent def get_network_usage(self): """获取网络带宽（模拟数据）""" return random.uniform(100, 900) def get_mlu_usage(self): """获取MLU利用率（模拟数据）""" return random.uniform(10, 95) if name == "main": SystemMonitorApp() 该程序的圆环进度条起始点是6点钟方向，怎么修改为12点钟方向

<think>我们正在使用自定义的EnhancedCircularProgressBar类，它继承自......

收起资源包目录

匿名管道-双向通讯-2012-09-07.rar （6个子文件）

Test.sln 878B

Test

TestPipes.cpp 2KB

stdafx.h 233B

stdafx.cpp 209B

Test.vcproj 4KB

targetver.h 498B

#include "stdafx.h" #include <windows.h> #include <stdio.h> #include <stdlib.h> #include <string> #include <vector> #include <iostream> using namespace std; char cmdbuffer[1024]; HANDLE hReadPipe; HANDLE hReadPipe2; HANDLE hWritePipe; HANDLE hWritePipe2; DWORD __stdcall ThreadFun(void *pVoid) { SECURITY_ATTRIBUTES sat; STARTUPINFO startupinfo; PROCESS_INFORMATION pinfo; BYTE buffer[1024]; DWORD byteRead; sat.nLength=sizeof(SECURITY_ATTRIBUTES); sat.bInheritHandle=true; sat.lpSecurityDescriptor=NULL; if(!CreatePipe(&hReadPipe,&hWritePipe,&sat,NULL)) { //MessageBox(NULL, "Create Pipe Error! ", "error! ",MB_OK); return 0; } if(!CreatePipe(&hReadPipe2,&hWritePipe2,&sat,NULL)) { //MessageBox(NULL, "Create Pipe2 Error! ", "error! ",MB_OK); return 0; } startupinfo.cb=sizeof(STARTUPINFO); GetStartupInfo(&startupinfo); startupinfo.hStdError=hWritePipe; startupinfo.hStdOutput=hWritePipe; startupinfo.hStdInput=hReadPipe2; startupinfo.dwFlags=STARTF_USESHOWWINDOW | STARTF_USESTDHANDLES; startupinfo.wShowWindow=SW_HIDE; if(!CreateProcess(NULL, "C:\\Windows\\adb.exe shell",NULL, NULL, TRUE, NULL, NULL, NULL,&startupinfo,&pinfo)) { DWORD dwError = GetLastError(); //MessageBox(NULL, "create process error! ", "Error! ",MB_OK); return 0; } CloseHandle(hWritePipe); CloseHandle(hReadPipe2); CloseHandle(pinfo.hThread); CloseHandle(pinfo.hProcess); while(true) { RtlZeroMemory(buffer,1024); if(ReadFile(hReadPipe,buffer,1023,&byteRead,NULL)==NULL) break; cout<<buffer<<endl; } CloseHandle(hReadPipe); CloseHandle(hWritePipe2); return 0; } int main() { //创建线程： HANDLE hThread = NULL; DWORD dwThread = 0; hThread = CreateThread( NULL, 0, ThreadFun, 0, 0, &dwThread);//这个参数你根据你的需要决定要不要 if (hThread != NULL) { //MessageBox( "CreateThread failed. ", "main ", MB_OK ); } else { CloseHandle( hThread ); } char command[128] = {0}; while ( true) { memset( command, 0, 128); gets_s(command); strcat_s(command, " \r\n");//一定要有回车换行 //往adb程序里写命令： DWORD byteRead = 0; WriteFile(hWritePipe2,command,strlen(command),&byteRead,NULL); } return 0; }

评论收藏

内容反馈