PythonSTG开发心路

PythonSTG开发心路

前言

相关的文档参考详见

PySTG Docs
Python + OpenGL 东方 Project 风格弹幕射击游戏引擎文档

仓库地址在

GitHub - qwqpap/PythonSTG: Finally we find that python has invaded Touhou
Finally we find that python has invaded Touhou. Contribute to qwqpap/PythonSTG development by creating an account on GitHub.

这里还是主要介绍一下心路历程和技术细节啊嗯。

怎么来的

最开始想做这个的来源是2024年末想要用Luastg搓点好玩的弹幕小游戏在你矿1.5次例会上面玩,最开始就是一直用的Luastg,包括在2025年的时候在第二届百校天则大大方方卖的东方做题狙也是这么来的。

当时还非常非常原始,用Luastg的原版demo爆改了三面和一些ui贴图啥的就拿来卖,感觉也对不起当时花金币的游客。

之后就沉寂了很久,期间倒是一直有厚米催我更新,但是比较懒。

直到后面办九州拾遗,要做一个惊天地泣鬼神的牛逼stg,就想着直接整个重写一次得了,来接入Nonebot,其实现在想想我估计Luastg也能支持这个,但是总是有反复造轮子的快乐在里面的。

于是去年考研结束后就开始开发这个,最开始想法是用Rust做底层计算,再以一个Python库的形式供流程脚本调用,事后发现Rust的严格内存管理和我的狂放风格不是很搭边。

于是捡起来了大一用过的Numba+Numpy的组合直接在Python内部实现弹幕计算,惊人的,效率高到难以想象,甚至比Luastg和Zun自己的cpp引擎效率还要高得多。

这点让我很高兴。

高速的Python

GLI的罪恶

通常认为Python是非常低效的语言,因为使用了解释器导致无法预编译程序,所以效率瓶颈在解释器的执行速度上,所以在面对超大数组的遍历等操作时就会被别的非解释型语言拉开非常大的差距。

在多线程互锁行为中,Python提供了非常复古的GIL方式,通过上锁来确保每次访问机器码的只有一个进程,这还不是最卡顿的。

CPython采用了时间片轮转算法来控制各个进程的使用,并且每5ms就将当前执行之进程直接踹死到就绪队列,在本例的计算密集行为中,这种反复剥夺再重新分配同一个进程的行为使得高速计算直接不可使用。

亦有消息说Python正在大力推广No-GLI方法,我们静待花开。

于是我们引入两个高手工具:

Numpy与Numba让你的脚本飞起来

众所周知Python是弱类型语言,并且支持初始不指定类型,这在很多快速开发中很方便,但是在需要精确对其内存空间来实现数组操作的本例来说就炸飞了。

这里引入一小段碰撞检查的代码,其实大部分的计算都在这里。

@njit(cache=True)
def _check_player_vs_bullets(
    player_x: float, 
    player_y: float, 
    player_radius: float,
    bullet_pos: np.ndarray,  # shape: (N, 2)
    bullet_radius: np.ndarray,  # shape: (N,)
    bullet_alive: np.ndarray,  # shape: (N,)
) -> int:
    """
    检查玩家与子弹的碰撞(Numba加速)
    
    Returns:
        碰撞的子弹索引,-1表示无碰撞
    """
    n = bullet_pos.shape[0]
    for i in range(n):
        if bullet_alive[i] == 0:
            continue
        
        dx = bullet_pos[i, 0] - player_x
        dy = bullet_pos[i, 1] - player_y
        dist_sq = dx * dx + dy * dy
        
        combined_r = player_radius + bullet_radius[i]
        if dist_sq < combined_r * combined_r:
            return i
    
    return -1

我们以这个为例子

静态类型与局部性原理

通过直接在初始化时指定坐标的数据类型,这样在Numba编译时就会直接跳过查询这个变量类型的步骤,非常恐怖效率。

与此同时那个数组bullet_pos将会直接被作为整组数据装入内存的连续空间,具有极其友好的空间局部性。此时对于数组元素的访问将是极端快速的指针加减。效率不知道高到哪里去了。

对于单个子弹的两个坐标将是连续存放的,此时读取效率很高,在空间局部性下,所有的数据将会被直接放入缓存,直接省下一大笔访存时间。

投机取巧的神秘优化

