AFSim学习-自定义Processor开发5-通过文件进行python强化学习交互示意

1. 本文目标

上一篇中已经完成了一个基础闭环：

AFSim 自定义 Processor 通过文件和 Python 进行交互。

也就是说，AFSim 侧会周期性写出观测信息，Python 侧读取观测信息后，再把控制动作写回文件，AFSim 再读取动作并控制平台运动。

本文在上一篇的基础上继续向强化学习方向推进。

本篇的核心变化是：

AFSim 工程和 Processor 代码保持上一篇状态不动，只修改 Python 文件，让 Python 主动启动和管理 AFSim 进程，从而形成一个类似强化学习环境的 reset() / step() 交互结构。

这样做之后，Python 不再只是一个被动读取文件、写入动作的脚本，而是可以像强化学习环境一样控制每一局仿真的开始、推进和结束。

---

2. 总体思路

上一篇的交互方式可以理解为：

手动启动 AFSim
    ↓
AFSim 写 observation.txt
    ↓
Python 读取 observation.txt
    ↓
Python 写 action.txt
    ↓
AFSim 读取 action.txt
    ↓
AFSim 推进下一步

本文改成：

Python 启动 AFSim 进程
    ↓
Python 等待第一帧 observation.txt
    ↓
Python 根据 observation 计算 action
    ↓
Python 写入 action.txt
    ↓
AFSim 读取 action 并推进
    ↓
Python 等待下一帧 observation.txt
    ↓
循环 step
    ↓
episode 结束后 Python 关闭 AFSim

也就是说，Python 负责管理整个 episode 的生命周期：

reset()：启动一局新的 AFSim 仿真
step() ：写动作，等待 AFSim 返回下一帧观测
close()：关闭 AFSim 进程

这已经很接近强化学习中常见的 Gym 环境结构。

---

3. 保持 AFSim 侧不变

本篇有一个重要前提：

AFSim 保持上一篇博文的状态不动。

也就是说，上一篇中已经实现的内容不需要修改：

1. AFSim 自定义 Processor 仍然负责写出 observation.txt；
2. AFSim 自定义 Processor 仍然负责读取 action.txt；
3. 文件格式保持不变；
4. 平台控制逻辑保持不变；
5. 只修改 Python 文件。

本文重点不是重新改 AFSim 侧，而是把 Python 侧改造成一个可以主动调用 AFSim 进程的环境控制器。

---

4. 文件交互接口

本文使用三个主要文件：

OBS_FILE = observation.txt
ACTION_FILE = action.txt
DECISION_FILE = controller_decision.csv

其中：

文件	作用
observation.txt	AFSim 写给 Python 的观测信息
action.txt	Python 写给 AFSim 的动作信息
controller_decision.csv	可选的控制决策记录文件

当前主要使用前两个文件完成闭环交互。

---

5. observation 文件格式

Python 侧读取的 observation.txt 格式为：

step_id sim_time range_m relative_bearing_deg elevation_deg closing_speed course_deg

例如：

12 60.0 185000.0 -3.25 0.02 450.0 91.0

各字段含义如下：

字段	含义
step_id	当前仿真步编号
sim_time	当前仿真时间
range_m	双方距离，单位 m
relative_bearing_deg	相对方位角，单位度
elevation_deg	高低角，单位度
closing_speed	接近速度
course_deg	当前航向角

Python 每次读取时，会判断 step_id 是否变化。

只有当新的 step_id 和上一次不同，才认为 AFSim 已经推进到了新的状态。

---

6. action 文件格式

Python 写给 AFSim 的 action.txt 格式为：

step_id heading_delta_deg

例如：

12 5.0

含义是：

字段	含义
step_id	当前动作对应的仿真步编号
heading_delta_deg	航向调整量，单位度

本文中动作被限制在：

-10.0 <= heading_delta_deg <= 10.0

也就是每一步最多调整 10 度航向。

---

7. Python 主动启动 AFSim

本文最关键的变化是这一段：

proc = subprocess.Popen(
    [AFSIM_EXE, "-sm", MISSION_FILE],
    cwd=AFSIM_WORKDIR,
    stdout=log_fp,
    stderr=subprocess.STDOUT,
    text=True
)

