在无数条错误路径中寻找真相

11 10 00 01 01 11 00

11 10 01 01 01 11 00

11 10 00 01 01 11 00

11 10 00 01 01 11 00

def encode_convolutional(bits, terminate=True):
    """
    速率1/2、约束长度K=3的卷积编码器。

    生成多项式：(7, 5) 八进制

    状态包含前两个输入比特：
        (m1, m2)

    输出：
        v1 = u XOR m1 XOR m2
        v2 = u XOR m2
    """

    if set(bits) - {"0", "1"}:
        raise ValueError("输入必须是二进制字符串")

    # 尾部添加两个终止位
    source = bits + ("00" if terminate else "")

    m1, m2 = 0, 0
    output = []

    for char in source:
        u = int(char)

        v1 = u ^ m1 ^ m2
        v2 = u ^ m2

        output.append(f"{v1}{v2}")

        m1, m2 = u, m1

    return output


message = "10110"

encoded = encode_convolutional(message)

print("原始消息:", message)
print("编码序列:", " ".join(encoded))

assert encoded == [
    "11", "10", "00", "01",
    "01", "11", "00",
]

原始消息: 10110
编码序列: 11 10 00 01 01 11 00

STATES = ("00", "01", "10", "11")


def transition(state, input_bit):
    """
    state: 两字符状态，例如"10"
    input_bit: 整数0或1

    返回：
        (next_state, encoded_pair)
    """

    m1 = int(state[0])
    m2 = int(state[1])

    u = input_bit

    v1 = u ^ m1 ^ m2
    v2 = u ^ m2

    next_state = f"{u}{m1}"
    output_pair = f"{v1}{v2}"

    return next_state, output_pair


for state in STATES:
    print(f"\n当前状态: {state}")

    for bit in (0, 1):
        nxt, out = transition(state, bit)

        print(
            f"输入{bit} -> 输出{out} -> 新状态{nxt}"
        )

def hamming_distance(a, b):
    if len(a) != len(b):
        raise ValueError("长度不一致")

    return sum(x != y for x, y in zip(a, b))


def viterbi_decode(received, terminated=True):
    """
    维特比硬判决译码。

    参数：
        received: 编码符号对列表，例如
            ["11", "10", "01", ...]
        terminated:
            True  -> 约定终止状态为00
            False -> 从最终可达状态中选最佳者

    返回：
        解码输入序列、最小累计距离、最佳路径状态序列
    """

    INF = float("inf")

    metrics = {
        state: INF
        for state in STATES
    }

    # 编码器起始状态固定为00
    metrics["00"] = 0

    backpointers = []

    for observed in received:
        if len(observed) != 2:
            raise ValueError("每组观测必须为两个比特")

        if set(observed) - {"0", "1"}:
            raise ValueError("观测必须为二进制")

        next_metrics = {
            state: INF
            for state in STATES
        }

        parents = {}

        for state in STATES:
            if metrics[state] == INF:
                continue

            for input_bit in (0, 1):
                next_state, expected = transition(
                    state, input_bit
                )

                branch_cost = hamming_distance(
                    expected, observed
                )

                candidate = (
                    metrics[state] + branch_cost
                )

                if candidate < next_metrics[next_state]:
                    next_metrics[next_state] = candidate

                    parents[next_state] = (
                        state, input_bit
                    )

        metrics = next_metrics
        backpointers.append(parents)

    if terminated:
        state = "00"
    else:
        state = min(
            STATES,
            key=lambda s: metrics[s]
        )

    if metrics[state] == INF:
        raise ValueError("没有合法的终止路径")

    best_cost = metrics[state]

    decoded_reversed = []
    states_reversed = [state]

    # 从终点沿最佳前驱逐步回溯
    for t in range(len(received) - 1, -1, -1):
        previous, input_bit = backpointers[t][state]

        decoded_reversed.append(str(input_bit))

        state = previous
        states_reversed.append(state)

    decoded_full = "".join(
        reversed(decoded_reversed)
    )

    state_path = list(
        reversed(states_reversed)
    )

    return {
        "decoded_full": decoded_full,
        "decoded_data": (
            decoded_full[:-2]
            if terminated
            else decoded_full
        ),
        "cost": best_cost,
        "states": state_path,
    }


