ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Kivy 实用代码片段指南:自定义控件、线程安全 UI 与异步工作流实战

Kivy 实用代码片段指南:自定义控件、线程安全 UI 与异步工作流实战 Kivy 实用代码片段指南自定义控件、线程安全 UI 与异步工作流实战【免费下载链接】kivyOpen source UI framework written in Python, running on Windows, Linux, macOS, Android and iOS项目地址: https://gitcode.com/gh_mirrors/ki/kivy本篇技术指南以 Kivy 官方入门文档中的 Snippets 片段集为主体逐条剖析六个可直接复用的实战片段自定义控件封装、mainthread线程安全更新 UI、基于 Trio 的异步事件循环、通过EventDispatcher自定义事件与插件、用 KV 语言组织跨屏幕工作流以及在 Windows 上也能运行的 Flask 内嵌方案。读完本文你将掌握这些片段背后的 Kivy 核心机制事件系统、属性绑定、时钟调度并能直接复制到自己的项目中运行。Snippets 片段集是什么在 Kivy 文档的入门Getting Started章节中Snippets是一组由社区贡献的、短小精悍的即插即用代码片段覆盖了日常开发中最容易踩坑的场景如何组织自定义控件、如何安全地在子线程中更新界面、如何接入trio异步生态、如何自定义事件让 KV 代码更简洁以及如何把 Web 服务嵌入桌面应用。其组织方式比较特殊入口文档 doc/sources/gettingstarted/snippets.rst 本身只包含用于渲染轮播carousel的 HTML 样式与对_code_snippets_slides.html的引用真正的片段正文存放在 doc/sources/snippets_slides/ 目录下由 doc/sources/snippets_slides.js 负责按需异步加载loadSlideContent通过fetch拉取每个data-src对应的 HTML 并注入.rst-content容器实现轮播切换时的懒加载与卸载。因此阅读这些片段的完整内容应直接查看snippets_slides目录下的各.rst文件。片段一自定义控件 CustomWidget第一个片段 custom_widgets.rst 演示了如何编写自己的控件把纯 Python 逻辑从 KV 脚本中剥离出来让项目保持整洁、易于维护。片段给出的实践原则是不要把大段 Python 塞进 KV 脚本自定义控件通常是更好控制代码、区分短任务与长任务的最佳选择。# Filepath: libs/customwidget.py from kivy.lang import Builder from kivy.properties import ColorProperty from kivy.uix.behaviors import ButtonBehavior from kivy.uix.widget import Widget __all__ (CustomWidget,) Builder.load_string( CustomWidget: on_release: print(Short code goes here) # Remove this line if you are using the call from the Python script itself canvas.before: Color: rgba: self.color Rectangle: size: self.size pos: self.pos ) class CustomWidget(ButtonBehavior, Widget): color ColorProperty((.5, .5, .5, .8)) def on_release(self): print(Long code goes here) # remove this definition if you are instead using on_release from your KV code这个片段的关键点在于多重继承组合能力CustomWidget继承自ButtonBehavior与Widget。ButtonBehavior实现在 kivy/uix/behaviors/button.py是一个不负责绘制、只提供按下/释放交互行为的 Mixin 类它与Widget组合即可得到一个带按钮语义的自绘控件而无需引入完整的Button。ColorProperty声明式属性color ColorProperty((.5, .5, .5, .8))是 Kivy 属性系统的应用——属性的每次赋值都会自动触发绑定刷新KV 中的Color: rgba: self.color会随属性变化自动重绘无需手写 setter。KV 内联与 Python 回调的两种用法KV 模板里用on_release: print(...)处理短逻辑Python 类里再定义同名on_release方法处理长逻辑二者保留其一即可注释里已明确说明如何取舍。使用方式有两种# 在 Python 脚本中导入 from libs.customwidget import CustomWidget# 在 KV 语言中导入 #:import CustomWidget libs.customwidget.CustomWidget导入之后就可以结合 Kivy 的Factory机制kivy/factory.py在任意位置实例化或继承该控件。这与 Kivy 文档对 Factory 的定位一致它让你在 KV 中按名称注册类从而解耦类定义与使用位置。片段二子线程中安全更新 UImainthreadthreading_example.rst 展示了 Kivy 中最常见的并发陷阱UI 只能在主线程中修改子线程拿到数据后必须切回主线程再更新界面。片段的核心武器是kivy.clock.mainthread装饰器。import threading import time from kivy.clock import mainthread from kivy.lang import Builder from kivy.properties import NumericProperty, StringProperty from kivy.uix.boxlayout import BoxLayout __all__ (MyBL,) class MyBL(BoxLayout): counter NumericProperty() data_label StringProperty(Nothing yet!) def __init__(self, **kwargs): super().__init__(**kwargs) threading.Thread(targetself.get_data).start() def get_data(self): while App.get_running_app(): # get data here # sock.recv(1024) or how you do it time.sleep(1) # if you change the UI you need to do it on main thread self.set_data_label(self.counter) self.counter 1 mainthread def set_data_label(self, counter: int): self.data_label str(counter) Builder.load_string( MyBL: Label: font_size: 30sp text: Some Data Label: font_size: 30sp text: root.data_label ) if __name__ __main__: from kivy.app import App class MyApp(App): def build(self): return MyBL() MyApp().run()从源码层面看mainthread定义在 kivy/clock.pydef mainthread(func): Decorator that will schedule the call of the function for the next available frame in the mainthread. It can be useful when you use :class:~kivy.network.urlrequest.UrlRequest or when you do Thread ...也就是说被mainthread装饰的方法调用时并不会立即执行而是被调度到主线程的下一个可用帧中运行。这保证了data_label这类属性对界面的更新发生在主线程避免跨线程直接操作 UI 引发的竞态与崩溃。片段中的几个工程细节值得注意threading.Thread(targetself.get_data).start()在__init__里启动后台线程模拟sock.recv(1024)这类阻塞式数据采集循环条件while App.get_running_app()让线程在应用退出时自然结束避免僵尸线程counter与data_label分别使用NumericProperty和StringProperty定义于 kivy/properties.pyxKV 中text: root.data_label会自动随属性变化刷新需要说明的是片段中的App是在if __name__ __main__分支内导入的实际使用时应把from kivy.app import App移到模块顶部。片段三用 Trio 运行异步 Kivy 应用async_with_trio.rst 演示了如何把 Kivy 的事件循环交给trio结构化并发框架托管。Kivy 从 2.0 开始原生支持异步事件循环async_run并允许通过async_lib参数选择asyncio或trio作为底层实现。import trio from kivy.app import App from kivy.uix.label import Label class AsyncApp(App): def build(self): return Label(textHello Kivy) async def async_run(self): async with trio.open_nursery() as nursery: self.nursery nursery await super().async_run(async_libtrio) nursery.cancel_scope.cancel() if __name__ __main__: trio.run(AsyncApp().async_run)运行方式是trio.run(AsyncApp().async_run)以trio.run作为程序入口在async_run内打开一个 nurserytrio 的任务作用域再调用super().async_run(async_libtrio)让 Kivy 在 trio 的循环中运行。当 Kivy 退出时nursery.cancel_scope.cancel()负责取消其余任务保证结构化并发的清理语义。仓库自带的 examples/async/trio_basic.py 与 examples/async/trio_advanced.py 展示了更完整的 trio 用法在子任务中定期通过Clock更新界面、与 Kivy 事件循环协作可作为深入学习的参照。Kivy 官方也将其作为一等公民支持App.async_run的async_lib参数即用于在asyncio与trio之间切换。片段四基于 EventDispatcher 的自定义插件/事件custom_plugins.rst 解决了一个常见的 Python 陷阱普通类无法与 Kivy 的控件/布局多重继承。Kivy 的控件体系建立在EventDispatcher实现在 kivy/_event.pyx 编译产物之上接口见 kivy/event.py之上因此可复用逻辑也应从它派生。# Filepath: libs/customevent.py from kivy.event import EventDispatcher from kivy.properties import ColorProperty __all__ (CustomEvent,) class CustomEvent(EventDispatcher): color ColorProperty((.5, .5, .5, .8)) def is_going_to_happen(self): print(Kivy rules!)使用方式# 在 Python 脚本中导入 from libs.customevent import CustomEvent# 与布局/控件多重继承 class OtherLayout(BoxLayout, CustomEvent): ...为什么必须用EventDispatcher片段原文的解释是有时你无法用普通类class MyClass:与布局、控件做多重继承。其根源在于Kivy 的属性Property系统在类创建时通过元类EventDispatcher的 metaclass为每个属性生成_event_properties等内部注册信息Widget、Layout等基类也依赖这套机制完成事件绑定与属性通知。只有同样基于EventDispatcher的类才能与其他 Kivy 控件在 MRO 上正确协作、让ColorProperty等属性在多重继承下正常工作。这也是自定义插件的通用模式把可复用的状态与行为收敛进一个EventDispatcher子类再通过多重继承混入任意布局或控件。片段五用 KV 语言编排应用级工作流custom_workflows_with_kvlang.rst 展示了一种应用级工作流组织方式用ScreenManager 自定义事件把屏幕跳转逻辑全部写在 KV 里Python 只负责声明事件。这是片段集中最能体现 Kivy 声明式风格的一个。ScreenManager: id: screenmanager IdleScreen: name: idle on_press: screenmanager.current date_selection DateSelectionScreen: name: date_selection catalog: app.catalog on_submit: app.departure_datetime self.departure_datetime app.cart.filter_by_deadline(self.departure_datetime) screenmanager.current browse BrowseScreen: name: browse catalog: app.catalog cart: app.cart departure_datetime: app.departure_datetime on_date_edit: screenmanager.current date_selection on_back: screenmanager.current date_selection on_order_submit: app.create_order() screenmanager.current order OrderScreen: id: order_screen name: order cart: app.cart order: app.order on_back: screenmanager.current browse请注意on_submit、on_date_edit、on_back、on_order_submit等都是自定义事件。声明它们非常简单在类的__events__属性中列出事件名并实现对应的空处理器。ScreenManager与Screen实现在 kivy/uix/screenmanager.py。from kivy.uix.screenmanager import Screen class OrderScreen(Screen): __events__ (on_reset, on_back) def on_reset(self): pass def on_back(self): pass__events__是 Kivy 事件系统的核心约定事件分发器会依据__events__元组注册命名事件并自动调用同名的on_event处理器若存在。仓库中到处可见这一约定例如 kivy/animation.py 的__events__ (on_start, on_progress, on_complete)、kivy/app.py 的__events__ (on_start, on_ready, on_stop, on_pause, on_resume, ...)以及 kivy/base.py 的__events__ (on_start, on_pause, on_stop)。当多个屏幕需要复用同一批事件比如订单页和浏览页都有返回时把事件定义在基类、由各屏幕继承再在 KV 中通过on_event:绑定跳转逻辑就可以把流程控制整体上浮到 KV 层让 Python 保持为纯粹的领域逻辑。片段六把 Flask 嵌入 KivyWindows 也可用flask_with_kivy/snippet.rst 是片段集中最完整的实战示例把 Flask 后端以独立线程内嵌进 Kivy 应用二者通过共享的ListProperty交互并用 Server-Sent EventsSSE做实时推送附带一个测试客户端。作者特别注明works on Windows too——这通常意味着在 Windows 上无法依赖某些 Unix 专用 API方案应保持纯跨平台实现。服务端代码保存为独立文件如server.pyimport json import threading import time from collections.abc import Generator from typing import Any import trio import waitress # type: ignore[import-untyped] from flask import Flask, request from flask.typing import ResponseReturnValue from kivy.app import App # type: ignore[import-untyped] from kivy.lang.builder import Builder # type: ignore[import-untyped] from kivy.properties import ListProperty # type: ignore[import-untyped] api Flask(__name__) api.get(/messages) def message_stream() - ResponseReturnValue: app App.get_running_app() def stream_messages() - Generator[str, None, None]: next_index 0 while True: for index, message in enumerate( app.messages[next_index:], startnext_index, ): data {message: message} yield fdata: {json.dumps(data)}\n\n next_index index 1 time.sleep(1) yield :heartbeat\n return stream_messages() api.post(/messages) def post_message() - ResponseReturnValue: app App.get_running_app() data request.get_json() message data[message] app.messages.append(message) return , 201 def run_backend() - None: api_thread threading.Thread( targetwaitress.serve, kwargs{ app: api, host: 0.0.0.0, port: 8080, threads: 256, }, daemonTrue, ) api_thread.start() KV r Label: text: Messages:\n{}.format(\n.join(app.messages)) text_size: self.size valign: top class EmbeddedFlask(App): # type: ignore messages ListProperty() async def async_run(self) - None: async with trio.open_nursery() as nursery: self.nursery nursery run_backend() await super().async_run(async_libtrio) nursery.cancel_scope.cancel() def build(self) - Any: return Builder.load_string(KV) app EmbeddedFlask() if __name__ __main__: trio.run(app.async_run)客户端测试代码保存为client.py Usage: - python client.py - Listens to messages stream and displays the results - python client.py my message - Sends a new message to the backend import json import sys from collections.abc import Sequence import requests # type: ignore[import-untyped] API_URI http://localhost:8080/messages def main(args: Sequence[str]) - None: if len(args) 1: message .join(sys.argv[1:]) requests.post(API_URI, json{message: message}) return for line in requests.get(API_URI, streamTrue).iter_lines(): data line.decode(utf8).strip() if data.startswith(:): continue command, _, payload data.partition(:) if command ! data: continue event json.loads(payload) print(event) if __name__ __main__: main(sys.argv)这个示例的关键设计数据共享而非跨线程调用Flask 路由通过App.get_running_app()拿到运行中的 Kivy 应用实例直接追加到app.messages一个ListProperty。由于messages是 Kivy 属性KV 中的\n.join(app.messages)会在列表变化时自动刷新界面——这正是片段二所强调的属性系统让数据流自动驱动 UI的进阶应用。线程安全的前提示例中后台线程只追加messages列表数据UI 文本通过属性绑定重算Kivy 属性的线程行为虽非完全无锁但ListProperty的赋值/追加配合主线程帧刷新在演示场景下是可行且常见的做法。若需要更严格的线程安全应配合片段二的mainthread模式。waitress 作为 WSGI 服务器用waitress.serve在守护线程中启动threads256指明并发线程数host0.0.0.0、port8080是监听地址。SSE 实时推送GET /messages返回一个生成器按 SSE 协议逐条data: {...}\n\n推送新消息并以:heartbeat心跳维持连接客户端用requests.get(streamTrue).iter_lines()逐行消费跳过以:开头的心跳行解析data:载荷。trio 托管整体生命周期run_backend()在async_run内启动Kivy 退出时 nursery 的cancel_scope.cancel()一并结束应用进程。总结与延伸阅读六个片段覆盖了 Kivy 开发中反复出现的四类主题控件封装自定义 Widget ButtonBehavior、线程与异步mainthread、trio、async_run、事件系统EventDispatcher、__events__、自定义事件、以及进程内服务嵌入Flask SSE。它们的共同底层支撑是 Kivy 的属性系统kivy/properties.pyx与事件分发系统kivy/_event.pyx、kivy/event.py——理解了这两者所有片段都能举一反三。若想继续深入属性与事件的基础原理见 doc/sources/gettingstarted/properties.rst 与 doc/sources/gettingstarted/events.rstKV 语言规则见 doc/sources/guide/lang.rst官方异步示例位于 examples/async/asyncio_basic.py、trio_basic.py、trio_advanced.py时钟调度与mainthread的完整语义见 kivy/clock.py控件行为 Mixin 家族见 kivy/uix/behaviors/。这些片段由 Kivy 社区成员贡献并收录进官方文档是经过真实项目验证的惯用法直接对照片段源码与上述核心模块阅读是理解 Kivy 设计哲学最有效率的方式。【免费下载链接】kivyOpen source UI framework written in Python, running on Windows, Linux, macOS, Android and iOS项目地址: https://gitcode.com/gh_mirrors/ki/kivy创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表