一代明君的小屋 - python
type: channel
link: https://yidaimingjvn.xyz/index.php/tag/python/
items: 使用 pipx 专门管理 Python 应用
items: 基于mac的设备客户端任务队列调度服务 ——key设计
items: 基于mac的设备客户端任务队列调度服务
items: 分布式任务调度与资源管理系统设计
items: 关于鲁班猫rk3568适配的linux系统使用新版本python的情况
items: 自定协议，串口数据处理粘包方法
items: opencv运行报错找不到libGL.so.1
items: 如何通过python采集windows正在运行的进程信息
items: 关于pyqt6设置右对齐居中的示例
items: python怎么将项目的pip依赖拉成whl

使用 pipx 专门管理 Python 应用
type: item
date: 2026-07-13T23:10:44+08:00
description: 使用 pipx 专门管理 Python 应用pipx 是专门用来安装 Python 命令行工具（如 virtualenv、black、httpie）的隔离工具，它自动为每个工具创建独立的虚拟环境，完全符合您的“不用 -m venv”的要求，而且也不会触发 PEP 668。安装 pipx（推荐通过 apt 安装，避开权限问题）：sudo apt update sudo apt install pipx pipx ensurepath # 自动将 pipx 的 bin 目录加入 PATH（如果不想用 apt，也可以用 python3.12 -m pip install --user pipx）用 pipx 安装 virtualenv：pipx install virtualenv现在您可以全局使用 virtualenv 命令了：virtualenv .venv source .venv/bin/activate

基于mac的设备客户端任务队列调度服务 ——key设计
type: item
date: 2026-06-22T14:54:00+08:00
description: 调度服务前文基于mac的设备客户端任务队列调度服务一、MAC 任务调度相关 Keysensitive:mac:queue:<mac> — List（每个 MAC 一个）每个设备的任务队列。入队时 RPUSH，消费时 RPOP（LIFO 栈行为）。过期检查、死信重投也在此操作。sensitive:mac:active:<mac> — String（SET NX EX）运行锁，防止同一 MAC 被多个 Worker 同时调度。NX 保证原子性获得锁，EX 自动过期（默认 900 秒）防止死锁，消费完成后手动 DELETE 释放。sensitive:mac:schedule — ZSet（全局唯一）调度有序集合，score = 老化等待秒数（封顶 600 秒 × 权重）。ZREVRANGE 按等待时间从高到低获取候选 MAC，实现 "等最久的先调度" 策略。sensitive:mac:meta:<mac> — Hash（每个 MAC 一个）元数据，包含以下字段：字段说明pending_count队列积压任务数oldest_created_at最旧任务的创建时间，用于计算老化分数updated_at最后更新时间unschedulable_until退避到期时间戳（defer 指数退避时设置）二、Worker 管理相关 Keysensitive:worker:<hostname>:<pid> — Hash（每个 Worker 进程一个）Worker 资源注册与心跳，包含 CPU/GPU/内存利用率、活跃任务数、最后心跳时间等。Worker 启动时 HSET 注册，每 15 秒 心跳更新，TTL 过期（默认 30 秒）后自动清理。三、全局监控与过载保护 Keysensitive:global_pending — String（计数器）全局待处理任务总数。入队 INCR，出队 DECR，用于快速判断系统整体负载。sensitive:overloaded_macs — Set过载 MAC 集合。当单个 MAC 积压超过阈值（默认 200）时 SADD，恢复后 SREM。与 global_pending 一起用于 is_queue_overloaded() 过载判断。sensitive:watermark — String（JSON）队列水位快照，每 15 秒 由 publish_watermark 定时任务更新，包含 total_pending、mac_count、active_workers，TTL 为发布间隔 + 5 秒。四、死信与异常 Keysensitive:dead_letter — List死信队列。OCR 任务 defer 超过最大次数（默认 5 次）后移入此处，保留最近 10000 条供人工排查。abnormal:parsed_queue — List异常数据解析队列。decter 路由收到的异常数据 RPUSH 入队，限制上限 5000 条，TTL 20 秒兜底。五、Celery Broker 队列（由 Celery 管理）队列名用途sensitive.dispatch调度器队列，运行 dispatch_next_mac_sensitive_tasks / process_mac_sensitive_taskssensitive.ocrOCR 敏感检测子任务队列sensitive.text文本敏感检测子任务队列Key 关系图 ┌─────────────────────┐ │ sensitive:schedule │ ← ZSet, 按等待时间排序 └──────────┬──────────┘ │ ┌─────────────────────┼─────────────────────┐ ▼ ▼ ▼ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ queue:<mac_A> │ │ queue:<mac_B> │ │ queue:<mac_C> │ ← List, 任务数据 │ active:<mac_A> │ │ active:<mac_B> │ │ active:<mac_C> │ ← String, 运行锁 │ meta:<mac_A> │ │ meta:<mac_B> │ │ meta:<mac_C> │ ← Hash, 元数据 └──────────────────┘ └──────────────────┘ └──────────────────┘ │ ┌─────────────┴─────────────┐ ▼ ▼ ┌──────────────────┐ ┌──────────────────────┐ │ global_p…

