Python 工程语义：引用、导入、生成器与边界处理

这篇最大程度保留旧 Python 笔记中的语义、脚本与网络示例。可能移动原文件或缺少协议边界的代码已改写为可回退、可验证的版本。

模块、包与导入

推荐把项目作为包运行，而不是依赖当前工作目录碰巧可导入。

project/
├── pyproject.toml
├── src/
│   └── package_name/
│       ├── __init__.py
│       ├── core.py
│       └── tools/
│           ├── __init__.py
│           └── inspect.py
└── tests/

包内绝对导入：

from package_name.core import build_model

包内相对导入：

from ..core import build_model

从项目根目录运行模块：

python -m package_name.tools.inspect

若直接运行包内某个 .py 文件导致相对导入失败，优先修正启动方式和包结构，不要在代码里临时追加个人绝对路径到 sys.path。

列表保存的是对象引用

records = []
state = {"score": 1}

records.append(state)
state["score"] = 2
records.append(state)

print(records)
# [{'score': 2}, {'score': 2}]

两项指向同一个字典。如果需要每个时间步的快照：

from copy import deepcopy

records.append(deepcopy(state))

对于只包含不可变值的一层容器，state.copy() 可能足够；嵌套对象仍会共享引用。判断标准不是“看起来像复制”，而是后续是否会修改内部对象。

生成器是延迟计算，不是隐藏列表

def read_valid_lines(path):
    with open(path, encoding="utf-8") as stream:
        for line in stream:
            value = line.strip()
            if value:
                yield value

for value in read_valid_lines("input.txt"):
    print(value)

需要注意：

生成器通常只能消费一次；
代码执行到 yield 才暂停，并在下次迭代时继续；
with 中的资源会在生成器退出或被关闭时释放；
若确实要重复随机访问，应显式转成列表，并接受内存开销。

`*args`、`**kwargs` 与拆包

def configure(name, *layers, enabled=True, **metadata):
    return {
        "name": name,
        "layers": layers,
        "enabled": enabled,
        "metadata": metadata,
    }

options = {"enabled": False, "owner": "TEAM"}
result = configure("policy", 128, 64, **options)

函数签名应尽量显式。只有真正需要转发或扩展参数时才使用 **kwargs；否则拼写错误可能被悄悄收下。

容器拆包：

values = [1, 2, 3]
head, *middle, tail = values

卷积输出尺寸

对单个空间维度，普通卷积输出为：

$$ L_{\mathrm{out}} = \left\lfloor \frac{L_{\mathrm{in}} + 2P - D(K-1) - 1}{S} + 1 \right\rfloor $$

其中：

$L_{\mathrm{in}}$：输入长度；
$K$：卷积核大小；
$S$：步长；
$P$：padding；
$D$：dilation。

二维卷积对高和宽分别计算。PyTorch 验证示例：

import torch
from torch import nn

x = torch.randn(1, 1, 50, 50)
layer = nn.Conv2d(
    in_channels=1,
    out_channels=8,
    kernel_size=3,
    stride=2,
    padding=1,
    dilation=1,
)
y = layer(x)
print(y.shape)

旧公式缺少 floor 和 dilation，只在部分参数下碰巧正确。

批量媒体处理：先预演，再落盘

批处理最危险的部分不是图像缩放本身，而是文件覆盖、移动和异常恢复。推荐流程：

只读枚举输入，固定排序；
验证扩展名、尺寸和可解码性；
输出到新的临时目录；
核对数量和抽样结果；
最后再由人工决定是否替换原目录。

from pathlib import Path

source = Path("input_images")
files = sorted(
    path for path in source.iterdir()
    if path.suffix.lower() in {".jpg", ".jpeg", ".png"}
)

for index, path in enumerate(files):
    print(index, path.name)

不要让“生成视频”脚本顺便移动或删除原图。FFmpeg 调用也应检查退出码，并把输入帧率、输出帧率、编码器和像素格式显式写入项目记录。

UDP 与 TCP 的不同边界

UDP 保留数据报边界，但可能丢包、乱序或重复；TCP 可靠地传输字节流，却不保留消息边界。一次 send 不保证对应接收端的一次 recv。

工程中的 TCP 消息至少需要一种分帧协议：

固定长度；
换行等分隔符，并处理转义/长度上限；
固定长度头部携带正文长度；
使用已经定义好边界的上层协议。

客户端还应设置连接和读写超时、限制最大消息长度，并处理 DNS 返回多个地址的情况。服务端不应把任意网络输入直接交给 pickle、shell 或解释器执行。