# 一次比特翻转
received = [
    "11", "10", "01",
    "01", "01", "11", "00",
]

result = viterbi_decode(
    received,
    terminated=True
)

print("接收序列:", " ".join(received))
print("恢复输入:", result["decoded_full"])
print("原始消息:", result["decoded_data"])
print("最小距离:", result["cost"])
print("状态路径:", " -> ".join(result["states"]))

assert result["decoded_data"] == "10110"
assert result["cost"] == 1

接收序列: 11 10 01 01 01 11 00
恢复输入: 1011000
原始消息: 10110
最小距离: 1
状态路径: 00 -> 10 -> 01 -> 10 -> 11 -> 01 -> 00 -> 00

from itertools import product


def brute_force_decode(received):
    """
    只用于短消息验证。

    假设最后两个输入为固定的00终止比特。
    """

    data_length = len(received) - 2

    best_message = None
    best_cost = float("inf")

    for bits in product("01", repeat=data_length):
        message = "".join(bits)

        expected = encode_convolutional(
            message,
            terminate=True
        )

        cost = sum(
            hamming_distance(a, b)
            for a, b in zip(expected, received)
        )

        if cost < best_cost:
            best_cost = cost
            best_message = message

    return best_message, best_cost


received = [
    "11", "10", "01",
    "01", "01", "11", "00",
]

dp = viterbi_decode(received)
bf_message, bf_cost = brute_force_decode(received)

print("维特比:", dp["decoded_data"], dp["cost"])
print("暴力枚举:", bf_message, bf_cost)

assert dp["cost"] == bf_cost
assert dp["decoded_data"] == bf_message

print("两种方法结果一致")

清晰 模糊 清晰 清晰 模糊

import math


STATES = ("稳定", "干扰")

# 初始状态概率
START_PROB = {
    "稳定": 0.7,
    "干扰": 0.3,
}

# 状态转移概率
TRANSITION = {
    "稳定": {
        "稳定": 0.85,
        "干扰": 0.15,
    },
    "干扰": {
        "稳定": 0.25,
        "干扰": 0.75,
    },
}

# 观测概率
EMISSION = {
    "稳定": {
        "清晰": 0.9,
        "模糊": 0.1,
    },
    "干扰": {
        "清晰": 0.3,
        "模糊": 0.7,
    },
}


def viterbi_hmm(observations):
    """
    HMM最大联合概率状态路径推断。

    使用对数概率，避免长序列下溢。
    """

    if not observations:
        return []

    # 第一个时刻的最佳路径分数
    scores = {
        state: (
            math.log(START_PROB[state])
            + math.log(EMISSION[state][observations[0]])
        )
        for state in STATES
    }

    backpointers = []

    # 从第二个观测开始递推
    for observation in observations[1:]:
        next_scores = {}
        parent = {}

        for current in STATES:
            candidates = {}

            for previous in STATES:
                candidates[previous] = (
                    scores[previous]
                    + math.log(
                        TRANSITION[previous][current]
                    )
                    + math.log(
                        EMISSION[current][observation]
                    )
                )

            best_previous = max(
                candidates,
                key=candidates.get
            )

            next_scores[current] = (
                candidates[best_previous]
            )

            parent[current] = best_previous

        scores = next_scores
        backpointers.append(parent)

    # 选择最终最可能状态
    final_state = max(
        scores,
        key=scores.get
    )

    # 从末尾逐步回溯
    path = [final_state]

    for parent in reversed(backpointers):
        final_state = parent[final_state]
        path.append(final_state)

    path.reverse()

    return path


observations = [
    "清晰",
    "模糊",
    "清晰",
    "清晰",
    "模糊",
]

best_path = viterbi_hmm(observations)

print("观测序列:", " → ".join(observations))
print("最优状态:", " → ".join(best_path))

assert best_path == [
    "稳定",
    "稳定",
    "稳定",
    "稳定",
    "干扰",
]

观测序列: 清晰 → 模糊 → 清晰 → 清晰 → 模糊
最优状态: 稳定 → 稳定 → 稳定 → 稳定 → 干扰

candidate = metrics[state] + branch_cost

candidate = previous_score + log_probability

math.log(p)

def log_probability(p):
    return math.log(p) if p > 0 else float("-inf")