基于mac的设备客户端任务队列调度服务
type: item
date: 2026-06-18T17:39:19+08:00
description: 一、概述基于mac的设备客户端任务队列调度服务是一个基于 Redis 的、按设备标识（MAC）维度隔离的异步任务调度中间件，核心解决以下问题：多租户公平调度：大量设备共享有限的 Worker 资源，需保证各设备的任务都能被及时处理资源感知调度：不同任务对 CPU/GPU 资源需求不同，需感知 Worker 资源状况动态分配稳定性自保：防止某设备异常发量冲垮系统，也防止系统过载后雪崩该服务的核心思想是：为每个设备维护一个独立逻辑队列，通过老化优先级计算各设备的调度得分，结合分布式互斥锁实现多 Worker 安全的公平调度。二、数据模型整个调度系统的状态全部存储在 Redis 中，核心数据结构分为五类：┌─────────────────────────────────────────────────────────────────┐ │ MAC 队列调度系统 │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ 设备逻辑队列 │ 元数据 │ 调度集 │ 全局计数器 │ │ │ │ (List) │ (Hash) │ (ZSET) │ (String) │ │ │ │ │ │ │ │ │ │ │ task_mac_q:m1 │ meta:m1 │ schedule │ global_pend │ │ │ │ task_mac_q:m2 │ meta:m2 │ [m1:15.2] │ → 128 │ │ │ │ task_mac_q:m3 │ meta:m3 │ [m2:42.5] │ │ │ │ │ ... │ ... │ [m3:8.1] │ │ │ │ └────────────────┴──────────────┴───────────┴─────────────┘ │ │ │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ 运行锁 │ 过载集合 │ 死信队列 │ │ │ │ (String) │ (Set) │ (List) │ │ │ │ │ │ │ │ │ │ active:m1 → TTL │ overloaded: │ dead_letter │ │ │ │ active:m2 → TTL │ {m3, m7} │ [{task1}, ...] │ │ │ └──────────────────┴─────────────────┴──────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘2.1 数据结构详表用途Redis 类型键格式值说明设备队列Listsensitive:mac:queue:{mac}JSON 序列化的任务 payload，按入队顺序排列设备元数据Hashsensitive:mac:meta:{mac}字段：pending_count(积压数)、oldest_created_at(最早任务时间)、unschedulable_until(退避截止时间)、updated_at(更新时间)调度集合ZSetsensitive:mac:schedulemember = mac，score = 老化调度分数全局计数器Stringsensitive:global_pending所有设备积压任务总数（原子计数器）运行锁Stringsensitive:mac:active:{mac}SET NX + TTL，防止多 Worker 同时调度同一设备过载集合Setsensitive:overloaded_macs积压超过阈值的设备集合死信队列Listsensitive:dead_letter超过最大退避次数的任务记录水位快照Stringsensitive:watermark定期发布的系统运行状态 JSON三、核心操作3.1 入队流程（Enqueue）客户端请求 │ ▼ 检查全局过载 ──> 过载 ──> 拒绝入队，返回 503 │ ▼ (正常) 构建任务 payload (含 task_id, task_type, mac, created_at 等) │ ▼ Redis Pipeline: ├─ RPUSH queue_key, serialized_payload (追加到队列尾部) ├─ HINCRBY meta_key pending_count 1 (递增积压计数器) └─ INCR sensitive:global_pending (递增全局计数器) │ ▼ 如果是该设备的第一个任务 └─ HSET…