UDP 广播前则要确认：

发送 socket 开启了广播选项；
目标广播地址属于正确网段；
防火墙允许所需的 UDP 端口；
接收端绑定的接口与地址正确；
业务协议能处理丢包和重复包。

包、`init.py` 与公开接口

模块是一个 .py 文件，包用于组织模块和子包。传统包通常含有 __init__.py：

package_name/
├── __init__.py
├── core.py
└── tools/
    ├── __init__.py
    └── inspect.py

__init__.py 可以为空，也可以显式重导出稳定接口：

from .core import build_model

__all__ = ["build_model"]

__all__ 主要影响 from package_name import *，不会形成安全边界，也不会阻止调用者直接导入其他模块。不要在 __init__.py 中执行耗时训练、网络请求或依赖当前目录的副作用；导入包应该可预测。

Python 3.3 起支持没有 __init__.py 的 namespace package，适合由多个发行包共同贡献同一命名空间。普通单仓库项目不必为了“更新”而删除 __init__.py；是否采用 namespace package 应由打包结构决定。

相对导入只适用于包上下文：

# package_name/tools/inspect.py
from ..core import build_model

直接执行 python package_name/tools/inspect.py 可能没有正确包上下文；应从项目入口运行 python -m package_name.tools.inspect。

浅拷贝与深拷贝

下面用 deepcopy 保存列表状态。它是安全的通用演示，但不是所有对象都需要深拷贝：

from copy import copy, deepcopy

state = {"position": [1.0, 2.0], "done": False}
shallow = copy(state)
deep = deepcopy(state)

state["position"][0] = 9.0
print(shallow["position"])  # [9.0, 2.0]：嵌套列表仍共享
print(deep["position"])     # [1.0, 2.0]

NumPy 数组、PyTorch 张量和包含文件/socket 的对象还有各自的复制语义。保存训练状态时，应明确需要的是 Python 容器副本、数组数据副本，还是从计算图分离的张量，例如 tensor.detach().clone()。

生成器的完整迭代过程

生成器函数在调用时返回生成器对象，直到第一次迭代才开始执行函数体：

def simple_generator():
    print("produce 1")
    yield 1
    print("produce 2")
    yield 2
    print("done")

generator = simple_generator()
print(next(generator))
print(next(generator))

再次 next(generator) 会抛出 StopIteration。通常用 for 消费，由循环处理结束状态：

for value in simple_generator():
    print("received", value)

无限序列必须由调用者设置边界：

from itertools import islice

def fibonacci():
    left, right = 0, 1
    while True:
        yield left
        left, right = right, left + right

print(list(islice(fibonacci(), 10)))

生成器表达式使用圆括号，适合单次流式消费：

squares = (value * value for value in range(5))
print(list(squares))

send、throw 和 close 可以与生成器双向交互，但控制流更难理解。现代异步任务通常优先使用明确的迭代器、async/await 或队列，而不是把复杂协程协议藏进普通生成器。

参数收集与调用拆包

形参顺序可以包含位置参数、可变位置参数、仅关键字参数和可变关键字参数：

def configure(name, *layers, enabled=True, **metadata):
    return name, layers, enabled, metadata

configure("policy", 128, 64, enabled=False, owner="TEAM")

调用时，* 展开可迭代对象，** 展开字符串键字典：

layers = [128, 64]
options = {"enabled": False, "owner": "TEAM"}
configure("policy", *layers, **options)

若显式参数和 **options 中出现同名键，调用会报重复参数错误。公共 API 不应把所有配置都塞进 **kwargs；显式形参更容易被类型检查、IDE 和调用者发现。

图片序列生成视频

旧 OpenCV 版本的核心思路可以保留，但需要检查空目录、解码失败、尺寸不一致和 writer 状态，并且不能移动原图：

from pathlib import Path
import re

import cv2


def natural_key(path: Path):
    return [
        int(part) if part.isdigit() else part.lower()
        for part in re.split(r"(\d+)", path.name)
    ]