在汇编语言中实现乘法是极其方便的,通过反复的位运算(幂次运算)与加法来实现乘法能够在可控的时钟周期内直接得出结果,然而开平方运算将会花掉非常多时间。

        dx = bullet_pos[i, 0] - player_x
        dy = bullet_pos[i, 1] - player_y
        dist_sq = dx * dx + dy * dy
        
        combined_r = player_radius + bullet_radius[i]
        if dist_sq < combined_r * combined_r:
            return i

注意到这一坨,为了避免开平方,给需要比较的两端(其实是距离)同时平方,此时乘方效率远高于开方。

在引擎中我实现了处处都是用这样的数据流来达到最快的速度。理论上此时速度不会输给直接用c写。

在失去了立即渲染模式的世界里,究竟会开出什么样颜色的花朵。

因为同时要传输上万个弹幕的数据给显卡,所以我们同时需要做两件事情,第一,不要反复在cpu和gpu中间传递数据,做到 只绘制一次。第二,把疯狂的简单的大量坐标计算全部丢给GPU自己算去,来给CPU更新别的数据留出充分的时间。在Python中我们使用moderngl库来实现这一切。

You Only Draw Once

如果每个子弹你都调用绘制,在上万个子弹出现时你的PCIE通道就会被大量重复数据日满,为了能够一次性把所有数据噗叽啪一下塞入显卡,需要一些神秘的优化。

        # 创建VAO
        self.vao = self.ctx.vertex_array(
            self.program,
            [
                (self.vertex_vbo, '2f 2f', 'in_vert', 'in_uv_base'),
                (self.position_vbo, '2f/i', 'in_offset'),
                (self.angle_vbo, '1f/i', 'in_angle'),
                (self.uv_vbo, '4f/i', 'in_uv_rect'),
                (self.scale_vbo, '2f/i', 'in_scale'),
            ]
        )

注意到这一段创建子弹的代码中的"/i",说明了上面的顶点是重用的,而下面这一堆是各个子弹私有的。

        # 上传数据到GPU(连续内存,高效)
        self.position_vbo.write(positions.tobytes())
        self.angle_vbo.write(angles.tobytes())
        self.uv_vbo.write(uvs.tobytes())
        self.scale_vbo.write(scales.tobytes())
        
        # 实例化渲染
        self.vao.render(moderngl.TRIANGLES, instances=count)

直接进行一个轰轰烈烈的顶点大上传,同时值得注意的是这个地方的bullet_pool直接.tobytes()从上一节中提到的连续内存空间中整个送入了VBO。

与此同时最后一行实例化渲染支持把上面这一堆参数整个放到显卡重用单次渲染,而不是一份份子弹画。

在这一通牛之大逼的操作之后,渲染方面至少在数据流通上不会有过分的瓶颈了。

好了,还有什么人,要计算。

    def _init_shader(self):
        """初始化着色器"""
        vertex_shader = """
        #version 330
        
        // 顶点属性
        in vec2 in_vert;
        in vec2 in_uv_base;
        
        // 实例属性
        in vec2 in_offset;      // 位置
        in float in_angle;      // 角度
        in vec4 in_uv_rect;     // UV矩形 [u_left, v_top, u_right, v_bottom]
        in vec2 in_scale;       // 尺寸
        
        out vec2 v_uv;
        
        uniform float u_y_scale;  // Y轴缩放因子
        
        void main() {
            // 缩放
            vec2 scaled = in_vert * in_scale;
            
            // 旋转
            float s = sin(in_angle);
            float c = cos(in_angle);
            vec2 rotated = vec2(
                scaled.x * c - scaled.y * s,
                scaled.x * s + scaled.y * c
            );
            
            // 平移
            vec2 position = rotated + in_offset;
            
            // 宽高比校正
            position.y *= u_y_scale;
            
            gl_Position = vec4(position, 0.0, 1.0);
            
            // 计算UV
            v_uv = in_uv_base * vec2(
                in_uv_rect.z - in_uv_rect.x,
                in_uv_rect.w - in_uv_rect.y
            ) + in_uv_rect.xy;
        }
        """

我们注意到这段看着很不Python的代码。

这其实是顶点着色器代码,但是这里有个小巧思,CPU没有计算出具体的顶点位置,与之相反的,其直接把整个原始数据丢给了GPU,让GPU并行计算这一堆数据。在这一步优化后,基本彻底解放了CPU,可以为下一步编撰流程脚本做更多事情了。