它的作用是让 Python 启动一个新的 AFSim 进程。

其中：

AFSIM_EXE = "./buil/mission"
MISSION_FILE = "/home/ubuntu/afsim_ws/project1/12-rl-platform/two_platform_logger.txt"

对应命令大致等价于：

cd /home/ubuntu/afsim_ws/afsim-src/afsim/swdev
./buil/mission -sm /home/ubuntu/afsim_ws/project1/12-rl-platform/two_platform_logger.txt

如果自己的 AFSim 可执行文件路径不同，需要根据实际情况修改 AFSIM_EXE。

例如有些工程目录可能是：

AFSIM_EXE = "./build/mission"

本文示例中保持当前工程路径配置。

---

8. 为什么 reset 前要清理旧文件？

代码中有一个函数：

def clean_interface_files():
    """
    reset 前清理旧文件。
    这一步很重要，否则新一局可能读到上一局残留的 observation/action。
    """
    safe_remove(OBS_FILE)
    safe_remove(OBS_FILE + ".tmp")
    safe_remove(ACTION_FILE)
    safe_remove(ACTION_FILE + ".tmp")
    safe_remove(DECISION_FILE)

这个函数非常重要。

因为文件交互有一个常见问题：

如果上一局仿真结束后，observation.txt 或 action.txt 没有清理，那么下一局刚开始时，Python 可能会读到上一局残留的数据。

这会导致 reset 后读取到错误的初始状态。

所以每次新 episode 开始之前，要先删除旧的交互文件。

---

9. 为什么写 action 时要使用临时文件？

写 action 时没有直接写 action.txt，而是先写：

tmp_file = ACTION_FILE + ".tmp"

然后再执行：

os.replace(tmp_file, ACTION_FILE)

完整代码如下：

def write_action(step_id, heading_delta_deg):
    """
    写 action 给 AFSIM。
    格式：
    step_id heading_delta_deg
    """
    heading_delta_deg = clamp(
        heading_delta_deg,
        -MAX_HEADING_DELTA_DEG,
        MAX_HEADING_DELTA_DEG
    )

    tmp_file = ACTION_FILE + ".tmp"

    with open(tmp_file, "w", encoding="utf-8") as f:
        f.write(f"{step_id} {heading_delta_deg}\n")

    os.replace(tmp_file, ACTION_FILE)

这样做的原因是避免 AFSim 读到半截文件。

如果 Python 正在写 action.txt，而 AFSim 同时读取这个文件，就可能出现：

12

或者：

12 5

这种还没有写完整的数据。

使用临时文件后，流程变成：

Python 先完整写入 action.txt.tmp
    ↓
写完后一次性替换成 action.txt
    ↓
AFSim 读取 action.txt

这样更加安全。

---

10. AFSIM 环境封装

本文将 AFSim 封装成了一个简单的环境类：

class AfsimEnv:
    def __init__(self):
        self.proc = None
        self.log_fp = None
        self.episode_id = 0
        self.last_step_id = -1

这个类主要提供三个函数：

reset()
step()
close()

这和强化学习环境很相似。

---

11. reset 过程

reset() 的逻辑如下：

def reset(self):
    """
    重置环境：
    1. 停止旧 AFSIM
    2. 清理旧 observation/action 文件
    3. 启动新的 AFSIM
    4. 等第一帧 observation
    """
    stop_afsim(self.proc, self.log_fp)

    clean_interface_files()

    self.proc, self.log_fp = start_afsim(self.episode_id)

    self.last_step_id = -1

    obs = read_observation(self.last_step_id, timeout_sec=30.0)

    if obs is None:
        raise RuntimeError("reset failed: no initial observation from AFSIM")

    self.last_step_id = obs["step_id"]

    self.episode_id += 1

    return obs

它完成了四件事：

1. 停止上一局 AFSim；
2. 清理旧的交互文件；
3. 启动新的 AFSim 进程；
4. 等待 AFSim 写出第一帧 observation。

如果 30 秒内没有读到初始 observation，就认为 reset 失败。

---

12. step 过程

step() 的逻辑如下：

