Python相关使用

读取文件

在python读取文件一般采用下述代码

with open(fname, 'r', encoding='utf-8') as f:  # 打开文件
    lines = f.readlines()  # 读取指针位置之后所有行
    line = f.readline()  # 读取指针位置所在行

f其实是文件指针位置，一个文件只有一个指针，readlines是将文件从指针位置读到结尾，读完后指针在结尾，此时再使用readline就没有东西可以读了。此时，我们可以使用seek来帮助文件指针移动到指定位置。

以读取文件首行和尾行的代码为例：

fname = 'test.txt'

with open(fname, 'r', encoding='utf-8') as f:  # 打开文件
    lines = f.readlines()  # 读取所有行
    first_line = lines[0]  # 取第一行
    last_line = lines[-1]  # 取最后一行

    print('文件' + fname + '第一行为：'+ first_line)
    print('文件' + fname + '最后一行为：' + last_line)
    
# 第二种
with open(fname, 'rb') as f:  # 打开文件
# 在文本文件中，没有使用b模式选项打开的文件，只允许从文件头开始,只能seek(offset,0)

    first_line = f.readline()  # 取第一行
	offset = -50  # 设置偏移量
    
    while True:
        """
        file.seek(off, whence=0)：从文件中移动off个操作标记（文件指针），正往结束方向移动，负往开始方向移动。
        如果设定了whence参数，就以whence设定的起始位为准，0代表从头开始，1代表当前位置，2代表文件最末尾位置。  
        """
        f.seek(offset, 2)  # seek(offset, 2)表示文件指针：从文件末尾(2)开始向前50个字符(-50)
        lines = f.readlines()  # 读取文件指针范围内所有行
        
        if len(lines) >= 2:  # 判断是否最后至少有两行，这样保证了最后一行是完整的
            last_line = lines[-1]  # 取最后一行
            break
            
        # 如果off为50时得到的readlines只有一行内容，那么不能保证最后一行是完整的
        # 所以off翻倍重新运行，直到readlines不止一行
        offset *= 2

    print('文件' + fname + '第一行为：' + first_line.decode())
    print('文件' + fname + '最后一行为：' + last_line.decode())

python 环境搭建

创建虚拟环境+安装库

1. 用 txt

   conda create -n myenv python=3.10
   conda activate myenv
   pip3 install -r requirements.txt

   requirements.txt example :

   numpy==1.24.0
   pandas>=1.5.0
   requests
   flask==2.3.0

2. 用yml

   conda env create -f environment.yml

   environment.yml example:

   name: aloha
   channels:
     - pytorch
     - nvidia
     - conda-forge
   dependencies:
     - python=3.9
     - pip=23.0.1
     - pyquaternion=0.9.9
     - pyyaml=6.0
     - rospkg=1.5.0
     - pexpect=4.8.0
     - mujoco=2.3.7
     - py-opencv=4.7.0
     - pip:
       - dm_control==1.0.14

3. 以“可编辑模式”安装当前目录下的 Python 包 (开发过程中安装项目自身作为一个包)

   my_project/
   ├── my_package/
   │   └── __init__.py
   ├── setup.py
   └── ...

   cd my_projectq
   pip install -e .

   会运行 setup.py example:

   from distutils.core import setup
   from setuptools import find_packages
   
   setup(
       name='detr',
       version='0.0.0',
       packages=find_packages(),
       license='MIT License',
       long_description=open('README.md').read(),
   )

   Python 会把 my_package 当作一个可导入的包注册到环境中，路径会被链接到源代码目录。

参考文章：https://zhuanlan.zhihu.com/p/98007747

“editable install” 可编辑安装

在当前环境中创建一个指向你的源码目录的软链接（而不是复制一份）。这使得你修改源代码后立即生效，不需要重新安装。

• 文件结构

  my_project/           <-- 项目根目录
  ├── __init__.py 
  ├── aa/            
  │   ├── __init__.py 	<-- ✅ 被认为是包
  │   ├── cli.py
  │   └── util
  │        ├── __init__.py      <-- ✅ 递归识别为子包
  │        └── box.py
  ├── bb/
  │   └── module2.py           <-- ❌ 没有 __init__.py，不会被识别
  ├── setup.py
  ├── README.md
  └── environment.yml

• setup.py example:

  from setuptools import setup, find_packages
  
  setup(
      name="my_package",
      version="0.1.0",
      description="A sample Python package for demonstration",
      author="Your Name",
      author_email="you@example.com",
      packages=find_packages(),  # 自动查找所有含 __init__.py 的当前目录和子目录
      install_requires=[
          "numpy>=1.21",
          "opencv-python",
      ],
      entry_points={
          "console_scripts": [
              "mytool=mypkg.cli:main",  # 安装后可用 `mytool` 命令运行 mypkg/cli.py 的 main()
          ],
      },
      classifiers=[
          "Programming Language :: Python :: 3",
      ],
      python_requires=">=3.7",
  )