我比FastAPI还并发一万倍!如何通过协程来实现高速执行众多脚本。

并发与并行

强烈推荐阅读FastAPI的并发部分科普小漫画,不但可以学到什么是并发与并行,还能学到怎么调情。

并发 async / await - FastAPI
FastAPI framework, high performance, easy to learn, fast to code, ready for production

ok,简而言之就是通过让权等待的方式来达到狠狠地并发效果。

我们这里使用生成器函数来达到这个效果。

什么是生成器函数,就是类似按一下动一下的函数。我们用下面这个简单的例子举例

首先是第一个生成器函数

def stage_script():
    print("🎬 关卡开始!生成第一波敌人")
    yield "第一波敌人数据"  # 👈 暂停!函数在这里“存盘”,并把数据丢出去
    
    print("🔄 玩家击杀了第一波,继续向前走")
    yield "第二波敌人数据"  # 👈 再次暂停!
    
    print("🏁 关卡结束!")
    # 函数执行完毕,后续再调用 next() 会抛出 StopIteration 异常
  

注意到这函数没有return啊,其实是yield取代了return,每次调用生成器实例的next都会执行到下一个yield。

什么意思呢

# 1. 拿到生成器(此时一句话都不会打印)
coro = stage_script()

# 2. 按第一次播放键
res1 = next(coro) 
# 控制台打印: 🎬 关卡开始!生成第一波敌人
# res1 收到: "第一波敌人数据"

# 3. 按第二次播放键
res2 = next(coro)
# 控制台打印: 🔄 玩家击杀了第一波,继续向前走
# res2 收到: "第二波敌人数据"

通过这种方式就能让脚本能够假装主动地让出CPU时间给别的不闲着的脚本。

我们大大方方查看StageManager写了什么:

IO时让出你的CPU

对于大量IO的操作,必须给他狠狠地yield出来!

# ===== 阶段 1:加载画面 =====
self.loading_info = { ... }
yield  # 👈 第一次停顿:让游戏主循环有机会拿到 loading_info,把“Loading...”画面渲染到屏幕上

stage_dir = self._find_stage_directory(stage_class)
yield  # 👈 第二次停顿

if self._audio_manager and stage_dir:
    stage_bank.load_se_directory(se_dir)  # 💿 消耗时间的磁盘 I/O(读音效文件)
    yield  # 👈 第三次停顿:读完音效喘口气,让画面刷新一下,防止游戏界面卡死
    
    stage_bank.load_bgm_directory(bgm_dir) # 🎵 注册 BGM

关卡的实际流程处理

大大方方让出时间给关卡执行,在脚本结束后进行后处理,算是非常非常经典的游戏一般流程了。这样实现的是整个游戏流程就非常线性,从头到尾依次执行。

while stage._active:
    stage.update()
    yield  # 👈 游戏的大部分时间,都卡在这个 while 循环里,一帧一帧地打游戏

# 当 stage 脚本运行完(Boss 死了/通关了)
bullet_pool.clear_all()  # 🧹 清空满屏子弹

# 停留 120 帧(大约 2 秒),让玩家缓一缓
for _ in range(120):
    yield  # 👈 游戏通关后的结算/黑屏淡出时间

# 自动加载下一关
next_stage_cls = getattr(stage, '_next_stage_class', None)
if next_stage_cls is not None:
    self.load_stage(next_stage_cls)  # 🔄 协程套协程,套娃开始!

喜欢sleep的奶龙你完蛋了

不要在关卡内写任何sleep,Sleep是不让权等待,会阻塞整个游戏。

所以提供了这个方法

def wait(self, frames):
    """等待指定帧数(生成器函数)"""
    for _ in range(frames):
        yield

在游戏内直接

def boss_ai_script(ctx, stage_manager):
    # 1. 移动到屏幕中央
    boss.move_to(0, 0.5)
    
    # 2. 引擎,请帮我在这里卡 60 帧(1秒),让我摆个 Pose
    yield from stage_manager.wait(60)  # 👈 借用 wait 函数
    
    # 3. 释放第一阶段符卡
    ctx.play_se("spellcard_declare")
    boss.start_spellcard_1()

这样来让权等待

在这一套又一套的套套中,就能组织起来无比复杂的弹幕脚本逻辑了。

描述性游戏资产

懒得写了,,,,