def step(self, heading_delta_deg):
    """
    执行动作：
    1. Python 写 action
    2. 等 AFSIM 推进后写出下一帧 observation
    3. 计算 reward 和 done
    """
    current_step_id = self.last_step_id

    write_action(current_step_id, heading_delta_deg)

    obs = read_observation(self.last_step_id, timeout_sec=30.0)

    if obs is None:
        # AFSIM 可能已经自然结束
        return None, 0.0, True, {"reason": "no observation"}

    self.last_step_id = obs["step_id"]

    reward, done = compute_reward_done(obs)

    return obs, reward, done, {}

每次 step 做三件事：

写 action
    ↓
等待新的 observation
    ↓
计算 reward 和 done

返回值为：

obs, reward, done, info

这也和 Gym 环境的返回形式类似。

---

13. 第一版 reward 设计

当前 reward 函数如下：

def compute_reward_done(obs):
    """
    第一版 reward / done。
    后面接 PPO 时，这里就是环境的奖励函数。
    """
    range_m = obs["range_m"]
    rb = obs["relative_bearing_deg"]
    closing_speed = obs["closing_speed"]

    reward = (
        - range_m / 300000.0
        - abs(rb) / 180.0
        + closing_speed / 1000.0
    )

    done = False

    # 成功条件：距离小于 50 km
    if range_m < 50000.0:
        reward += 10.0
        done = True

    # 失败条件：距离太远
    if range_m > 400000.0:
        reward -= 10.0
        done = True

    return reward, done

这个 reward 只是第一版示意，主要包含三个因素：

- range_m / 300000.0

距离越远，惩罚越大。

- abs(rb) / 180.0

相对方位角偏差越大，惩罚越大。

+ closing_speed / 1000.0

接近速度越大，奖励越高。

成功条件：

range_m < 50000.0

也就是距离小于 50 km。

失败条件：

range_m > 400000.0

也就是距离大于 400 km。

后续真正接 PPO、DQN、SAC 等强化学习算法时，可以继续调整这个 reward。

---

14. 当前策略：规则策略代替强化学习

目前还没有真正接入神经网络策略，而是先使用一个简单规则策略：

def simple_policy(obs):
    """
    现在先用规则策略代替 RL。
    后面这里可以替换成神经网络输出动作。
    """
    rb = obs["relative_bearing_deg"]

    heading_delta_deg = clamp(
        rb,
        -MAX_HEADING_DELTA_DEG,
        MAX_HEADING_DELTA_DEG
    )

    return heading_delta_deg

它的逻辑很简单：

相对方位角是多少，就尝试向对应方向调整航向。

同时动作被限制在：

[-10°, 10°]

所以如果相对方位角是 25 度，实际动作会被限制为 10 度。

如果相对方位角是 -18 度，实际动作会被限制为 -10 度。

---

15. 完整 Python 示例代码

import os
import time
import subprocess
import signal


# ==============================
# 路径配置
# ==============================

AFSIM_WORKDIR = "/home/ubuntu/afsim_ws/afsim-src/afsim/swdev"

AFSIM_EXE = "./buil/mission"

MISSION_FILE = "/home/ubuntu/afsim_ws/project1/12-rl-platform/two_platform_logger.txt"

BASE_DIR = "/home/ubuntu/afsim_ws/project1/12-rl-platform"

OBS_FILE = os.path.join(BASE_DIR, "observation.txt")
ACTION_FILE = os.path.join(BASE_DIR, "action.txt")
DECISION_FILE = os.path.join(BASE_DIR, "controller_decision.csv")


MAX_HEADING_DELTA_DEG = 10.0


# ==============================
# 工具函数
# ==============================

def clamp(x, low, high):
    return max(low, min(high, x))


def safe_remove(path):
    try:
        os.remove(path)
    except FileNotFoundError:
        pass


def clean_interface_files():
    """
    reset 前清理旧文件。
    这一步很重要，否则新一局可能读到上一局残留的 observation/action。
    """
    safe_remove(OBS_FILE)
    safe_remove(OBS_FILE + ".tmp")
    safe_remove(ACTION_FILE)
    safe_remove(ACTION_FILE + ".tmp")
    safe_remove(DECISION_FILE)


