Python Curses异步改造后动态实体闪烁问题排查求助

用Python curses绘制大量可动态移动实体，改为异步实现移动速度功能后，实体出现间歇性闪烁。

核心代码片段 #

main.py #

def main(stdscr) -> None:
    win = stdscr

    width = win.getmaxyx()[1]
    height = win.getmaxyx()[0]

    ov = Overworld(win, width, height)
    system = System(ov, sysinfo)

    return asyncio.run(system.run())


if __name__ == "__main__":
    stdscr = setup_stdscr()
    main(stdscr)

System模块核心代码 #

async def refresh_overworld(self):
        while self._running:
            self.overworld.draw(force_static=self._static_update_iter % REFRESH_STATIC_AFTER == 0)
            await asyncio.sleep(0.3)

    async def update_system_info(self):
        while self._running:
            self.system_info.draw()
            await asyncio.sleep(0.1)

    async def refresh_stdscr(self):
        while self._running:
            self.overworld.stdscr.refresh()
            await asyncio.sleep(0.1)

    async def run(self) -> None:
        self.overworld.spawn_entities()
        self.overworld.stdscr.nodelay(True)

        update_task = asyncio.create_task(self.overworld.update())
        refresh_task = asyncio.create_task(self.refresh_overworld())

        refresh_stdscr = asyncio.create_task(self.refresh_stdscr())

        try:
            while self._running:
                self._static_update_iter += 1
                await asyncio.sleep(MINUTE_LENGTH)
        finally:
            refresh_stdscr.cancel()

            update_task.cancel()
            refresh_task.cancel()

            key_listener_task.cancel()

            self.overworld.stdscr.nodelay(False)
            self.overworld.stdscr.clear()
            self.overworld.stdscr.refresh()

            curses.endwin()

Overworld模块核心代码 #

def _draw_entity(self, entity: Entity, position: Position):
        biome = self.biome.get_biome_by_coords(position.x, position.y)
        biome_color = self.biome.get_biome_color(biome)

        char = entity.representation

        try:
            self.stdscr.addstr(position.y, position.x, char, biome_color)
        except curses.error:
            pass

    def draw(self, force_static: bool = False):
        """
        Draw all the existing entities on the board.
        """
        logging.debug("Drawing entities in the overworld.")
        if not self._static_drawn or force_static:
            self.biome.draw()

            static_entities = [entity for entity in self.entities if not entity.dynamic]
            for entity in static_entities:
                self._draw_entity(entity, entity.position)
            self._static_drawn = True

        dynamic_entities = [entity for entity in self.entities if entity.dynamic]

        for entity in dynamic_entities:
            self._draw_entity(entity, entity.position)
        logging.debug("Entities drawn.")

    async def update_entity(self, entity):
        while True:
            await entity.update(self, self.biome)
            await asyncio.sleep(MINUTE_LENGTH / entity.properties.movement_speed)

    async def update(self):
        """
        Update all the entities in the overworld.
        """
        logging.debug("Updating entities in the overworld.")
        update_tasks = [asyncio.create_task(self.update_entity(entity)) for entity in self.entities if entity.dynamic]

        await asyncio.gather(*update_tasks)

        logging.info("Entities updated.")

1. 绘制与刷新异步分离：refresh_overworld负责绘制，refresh_stdscr负责单独刷新屏幕，两者是独立异步任务，可能出现绘制未完成就刷新的情况，导致屏幕显示不完整画面，表现为闪烁。
2. 动态实体旧位置未清除：实体移动后，旧位置的字符没有被覆盖（比如用背景色或空格清除），导致残影残留，看起来像是实体在闪烁。
3. 刷新频率不匹配：绘制任务间隔0.3秒，刷新任务间隔0.1秒，会出现多次刷新同一帧或刷新不完整帧的情况，加剧闪烁。
4. 并发状态修改冲突：多个实体的update_entity异步任务同时修改实体位置，draw任务执行时可能捕捉到实体位置的中间修改状态，导致绘制出异常位置的实体，造成闪烁。

1. 合并绘制与刷新任务 #

将屏幕刷新操作放到绘制任务内部，确保绘制完成后再统一刷新屏幕，避免中间状态被显示。修改System模块的refresh_overworld：

async def refresh_overworld(self):
    while self._running:
        self.overworld.draw(force_static=self._static_update_iter % REFRESH_STATIC_AFTER == 0)
        self.overworld.stdscr.refresh()  # 绘制完成后立即刷新
        await asyncio.sleep(1/30)  # 调整为合适帧率，比如30帧/秒

同时删除单独的refresh_stdscr任务，避免重复刷新。

2. 清除动态实体旧位置 #

在绘制实体新位置前，先清除其旧位置的字符。修改Overworld的_draw_entity方法：

def _draw_entity(self, entity: Entity, position: Position):
    # 清除旧位置（需给Entity添加old_position属性保存上一位置）
    if hasattr(entity, 'old_position') and entity.old_position != position:
        # 获取旧位置的背景 biome 颜色
        old_biome = self.biome.get_biome_by_coords(entity.old_position.x, entity.old_position.y)
        old_color = self.biome.get_biome_color(old_biome)
        try:
            self.stdscr.addstr(entity.old_position.y, entity.old_position.x, ' ', old_color)
        except curses.error:
            pass
    
    # 绘制新位置
    biome = self.biome.get_biome_by_coords(position.x, position.y)
    biome_color = self.biome.get_biome_color(biome)
    char = entity.representation
    try:
        self.stdscr.addstr(position.y, position.x, char, biome_color)
    except curses.error:
        pass

同时在Entity的update方法中保存旧位置：

async def update(self, overworld, biome):
    # 保存当前位置为旧位置
    self.old_position = self.position.copy()  # 假设Position类实现了copy方法
    # 执行移动逻辑
    # ...

3. 加锁避免并发状态冲突 #

给实体列表加锁，确保绘制时不会有实体位置被修改，避免绘制出异常状态：

# 在Overworld类中添加锁
import asyncio
class Overworld:
    def __init__(self, win, width, height):
        # 其他初始化代码
        self.entity_lock = asyncio.Lock()

    # 修改draw方法，加锁保护实体列表
    def draw(self, force_static: bool = False):
        logging.debug("Drawing entities in the overworld.")
        # 加锁防止绘制时实体被修改
        with self.entity_lock:
            if not self._static_drawn or force_static:
                self.biome.draw()
                static_entities = [entity for entity in self.entities if not entity.dynamic]
                for entity in static_entities:
                    self._draw_entity(entity, entity.position)
                self._static_drawn = True
            dynamic_entities = [entity for entity in self.entities if entity.dynamic]
            for entity in dynamic_entities:
                self._draw_entity(entity, entity.position)
        logging.debug("Entities drawn.")

    # 修改update_entity方法，更新位置时加锁
    async def update_entity(self, entity):
        while True:
            async with self.entity_lock:
                await entity.update(self, self.biome)
            await asyncio.sleep(MINUTE_LENGTH / entity.properties.movement_speed)

内容的提问来源于stack exchange，提问作者youwish