• 运行

  cd my_project/
  pip install -e .

  被安装的包有['my_project', 'aa', 'aa.util']

• 查看查看 setup.py 中 find_packages() 实际找到了哪些包

  python -c "from setuptools import find_packages; print(find_packages())"

Python 会把 my_package 当作一个可导入的包注册到当前环境中，路径会被链接到源代码目录，只要在当前环境下，都可以使用该包。

在 Python 项目中导入文件

📁 假设项目结构如下：

my_project/
│
├── main.py
├── test.py
├── utils/
│   ├── __init__.py
│   ├── tools.py
│   └── helper.py
└── models/
    ├── __init__.py
    └── model.py

💡目标：在 main.py 中导入 helper.py 和 model.py 中的函数。均在my_project/下运行脚本

---

🧠 方法一：绝对导入（推荐用于模块包内）（从项目/包的根目录开始，给出完整路径）

在 main.py 中：

from utils.helper import some_function
from models.model import ModelClass

要求整个项目是一个包，也就是 utils/ 和 models/ 中含有 __init__.py 文件。

🧠 方法二：使用 sys.path.append() 动态添加路径

适用于临时脚本或测试用例，但不推荐长期使用：

import sys
sys.path.append('./utils')  # 添加utils目录到路径
from helper import some_function

🧠 方法三：把项目根目录添加到 PYTHONPATH 环境变量中

在终端：

export PYTHONPATH=/path/to/my_project:$PYTHONPATH

这样你就可以在任何地方运行：

from utils.helper import some_function

这个设置只对 当前终端会话 有效，关闭终端后就失效了。想要永久生效的话，可以把 export PYTHONPATH=... 写到 ~/.bashrc 或 ~/.zshrc 中。

🧠 方法四：使用相对导入（在包内部使用 . 或 ..）

只能在模块内使用（不能直接运行），例如：

在 utils/helper.py 中想导入 models/model.py和 utils/tools.py ：

from ..models.model import ModelClass  # .. 上一级包
from .tools import SomeTool # ⬅️ 使用相对导入
from utils.tools import SomeTool  # ⬅️ 使用绝对包路径导入

运行时需使用模块方式运行，例如：

python -m utils.helper # ✅ __name__ = 'utils.helper'
python utils/helper.py  # ❎ 如果有相对导入，这样会报错

1. 直接运行 helper.py时，Python 会把 utils/ 视为项目根目录 (sys.path[0])，也就是会将 utils/helper.py 当成一个顶层脚本，执行方式等价于： __name__ = '__main__'

2. 如果就要使用python utils/helper.py ,只有from tools import SomeTool可以被成功导入

Python 并发 / 并行

技术	解决什么	能否多核	是否受 GIL 影响	常见用途
多线程	I/O 等待	❌（算不了）	✅	读文件、网络
CPU 多进程	计算密集	✅	❌	数据处理、训练
GPU 多进程	大规模数值计算	✅（GPU）	❌	深度学习训练

---

1. Python 多线程（Threading）

定义：多个执行流，共享同一进程内存

• 所有线程共享：Python 对象、全局变量
• 创建和切换成本低

一个进程
 ├── 线程 A
 ├── 线程 B
 └── 线程 C

---

1.1 GIL（Global Interpreter Lock）

定义：同一时刻，只允许一个线程执行 Python 字节码。这是 CPython 的设计选择，用于：简化内存管理、保证对象操作安全

GIL 的直接后果：

• ❌ 对 CPU 密集任务：多线程 不能并行计算、线程轮流执行、多核 CPU 用不满

• ✅ 对 I/O 密集任务：有用，因为会 释放 GIL

1.2 “等 I/O”是什么意思（非常关键）

定义：I/O 已经提交给操作系统，但数据还没准备好，线程在等待结果

注意区分：

状态	是否占 CPU	是否占 GIL
执行 Python 代码	✅	✅
等 I/O（磁盘/网络）	❌	❌

多线程 I/O 的真实并发模型：👉 CPU 不干 I/O，只是协调

线程 A：发起 I/O → 等
线程 B：执行
线程 C：发起 I/O → 等

1.3 多线程能用多个 CPU 吗？

1. 执行 Python 代码时：❌ 不能（GIL 限制）
2. I/O 等待时：⚠️ CPU 可能调度到不同核，但不是在“算”
3. 调用 C/C++ 扩展时：✅ 可以（但不是 Python 线程本身）

1.4 典型适用场景

✅ 适合：文件读写、网络请求、爬虫、异步任务调度

❌ 不适合：for 循环算数、手写训练逻辑、大量数值计算