def start_afsim(episode_id):
    """
    启动一个新的 AFSIM 进程。
    这就相当于 reset 后开始新 episode。
    """
    log_file = os.path.join(BASE_DIR, f"afsim_episode_{episode_id}.log")

    log_fp = open(log_file, "w", encoding="utf-8")

    proc = subprocess.Popen(
        [AFSIM_EXE, "-sm", MISSION_FILE],
        cwd=AFSIM_WORKDIR,
        stdout=log_fp,
        stderr=subprocess.STDOUT,
        text=True
    )

    return proc, log_fp


def stop_afsim(proc, log_fp=None):
    """
    停止 AFSIM 进程。
    如果仿真已经结束，直接返回。
    如果还在等 action，就 terminate。
    """
    if proc is not None and proc.poll() is None:
        proc.terminate()

        try:
            proc.wait(timeout=2.0)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait()

    if log_fp is not None:
        log_fp.close()


def read_observation(last_step_id, timeout_sec=30.0):
    """
    等待 AFSIM 写新的 observation。
    只有 step_id 和 last_step_id 不同，才认为是新 observation。
    """
    start_time = time.time()

    while time.time() - start_time < timeout_sec:
        if not os.path.exists(OBS_FILE):
            time.sleep(0.01)
            continue

        try:
            with open(OBS_FILE, "r", encoding="utf-8") as f:
                line = f.readline().strip()
        except Exception:
            time.sleep(0.01)
            continue

        if not line:
            time.sleep(0.01)
            continue

        parts = line.split()

        if len(parts) < 7:
            time.sleep(0.01)
            continue

        try:
            step_id = int(parts[0])
            sim_time = float(parts[1])
            range_m = float(parts[2])
            relative_bearing_deg = float(parts[3])
            elevation_deg = float(parts[4])
            closing_speed = float(parts[5])
            course_deg = float(parts[6])
        except ValueError:
            time.sleep(0.01)
            continue

        if step_id == last_step_id:
            time.sleep(0.01)
            continue

        obs = {
            "step_id": step_id,
            "sim_time": sim_time,
            "range_m": range_m,
            "relative_bearing_deg": relative_bearing_deg,
            "elevation_deg": elevation_deg,
            "closing_speed": closing_speed,
            "course_deg": course_deg,
        }

        return obs

    return None


def write_action(step_id, heading_delta_deg):
    """
    写 action 给 AFSIM。
    格式：
    step_id heading_delta_deg
    """
    heading_delta_deg = clamp(
        heading_delta_deg,
        -MAX_HEADING_DELTA_DEG,
        MAX_HEADING_DELTA_DEG
    )

    tmp_file = ACTION_FILE + ".tmp"

    with open(tmp_file, "w", encoding="utf-8") as f:
        f.write(f"{step_id} {heading_delta_deg}\n")

    os.replace(tmp_file, ACTION_FILE)


def compute_reward_done(obs):
    """
    第一版 reward / done。
    后面接 PPO 时，这里就是环境的奖励函数。
    """
    range_m = obs["range_m"]
    rb = obs["relative_bearing_deg"]
    closing_speed = obs["closing_speed"]

    reward = (
        - range_m / 300000.0
        - abs(rb) / 180.0
        + closing_speed / 1000.0
    )

    done = False

    # 成功条件：距离小于 50 km
    if range_m < 50000.0:
        reward += 10.0
        done = True

    # 失败条件：距离太远
    if range_m > 400000.0:
        reward -= 10.0
        done = True

    return reward, done


def simple_policy(obs):
    """
    现在先用规则策略代替 RL。
    后面这里可以替换成神经网络输出动作。
    """
    rb = obs["relative_bearing_deg"]

    heading_delta_deg = clamp(
        rb,
        -MAX_HEADING_DELTA_DEG,
        MAX_HEADING_DELTA_DEG
    )

    return heading_delta_deg


# ==============================
# AFSIM 环境封装
# ==============================