分布式任务调度与资源管理系统设计
type: item
date: 2026-06-18T17:37:04+08:00
description: 一、系统概述本系统是一个面向内容检测场景的分布式异步任务调度与资源管理平台，核心提供三方面能力：任务调度：基于设备维度的逻辑队列 + 老化优先级调度，实现多租户公平调度与饥饿感知资源调度：Worker 节点资源注册与心跳，CPU/GPU/内存感知的任务分发稳定性保障：过载保护、背压机制、指数退避、死信队列、任务 TTL 过期等多层可靠性策略系统基于 Flask + Celery + Redis + PostgreSQL 技术栈构建，采用三层架构（Web 路由层 → 业务服务层 → 异步任务层），所有异步任务通过 Celery 分发至专用队列，由 Worker 集群消费。二、整体架构┌─────────────────────────────────────────────────────────┐ │ Web 路由层 (Flask) │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 检测结果上报 │ │ 图片处理任务 │ │ 文本处理任务 │ │ │ │ /api/decter │ │ /api/ocr/* │ │ /api/ocr/* │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ └─────────┼──────────────────┼──────────────────┼──────────┘ │ │ │ ▼ ▼ ▼ ┌─────────────────────────────────────────────────────────┐ │ 服务层 (Services) │ │ ┌────────────────┐ ┌──────────────────────────────┐ │ │ │ 检测结果处理服务 │ │ 任务处理服务 │ │ │ │ (同步处理) │ │ (异步入队 + 过载检查) │ │ │ └────────────────┘ └──────────┬───────────────────┘ │ │ │ │ │ ┌───────────────────────────────▼────────────────────┐ │ │ │ MAC 队列调度服务 │ │ │ │ 按设备维度逻辑队列 · 老化优先级排序 · 分布式互斥锁 │ │ │ │ 指数退避 · 死信 · 饥饿检测 · 动态批大小 │ │ │ └───────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ 异步任务层 (Celery + Redis) │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ 调度队列 │ │ 图片处理队列 │ │ 文本处理队列 │ │ │ │ dispatch │ │ ocr │ │ text │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌──────────────────────────────────────────────────┐ │ │ │ Worker 集群 │ │ │ │ GPU Worker (消费所有队列) · CPU Worker (消费部分) │ │ │ │ Worker 注册/心跳 · 过载自检 · 自适应 CPU/GPU │ │ │ └──────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ 数据持久层 (PostgreSQL) │ │ 任务记录 · 处理结果 · 统计报表 │ └─────────────────────────────────────────────────────────┘三、任务调度系统3.1 核心调度模型系统采用两级调度模型：第一级：全局调度器 (Celery Beat + dispatch Task) → 从 Sorted Set 中选取…

关于鲁班猫rk3568适配的linux系统使用新版本python的情况
type: item
date: 2026-04-29T17:13:48+08:00
description: 前言开发了一些程序需要放到3568上运行。但是开发环境是python3.12，而最新版本的rk3568适配的系统是ubt20，系统默认的版本是python3.8其次，为了规避linux系统使用python安装依赖与系统依赖冲突的问题。需要采用python虚拟环境来运行情况1 默认版本是3.8需要编译3.12从源码编译（最通用、最干净）这是 ARM 开发板上安装新版 Python 的标准做法，虽然编译时间稍长，但兼容性最好，也不会破坏系统自带的 python3。1.安装编译依赖（已安装过的可跳过）sudo apt update sudo apt install -y wget build-essential libreadline-dev libncursesw5-dev \ libssl-dev libsqlite3-dev tk-dev libgdbm-dev libc6-dev libbz2-dev \ libffi-dev zlib1g-dev2.下载 Python 3.12 源码去 Python 官网 找最新的 3.12 版本链接（例如 3.12.4）：wget https://www.python.org/ftp/python/3.12.4/Python-3.12.4.tgz tar -xf Python-3.12.4.tgz cd Python-3.12.43.配置、编译、安装（使用 altinstall 避免覆盖系统 python3）# 配置（可选 --enable-optimizations 会提升性能但编译时间翻倍） ./configure --enable-optimizations # 使用全部CPU核心编译，RK3568大概需要30-60分钟 make -j$(nproc) # 关键：用 altinstall 而不是 install sudo make altinstall4.验证python3.12 --version你会看到 Python 3.12.x，同时系统默认的 python3 依然是 3.8.2，互不冲突。情况2 安装python3.12-venv报错这里的处理很简单，编译版本的python环境已经带了venv工具，直接使用python3.12 -m venv venv工具既可

自定协议，串口数据处理粘包方法
type: item
date: 2026-03-06T11:07:54+08:00
description: 问题处理串口数据过程中，当前后有连续的数据的情况下，很有可能出现数据粘包的情况。如需要的数据是ef0001和de002但是数据传输和接收过快产生粘包得到数据ef0001de00处理思路(自定协议)再数据包前增加1位或者2位的数据长度帧，通过先获取长度帧判断需要取的数据位。参考代码from .FingerProtocol import * import serial import time from ._Types import * from colorama import Fore, Style __all__ = [ "FingerManager", ] class FingerManager: def __init__(self,serial_conn:serial.Serial) -> None: self.serial_conn = serial_conn def send_cmd(self, packet): """发送命令包""" self.serial_conn.write(packet) self.serial_conn.flush() def read_packet(self,timeout=2): """ 读取一个完整的数据包 1. 读取包头9字节 (Header+Addr+ID+Len) 2. 解析包长度 (大端) 3. 读取剩余数据 (Len指定长度) 1. 检查是否超时 2. 读取数据 3. 检查是否完整 """ data = b'' start_time = time.time() # 1. 读取包头9字节 (Header+Addr+ID+Len) while len(data) < 9: if time.time() - start_time > timeout: return None if self.serial_conn.in_waiting > 0: data += self.serial_conn.read(min(9 - len(data), self.serial_conn.in_waiting)) else: time.sleep(0.001) # 2. 解析包长度 (大端) pkg_len = (data[7] << 8) | data[8] # 3. 读取剩余数据 (Len指定长度) total_len = 9 + pkg_len while len(data) < total_len: if time.time() - start_time > timeout: return None if self.serial_conn.in_waiting > 0: data += self.serial_conn.read(min(total_len - len(data), self.serial_conn.in_waiting)) else: time.sleep(0.001) return data

opencv运行报错找不到libGL.so.1
type: item
date: 2026-01-19T11:29:18+08:00
description: 报错：ImportError: libGL.so.1: cannot open shared object file: No such file or directory处理:sudo apt update sudo apt install -y libgl1 libglib2.0-0 libsm6 libxext6 libxrender1 libgomp1

如何通过python采集windows正在运行的进程信息
type: item
date: 2025-12-04T10:04:24+08:00
description: 基本思路使用psutil采集进程数据，通过pid对比提取出父子关系。示例import psutil import os import win32api import argparse import csv import json from datetime import datetime # 常见 Windows 系统关键进程（不完全列表，用于标记） SYSTEM_PROCESS_NAMES = { "System", "Idle", "smss.exe", "csrss.exe", "wininit.exe", "winlogon.exe", "services.exe", "lsass.exe", "lsm.exe", "svchost.exe", "dwm.exe", "conhost.exe", "fontdrvhost.exe", "audiodg.exe", "taskhostw.exe", "RuntimeBroker.exe", "sihost.exe", "ctfmon.exe", "dllhost.exe" } def get_file_version_info(file_path): try: lang, codepage = win32api.GetFileVersionInfo(file_path, '\\VarFileInfo\\Translation')[0] string_file_info = f'\\StringFileInfo\\{lang:04x}{codepage:04x}\\' company = win32api.GetFileVersionInfo(file_path, string_file_info + 'CompanyName') description = win32api.GetFileVersionInfo(file_path, string_file_info + 'FileDescription') return (company or "Unknown").strip(), (description or "No description").strip() except Exception: return "Unknown", "【版本信息不可读】文件可能损坏、非标准PE格式或权限不足" def get_process_source(proc): """推断进程启动来源""" try: parent_pid = proc.ppid() if parent_pid == 0: return "系统初始进程" parent = psutil.Process(parent_pid) parent_name = parent.name() if parent_name.lower() == "explorer.exe": return "用户启动（通过资源管理器）" elif parent_name.lower() in ["cmd.exe", "powershell.exe", "pwsh.exe", "windowsterminal.exe"]: return "用户启动（通过命令行）" elif parent_name.lower() in ["services.exe", "svchost.exe"]: return "系统服务启动" else: return f"{parent_name}" except (psutil.NoSuchProcess, psutil.AccessDenied): return "父进程已退出或无权限访问" except Exception: return "无法确定来源" def collect_running_processes(): processes_info = [] # 获取当前登录用户名（用于辅助判断） current_user = None try: current_user = psutil.users()[0].name if psutil.users() else "Unknown" except Exception: current_user = "Unknown" for proc in psutil.process_iter(['pid', 'name', 'username']): pid = proc.info['pid'] name = proc.info['name'] username = proc.info.get('username') or "Unknown" exe = None company = "Unknown" description = "" tag = "普通进程" sou…

关于pyqt6设置右对齐居中的示例
type: item
date: 2025-09-28T10:19:25+08:00
description: class InfoLayout(QVBoxLayout): def __init__(self): super().__init__() # 空白窗口 space_window = QWidget() space_window.setFixedHeight(100) # 英文标题 english_title_window = QWidget() english_title_window.setFixedHeight(80) english_title_window.setContentsMargins(0,0,0,-80) english_title_window.setStyleSheet("") ethl = QHBoxLayout() icon = QLabel("●") icon.setFont(QFont("SimHei", 8, QFont.Weight.Bold)) icon.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter) # 设置右对齐垂直居中 icon.setStyleSheet("color: #4CC9F0;") system_name = QLabel("SECURE ACCESS") system_name.setFont(QFont("SimHei", 24, QFont.Weight.DemiBold)) system_name.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter) # 设置左对齐垂直居中 system_name.setStyleSheet("color: white;") ethl.setStretch(0,1) ethl.setStretch(1,3) ethl.addWidget(icon) ethl.addWidget(system_name) english_title_window.setLayout(ethl) # 中文标题 title_window = QWidget() title_window.setFixedHeight(80) thl = QHBoxLayout() space_label = QLabel("") space_label.setAlignment(Qt.AlignmentFlag.AlignRight) title_label = QLabel("安全认证系统") title_label.setFixedHeight(80) title_label.setFont(QFont("SimHei", 18, QFont.Weight.Bold)) title_label.setAlignment(Qt.AlignmentFlag.AlignLeft) title_label.setStyleSheet("color: #8B98A9; margin: 20px;") thl.addWidget(space_label) thl.addWidget(title_label) title_window.setLayout(thl) # 产品介绍 piw = ProjectIntraduce() # 组合窗口 self.addWidget(space_window) self.addWidget(english_title_window) self.addWidget(title_window) self.addWidget(piw,1)

python怎么将项目的pip依赖拉成whl
type: item
date: 2025-06-25T10:51:00+08:00
description: 1. 生成依赖清单先确定项目的所有依赖项，生成requirements.txt文件：pip freeze > requirements.txt你也能手动编辑此文件，把不需要的依赖去除。2. 创建存放whl文件的目录mkdir wheels && cd wheels3. 下载依赖的whl包借助pip的download命令，依据requirements.txt来下载所有依赖的whl文件：pip download -r ../requirements.txt -d .若你有特定的Python版本或平台需求，可添加--python-version、--platform、--abi等参数。4. 下载项目自身的whl包（可选）若你想把项目自身也打包成whl文件，要先保证项目里有setup.py或者pyproject.toml文件，接着执行以下命令：cd .. # 返回项目根目录 python setup.py bdist_wheel生成的whl文件会存放在dist/目录下，你可以将其复制到之前创建的wheels/目录中。5. 验证whl文件要确保wheels/目录包含了所有需要的whl文件。离线环境安装方法在没有网络的环境中，可以用以下命令安装这些whl包：pip install --no-index --find-links=/path/to/wheels -r requirements.txt补充说明版本锁定：为避免版本不兼容的情况，建议使用pip freeze生成精确的依赖版本。环境隔离：推荐在虚拟环境中操作，防止混入不必要的依赖。平台兼容性：若要跨平台使用，需在对应的系统上分别下载whl包。