背景

公司是做AR实景产品的，近几年无人机特别的火，一来公司比较关注低空经济这个新型领域，二来很多政企、事业单位都采购了无人机用于日常工作。那么就衍生了将无人机接入到AR实景项目中的需求。

本工具的作用就是在AR实景项目中模拟无人机飞行时候产生的飞控数据。

工具界面展示

功能

• 用户填写kafka连接地址，
• 用户参考消息示例模板编写消息内容，并发送到kafka 的 topic 中。
• 所有操作的log输出到用户界面

代码讲解

运行环境

• python：3.9.7
• pip：pip3(pip 21.2.3 )
• OS：win11
• GUI：TKinter
• kafka-client：confluent_kafka

创建GUI程序

• 通过 tk.Tk() 实例化一个GUI程序。
• 通过 ttk.Style() 实例化主题并通过style.configure 设置公共样式。
• 通过 KafkaMessageSenderApp(root) 的构造函数将root实例传入业务代码中
• 最后通过root.mainloop()让程序保持前台运行

if __name__ == "__main__":
    # 设置样式
    root = tk.Tk()
    style = ttk.Style()

    # 尝试配置主题
    try:
        if "clam" in style.theme_names():
            style.theme_use("clam")
    except:
        pass

    # 自定义样式
    style.configure("TLabel", font=("SimHei", 10))
    style.configure("TButton", font=("SimHei", 10))
    style.configure("TEntry", font=("SimHei", 10))
    style.configure("TFrame", background="#f0f0f0")
    style.configure("TLabelframe", background="#f0f0f0")
    style.configure(
        "TLabelframe.Label", background="#f0f0f0", font=("SimHei", 10, "bold")
    )

    # 创建强调按钮样式
    style.configure("Accent.TButton", foreground="white", background="#4a86e8")

    app = KafkaMessageSenderApp(root)
    root.mainloop()
               

搭建前端样式

• 在KafkaMessageSenderApp 的构造函数中，通过create_widgets函数完成前端样式的编写，例如 label 的位置，button 的名字及绑定的回调处理函数
• 通过 start_log_consumer 处理日志
• 通过generate_example_message生成模板消息

        self.create_widgets()
        self.start_log_consumer()

        # 生成示例消息
        self.generate_example_message()
                               

编写功能实现代码

通过为 前端各种控件绑定回调函数来实现业务代码的编写。例如通过为 发送消息 按钮绑定回调处理消息的发送。

按钮样式和回调函数绑定：

# 按钮区域
        button_frame = ttk.Frame(main_frame)
        button_frame.pack(fill=tk.X, pady=5)

        # 创建按钮并保存引用
        self.send_button = ttk.Button(
            button_frame,
            text="发送消息",
            command=self.send_message,
            width=20,
            style="Accent.TButton",
        )
                               

发送消息：

def send_message(self):
        """发送消息到Kafka"""
        if self.producer is None:
            messagebox.showerror("错误", "请先连接到Kafka服务器")
            return

        topic = self.topic_var.get()
        if not topic:
            messagebox.showerror("错误", "请输入Topic")
            return

        message_text = self.message_text.get(1.0, tk.END).strip()
        if not message_text:
            messagebox.showerror("错误", "消息内容不能为空")
            return

        # 解析JSON
        try:
            message_data = json.loads(message_text)
        except json.JSONDecodeError as e:
            messagebox.showerror("JSON解析错误", f"无效的JSON格式: {str(e)}")
            return

        # 在新线程中发送消息，避免阻塞UI
        threading.Thread(
            target=self._send_message_thread, args=(topic, message_data), daemon=True
        ).start()