class AfsimEnv:
    def __init__(self):
        self.proc = None
        self.log_fp = None
        self.episode_id = 0
        self.last_step_id = -1

    def reset(self):
        """
        重置环境：
        1. 停止旧 AFSIM
        2. 清理旧 observation/action 文件
        3. 启动新的 AFSIM
        4. 等第一帧 observation
        """
        stop_afsim(self.proc, self.log_fp)

        clean_interface_files()

        self.proc, self.log_fp = start_afsim(self.episode_id)

        self.last_step_id = -1

        obs = read_observation(self.last_step_id, timeout_sec=30.0)

        if obs is None:
            raise RuntimeError("reset failed: no initial observation from AFSIM")

        self.last_step_id = obs["step_id"]

        self.episode_id += 1

        return obs

    def step(self, heading_delta_deg):
        """
        执行动作：
        1. Python 写 action
        2. 等 AFSIM 推进后写出下一帧 observation
        3. 计算 reward 和 done
        """
        current_step_id = self.last_step_id

        write_action(current_step_id, heading_delta_deg)

        obs = read_observation(self.last_step_id, timeout_sec=30.0)

        if obs is None:
            # AFSIM 可能已经自然结束
            return None, 0.0, True, {"reason": "no observation"}

        self.last_step_id = obs["step_id"]

        reward, done = compute_reward_done(obs)

        return obs, reward, done, {}

    def close(self):
        stop_afsim(self.proc, self.log_fp)
        self.proc = None
        self.log_fp = None


# ==============================
# 测试运行
# ==============================

if __name__ == "__main__":
    env = AfsimEnv()

    num_episodes = 3

    for ep in range(num_episodes):
        print(f"\n========== Episode {ep} reset ==========")

        obs = env.reset()

        total_reward = 0.0

        print(
            f"[RESET] step={obs['step_id']} "
            f"t={obs['sim_time']} "
            f"range={obs['range_m']:.1f} "
            f"rb={obs['relative_bearing_deg']:.4f}"
        )

        for t in range(100):
            action = simple_policy(obs)

            obs, reward, done, info = env.step(action)

            total_reward += reward

            if obs is None:
                print("[DONE] no observation, AFSIM ended")
                break

            print(
                f"[STEP] ep={ep} "
                f"step={obs['step_id']} "
                f"sim_t={obs['sim_time']:.1f} "
                f"range={obs['range_m']:.1f} "
                f"rb={obs['relative_bearing_deg']:.4f} "
                f"action={action:.4f} "
                f"reward={reward:.4f} "
                f"done={done}"
            )

            if done:
                print(f"[DONE] episode={ep}, total_reward={total_reward:.4f}")
                break

        env.close()

    print("All episodes finished.")

---

16. 运行方式

假设 Python 文件名为：

rl_file_env.py

可以直接运行：

python3 rl_file_env.py

运行后，Python 会自动启动 AFSim，并打印每一步的状态：

========== Episode 0 reset ==========
[RESET] step=0 t=0.0 range=...
[STEP] ep=0 step=1 sim_t=... range=... rb=... action=... reward=... done=False
[STEP] ep=0 step=2 sim_t=... range=... rb=... action=... reward=... done=False
...
[DONE] episode=0, total_reward=...

同时，每个 episode 会生成一个 AFSim 日志文件：

afsim_episode_0.log
afsim_episode_1.log
afsim_episode_2.log

这样便于排查 AFSim 是否正常启动、是否正常读取 action、是否正常写出 observation。

---

17. 当前代码对应的强化学习结构

虽然当前还没有真正接入 PPO，但这个结构已经具备强化学习环境的基本形式。

对应关系如下：

强化学习概念	当前代码
环境	AfsimEnv
重置环境	env.reset()
执行动作	env.step(action)
状态	obs
动作	heading_delta_deg
奖励	reward
是否结束	done
策略	simple_policy(obs)
episode	一次完整 AFSim 仿真

当前主循环：

for ep in range(num_episodes):
    obs = env.reset()

    for t in range(100):
        action = simple_policy(obs)
        obs, reward, done, info = env.step(action)

        if done:
            break

后续接强化学习时，只需要把：

action = simple_policy(obs)

替换成：

action = policy_network(obs)

或者 PPO 中的：

action, log_prob, value = agent.select_action(obs)

即可。

---

18. 当前版本的意义

这一版的意义主要有三点。

第一，Python 已经可以主动管理 AFSim 进程。

上一篇中，AFSim 需要手动启动。现在每次 episode 都可以由 Python 自动启动和关闭。

第二，已经形成了类似 Gym 的环境接口。