---

2. CPU 多进程（Multiprocessing）

定义：多个进程，各自拥有独立的 Python 解释器和 GIL

CPU 核 0 ← 进程 A（GIL A）
CPU 核 1 ← 进程 B（GIL B）
CPU 核 2 ← 进程 C（GIL C）

---

2.1 为什么多进程能真正多核？

因为每个进程：有自己的 GIL、有自己的内存空间、操作系统调度到不同 CPU 核

👉 是真正的并行计算

2.2 CPU 密集任务

定义：绝大多数时间花在计算上，而不是等待 I/O

典型特征：CPU 使用率接近 100%，任务复杂度随数据量线性或平方增长

2.3 多进程的代价

• 优点：绕过 GIL、多核利用率高、稳定（一个进程崩不影响其他）

• 缺点：内存不共享、进程通信（IPC）成本高、启动慢于线程

2.4 常见使用方式

• multiprocessing.Pool
• concurrent.futures.ProcessPoolExecutor
• joblib

2.5 典型适用场景

✅ 适合：数据预处理、特征工程、图像/音频解码、CPU 上的模型训练

❌ 不适合：高频小任务、强依赖共享状态的逻辑

---

3. GPU 多进程（分布式并行）

3.1 多 GPU 的本质

一张 GPU 对应一个进程

这是深度学习框架的标准设计

GPU 0 ← 进程 0
GPU 1 ← 进程 1
GPU 2 ← 进程 2

3.2 为什么不用多线程？

CUDA 上下文复杂、Python GIL 冲突、显存是 GPU 私有资源

👉 多线程模型不安全、不稳定、不高效

3.3 多 GPU 的并行模式

数据并行（Data Parallel）：每个 GPU 处理一部分数据，模型结构相同

流程：每个进程各自 forward + backward → 同步梯度（AllReduce） → 更新模型参数

3.4 这是“分布式”吗？

场景	性质
单机多卡	本地分布式
多机多卡	真正分布式

共同点：多进程、进程间通信、显式同步

3.5 GPU 内部的并行（容易混淆）

GPU 内部的上千线程 ≠ Python 线程

这是：硬件级并行（SIMT）、由 CUDA / 框架管理、Python 完全感知不到

3.6 典型适用场景

✅ 适合：深度学习训练、大规模矩阵运算、Transformer / CNN

❌ 不适合：小数据、轻量逻辑、频繁 CPU↔GPU 拷贝

---

好，这里我在原有笔记结构不变的基础上，直接给你补上
👉 三个「最小可运行代码模板」
每个都满足：能跑 / 能观察现象 / 不掺杂无关细节。

你可以把它们当成 “标准参考实现” 收藏。

---

4. 最小可运行代码模板（Very Important）

1️⃣ 多线程（Thread）模板

示例：多个线程同时等待 I/O 👉 I/O 密集任务（有效）

from concurrent.futures import ThreadPoolExecutor
import time

def io_task(i):
    print(f"任务 {i} 开始 I/O")
    time.sleep(2)   # 模拟 I/O 阻塞
    print(f"任务 {i} I/O 结束")
    return i

start = time.time()

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(io_task, range(4)))

print("结果:", results)
print("总耗时:", time.time() - start)

---

2️⃣ CPU 多进程（Process）模板

示例：多进程并行计算 👉 CPU 密集任务（真正并行）

from concurrent.futures import ProcessPoolExecutor
import time
import os

def cpu_task(n):
    print(f"进程 {os.getpid()} 开始任务 {n}")
    x = 0
    for i in range(50_000_000):
        x += i
    return os.getpid()

if __name__ == "__main__":
    start = time.time()

    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(cpu_task, range(4)))

    print("使用的进程 PID:", results)
    print("总耗时:", time.time() - start)
    
# 任务函数必须：顶层定义、可 pickle

---

3️⃣ GPU 多进程（DDP）模板

👉 多 GPU 数据并行（PyTorch）

这是最小“正确”的 DDP 示例，去掉了一切花活

文件：ddp_demo.py

import os
import torch
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
import torch.nn as nn

def main():
    dist.init_process_group("nccl")

    local_rank = int(os.environ["LOCAL_RANK"])
    torch.cuda.set_device(local_rank)

    model = nn.Linear(10, 1).cuda()
    model = DDP(model, device_ids=[local_rank])

    x = torch.randn(32, 10).cuda()
    y = model(x)
    y.mean().backward()

    print(f"Rank {local_rank} finished one step")

    dist.destroy_process_group()

if __name__ == "__main__":
    main()

---

启动方式（非常重要）

torchrun --nproc_per_node=2 ddp_demo.py

含义是：

• 启动 2 个进程
• 每个进程：绑定 1 张 GPU、拥有一份模型副本