def images_to_video(source_dir, output_file, fps=30.0):
    source = Path(source_dir)
    output = Path(output_file)
    if output.exists():
        raise FileExistsError(output)

    extensions = {".png", ".jpg", ".jpeg", ".bmp", ".tif", ".tiff"}
    frames = sorted(
        (path for path in source.iterdir() if path.suffix.lower() in extensions),
        key=natural_key,
    )
    if not frames:
        raise ValueError("no readable image candidates")

    first = cv2.imread(str(frames[0]))
    if first is None:
        raise ValueError(f"cannot decode {frames[0].name}")
    height, width = first.shape[:2]

    writer = cv2.VideoWriter(
        str(output),
        cv2.VideoWriter_fourcc(*"mp4v"),
        fps,
        (width, height),
    )
    if not writer.isOpened():
        raise RuntimeError("video writer could not be opened")

    try:
        for path in frames:
            frame = cv2.imread(str(path))
            if frame is None:
                raise ValueError(f"cannot decode {path.name}")
            if frame.shape[:2] != (height, width):
                raise ValueError(f"frame size mismatch: {path.name}")
            writer.write(frame)
    finally:
        writer.release()

这个版本只读输入目录、拒绝覆盖已有输出，也不会为了凑编号而重命名原图。运行后仍要核对帧数、时长和抽样画面。

已有严格编号的图片也可直接交给 FFmpeg，并用 -n 拒绝覆盖：

ffmpeg -n -framerate 30 -i 'frames/%04d.png' \
  -c:v libx264 -pix_fmt yuv420p output.mp4

去除音频且不重编码视频：

ffmpeg -n -i input.mp4 -map 0:v:0 -c:v copy -an output-silent.mp4

-map 0:v:0 明确选择第一条视频流；如果输入含多条视频、字幕或附件，应先用 ffprobe 查看流并决定保留范围。

图片缩放与 SVG 转 PDF

输出使用不同文件名，避免覆盖原件：

from pathlib import Path
from PIL import Image

source = Path("input.jpg")
destination = Path("output-resized.jpg")
if destination.exists():
    raise FileExistsError(destination)

with Image.open(source) as image:
    resized = image.resize((240, 320), Image.Resampling.LANCZOS)
    resized.save(destination)

这里会强制变成指定宽高，可能改变宽高比。需要保持比例时用 thumbnail 或根据原始尺寸计算目标尺寸。

CairoSVG 已在项目依赖中时，可转换 SVG：

from pathlib import Path
import cairosvg

source = Path("figure.svg")
destination = Path("figure.pdf")
if destination.exists():
    raise FileExistsError(destination)
cairosvg.svg2pdf(url=str(source), write_to=str(destination))

SVG 可以引用外部资源；只转换可信输入，并在交付前检查字体、透明度和页面边界。

UDP 最小示例

单次发送和接收可以用标准库完成。地址、端口都是占位符：

import socket


def send_udp(host, port, message):
    payload = message.encode("utf-8")
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.sendto(payload, (host, port))


def receive_udp(bind_host, port, max_size=4096):
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.bind((bind_host, port))
        sock.settimeout(5.0)
        payload, peer = sock.recvfrom(max_size)
        return payload.decode("utf-8"), peer

广播发送需要显式启用广播，并使用当前网段的正确广播地址：

with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    sock.sendto(b"discovery", ("BROADCAST_ADDRESS", 9999))

BROADCAST_ADDRESS 必须替换并核对。UDP 没有内建身份验证；发现协议不应把收到的任意内容当成可信指令。

TCP 长度前缀示例

TCP 是字节流。发送一条 UTF-8 消息时，可先发送固定长度头部：

import socket
import struct


def send_message(host, port, message, timeout=5.0):
    payload = message.encode("utf-8")
    if len(payload) > 1_000_000:
        raise ValueError("message too large")
    frame = struct.pack("!I", len(payload)) + payload
    with socket.create_connection((host, port), timeout=timeout) as sock:
        sock.sendall(frame)

接收端必须循环读取准确字节数，不能假定一次 recv 就得到完整内容：

def recv_exact(sock, size):
    chunks = []
    remaining = size
    while remaining:
        chunk = sock.recv(remaining)
        if not chunk:
            raise ConnectionError("connection closed mid-frame")
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)


def recv_message(sock, max_size=1_000_000):
    size = struct.unpack("!I", recv_exact(sock, 4))[0]
    if size > max_size:
        raise ValueError("message too large")
    return recv_exact(sock, size).decode("utf-8")

长度前缀只解决消息边界，不提供加密或认证。跨不可信网络应使用 TLS 或成熟的上层协议。

小结

Python 工程中的许多问题并非语法错误，而是边界假设错误：把引用当副本、把工作目录当包路径、把 TCP 当消息队列、把批处理成功当作原数据安全。把这些假设写成可验证的检查，代码会比增加更多“技巧”可靠。