当前代码已经具备：

reset()
step()
close()

这为后续接入 PPO、DQN、SAC 等强化学习算法打下基础。

第三，文件交互方式仍然保持简单。

虽然文件交互不是最高效的方式，但它非常适合早期验证，因为：

1. 容易调试；
2. 不需要复杂通信框架；
3. AFSim 和 Python 解耦；
4. 任何一侧出错都可以通过文件和日志排查。

---

19. 当前版本需要注意的问题

当前版本仍然是一个示意版本，后续还需要继续完善。

19.1 文件轮询效率不高

当前 Python 使用：

time.sleep(0.01)

不断轮询 observation.txt。

这种方式简单可靠，但效率不算高。后续如果训练规模变大，可以考虑：

1. socket 通信；
2. ZeroMQ；
3. gRPC；
4. 共享内存；
5. AFSim 插件直接调用 Python 接口。

不过在早期验证阶段，文件交互已经足够。

19.2 reward 还比较粗糙

当前 reward 只是第一版示意：

reward = (
    - range_m / 300000.0
    - abs(rb) / 180.0
    + closing_speed / 1000.0
)

后续可以继续加入：

1. 是否保持合理航向；
2. 是否进入有效攻击包线；
3. 是否避免过大机动；
4. 是否保持能量优势；
5. 是否接近目标但不过冲；
6. 是否满足任务约束。

19.3 done 条件还比较简单

当前 done 条件只有：

range_m < 50000.0
range_m > 400000.0

后续可以加入更多终止条件，例如：

1. 最大仿真时间；
2. 平台被摧毁；
3. 目标丢失；
4. 油量或能量不足；
5. 达到任务成功条件；
6. 达到任务失败条件。

19.4 action 维度目前只有一个

当前动作只有：

heading_delta_deg

也就是只控制航向变化。

后续可以扩展为多维动作，例如：

heading_delta_deg
speed_delta
altitude_delta
weapon_command
sensor_mode

这样才能支持更复杂的空战、拦截或任务规划场景。

---

20. 后续接 PPO 的方向

当前代码已经可以作为 PPO 环境的基础。

后续大致可以这样改造：

20.1 将 observation 转成向量

当前 observation 是字典：

obs = {
    "range_m": ...,
    "relative_bearing_deg": ...,
    "elevation_deg": ...,
    "closing_speed": ...,
    "course_deg": ...
}

后续需要转成神经网络输入向量：

obs_vec = [
    range_m / 300000.0,
    relative_bearing_deg / 180.0,
    elevation_deg / 90.0,
    closing_speed / 1000.0,
    course_deg / 180.0
]

20.2 将策略替换为神经网络

当前是：

action = simple_policy(obs)

后续变成：

action = agent.select_action(obs_vec)

20.3 收集轨迹

PPO 需要收集：

state
action
reward
next_state
done
log_prob
value

当前环境已经能提供：

state
action
reward
next_state
done

只需要在智能体侧补充：

log_prob
value

即可。

20.4 批量 episode 训练

当前测试是：

num_episodes = 3

后续训练可以改成：

num_episodes = 1000

或者按照 PPO 的 rollout step 数来组织训练。

---

21. 小结

本文在上一篇“AFSim 与 Python 通过文件交互”的基础上，只修改 Python 文件，实现了由 Python 主动启动和管理 AFSim 进程的交互方式。

当前版本完成了以下功能：

1. Python 自动启动 AFSim；
2. Python 自动清理旧交互文件；
3. Python 等待 AFSim 写出 observation；
4. Python 根据 observation 计算 action；
5. Python 写入 action 文件；
6. AFSim 读取 action 并推进仿真；
7. Python 继续读取下一帧 observation；
8. Python 计算 reward 和 done；
9. Python 支持多 episode 运行；
10. 整体结构接近强化学习环境接口。

当前还没有真正接入 PPO，但已经形成了强化学习环境的基本雏形：

obs = env.reset()

while not done:
    action = policy(obs)
    obs, reward, done, info = env.step(action)

后续只需要将规则策略 simple_policy() 替换成神经网络策略，并完善 reward、done 和 observation 向量化，就可以继续向 PPO 强化学习训练推进。