【Python实现】CRC-CCITT —— 用socket演示

前言

这是我大三上学期的一个作品。然后为了能够显示中间过程，用面向过程的思想编写。

流程图

代码 CRC.py

# -*- coding:utf-8 -*-

import numpy as np


def np2str(npArray):
    """
    [0 1 1 0 0 0 0 1] -> '01100001'
    """
    S = ""
    for s in npArray:
        S += str(s)
    return S


def str2np(Str):
    """
    '01100001' -> [0 1 1 0 0 0 0 1]
    """
    return np.array([i for i in Str], dtype=np.uint8)


def m2data(m, test_mode=False):
    """
    消息转CRC码，字符串形式
    'a' -> '011000010111110010000111'
    'hello' -> '011010001110110110101110 011001010011110000000011 011011001010110100101010 011011001010110100101010 011011111001110101001001'
    """
    c = CRC_CCITT()

    # 消息 -> ASCII码 -> 二进制
    ASCII = np.frombuffer(m.encode('utf-8'), dtype=np.uint8)
    Bit = np.unpackbits(ASCII)

    # 分组
    l = len(Bit)
    N = 1
    while True:
        if l <= 8 * N:
            break
        N += 1
    Bit = Bit.reshape([N, 8])
    
    # 分别计算每组的CRC校验码
    data = ""
    for Bit_i in Bit:
        print("n信息码", Bit_i)
        CRC_code = c.get_CRC_code(Bit_i)
        
        # 测试模式，故意错一个位
        if test_mode:
            CRC_code[3] = 0 if CRC_code[3] == 1 else 1
            print("   错误的CRC码 ", CRC_code)

        data += ' ' + np2str(CRC_code)
    data = data[1:]

    return data


def data2m(data):
    c = CRC_CCITT()
    
    # CRC-CCITT
    crc = c.get_CRC()
    
    # 分组检查CRC码
    M = ""
    for d in data.split(' '):
        d = str2np(d)
        print(d)
        print("n检查校验码 计算过程")
        
        if c.check_code(d, crc):
            print('okn')
            correct = True
        else:
            print('errorn')
            correct = False

        M += chr(np.packbits(d[:8])[0])

    return M, correct


class CRC_CCITT:
    def get_CRC(self):

        """
        [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        """

        CRC = [0] * 17
        for i in [-16, -12, -5, 0]:
            CRC[i - 1] = 1
        CRC = np.array(CRC, dtype=np.uint8)
        return CRC


    def get_code(self, x, CRC):

        """
        由信息码生成CRC校验码
        信息码 [0 1 1 0 0 0 0 1]
        校验码                 [0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]
        """
        
        x1 = x[0:17]
        x2 = x[17:]
        i = 0
        while x1[i] != 1:
            i += 1

        tmp_x = np.append(x1[i:], x2[:i])

        if len(tmp_x) < 17:
            tmp_x = np.append(np.zeros(16 - len(tmp_x), dtype=np.uint8), tmp_x)
            return tmp_x

        x = np.append(np.bitwise_xor(tmp_x, CRC), x2[i:])
        
        print('t   ', tmp_x)
        print('tCRC', CRC)
        print('tXOR', x[:17], 'n')

        if len(x) > 17:
            return CRC_CCITT.get_code(self, x, CRC)
        else:
            return x[1:]
            
    def check_code(self, x, CRC):

        """
        检查CRC码，计算得[0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0]即正确，否则不正确
        """
            
        x1 = x[0:17]
        x2 = x[17:]
        i = 0
        while x1[i] != 1:
            i += 1

        tmp_x = np.append(x1[i:], x2[:i])

        if len(tmp_x) < 17:
            tmp_x = np.append(np.zeros(17 - len(tmp_x), dtype=np.uint8), tmp_x)

        x = np.append(np.bitwise_xor(tmp_x, CRC), x2[i:])

        print('t   ', tmp_x)
        print('tCRC', CRC)
        print('tXOR', x[:17], 'n')

        if 1 not in x:
            return True

        if len(x) > 17:
            return CRC_CCITT.check_code(self, x, CRC)
        else:
            if 1 in x:
                return False


    def get_CRC_code(self, Bit_i):

        """
        左移
        调用 get_code 函数生成CRC校验码
        组合信息码和CRC校验码

        信息码 [0 1 1 0 0 0 0 1]
        校验码                 [0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]
        CRC码  [0 1 1 0 0 0 0 1 0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]

        返回CRC码
        """

        Bit_left_16 = np.append(Bit_i, np.zeros(16, dtype=np.uint8))
        print("左移后", Bit_left_16)
        print("n生成校验码 计算过程")
        Code = CRC_CCITT.get_code(self, Bit_left_16, CRC_CCITT.get_CRC(self))
        print("t信息码", Bit_i)
        print("t校验码                ", Code)
        CRC_code = np.append(Bit_i, Code)
        print("tCRC码 ", CRC_code)
        return CRC_code

server.py

import socket
from CRC import data2m, m2data

HOST = ''
PORT = 10888
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((HOST, PORT))
s.listen(1)
conn, addr = s.accept()
print('Client's Address:', addr)
i = 1
while True:
    data = conn.recv(4096)
    if not data:
        print("n对方离开，关闭连接")
        conn.close()
        break
    else:
        print('n', '-' * 30, 'n')
        Data, correct = data2m(data.decode('utf-8'))
        if correct:
            print("接受的第 {} 条信息：{}".format(i, Data))
        else:
            print("信息错误")
        i += 1

client.py

import socket
from CRC import m2data

HOST = 'localhost'
PORT = 10888
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
i = 1
test_mode = eval(input("是否开启测试模式 True or False:"))
while True:
    print('n', '-' * 30, 'n')
    print('输入要发送的信息（第 %d 条）:'% i, end='')
    data = input()
    print(data)
    if data == 'exit':
        s.close()
        break
    i += 1
    
    Data = m2data(data, test_mode)
    s.sendall(Data.encode('utf-8'))

演示1 【接收正确】

在存放代码的文件夹打开两个终端，按顺序执行以下命令
第一个终端（执行程序后变成socket服务端，以下简称服务端）

python server.py

第二个终端（执行程序后变成socket客户端，以下简称客户端）

python client.py

客户端提示是否开启测试模式 True or False:
输入 False
客户端提示输入要发送的信息（第 1 条）:
输入 a

以下是上面这一过程显示的内容
客户端

是否开启测试模式 True or False:False

 ------------------------------

输入要发送的信息（第 1 条）:a
a

信息码 [0 1 1 0 0 0 0 1]
左移后 [0 1 1 0 0 0 0 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0]

生成校验码 计算过程
            [1 1 0 0 0 0 1 0 0 0 0 0 0 0 0 0 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 0 0 1 0 1 0 0 0 0 1 0 0 0 0 1]

            [1 0 0 1 0 1 0 0 0 0 1 0 0 0 0 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 0 0 1 1 1 0 0 0 0 1 1 0 0 0 1 1]

            [1 1 1 0 0 0 0 1 1 0 0 0 1 1 0 0 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 1 0 1 0 0 1 1 0 0 1 1 1 0 0 1]

            [1 1 0 1 0 0 1 1 0 0 1 1 1 0 0 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 0 1 1 0 1 1 0 0 1 0 1 0 0 1 1]

            [1 0 1 1 0 1 1 0 0 1 0 1 0 0 1 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]

        信息码 [0 1 1 0 0 0 0 1]
        校验码                 [0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]
        CRC码  [0 1 1 0 0 0 0 1 0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]

 ------------------------------

服务端

Client's Address: ('127.0.0.1', 6691)

 ------------------------------

[0 1 1 0 0 0 0 1 0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]

检查校验码 计算过程
            [1 1 0 0 0 0 1 0 1 1 1 1 1 0 0 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 0 0 1 0 1 0 1 1 1 0 1 0 0 1 1]

            [1 0 0 1 0 1 0 1 1 1 0 1 0 0 1 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 0 0 1 1 1 0 1 1 1 0 0 0 0 1 1 1]

            [1 1 1 0 1 1 1 0 0 0 0 1 1 1 0 0 1]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 1 0 0 1 1 0 0 0 0 0 1 1 0 0 0]

            [1 1 0 0 1 1 0 0 0 0 0 1 1 0 0 0 1]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0]

            [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0]

ok

接受的第 1 条信息：a

演示结束
客户端输入 exit 离开

演示2 【接收错误】

流程和演示1差不多，只不过开启了测试模式

客户端提示是否开启测试模式 True or False:
输入 True
客户端提示输入要发送的信息（第 1 条）:
输入 a

以下是上面这一过程显示的内容
客户端

是否开启测试模式 True or False:True

 ------------------------------

输入要发送的信息（第 1 条）:a
a

信息码 [0 1 1 0 0 0 0 1]
左移后 [0 1 1 0 0 0 0 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0]

生成校验码 计算过程
            [1 1 0 0 0 0 1 0 0 0 0 0 0 0 0 0 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 0 0 1 0 1 0 0 0 0 1 0 0 0 0 1]

            [1 0 0 1 0 1 0 0 0 0 1 0 0 0 0 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 0 0 1 1 1 0 0 0 0 1 1 0 0 0 1 1]

            [1 1 1 0 0 0 0 1 1 0 0 0 1 1 0 0 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 1 0 1 0 0 1 1 0 0 1 1 1 0 0 1]

            [1 1 0 1 0 0 1 1 0 0 1 1 1 0 0 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 0 1 1 0 1 1 0 0 1 0 1 0 0 1 1]

            [1 0 1 1 0 1 1 0 0 1 0 1 0 0 1 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]

        信息码 [0 1 1 0 0 0 0 1]
        校验码                 [0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]
        CRC码  [0 1 1 0 0 0 0 1 0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]
   错误的CRC码  [0 1 1 1 0 0 0 1 0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]

 ------------------------------

输入要发送的信息（第 2 条）:exit
exit

服务端

Client's Address: ('127.0.0.1', 7882)

 ------------------------------

[0 1 1 1 0 0 0 1 0 1 1 1 1 1 0 0 1 0 0 0 0 1 1 1]

检查校验码 计算过程
            [1 1 1 0 0 0 1 0 1 1 1 1 1 0 0 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 1 0 1 0 1 0 1 1 1 0 1 0 0 1 1]

            [1 1 0 1 0 1 0 1 1 1 0 1 0 0 1 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 0 1 1 1 0 1 1 1 0 0 0 0 1 1 1]

            [1 0 1 1 1 0 1 1 1 0 0 0 0 1 1 1 0]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 0 1 1 0 0 1 1 1 0 0 1 0 1 1 1 1]

            [1 1 0 0 1 1 1 0 0 1 0 1 1 1 1 0 1]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 1 0 0 0 1 1 0 0 1 0 0 1 1 1 0 0]

            [1 0 0 0 1 1 0 0 1 0 0 1 1 1 0 0 1]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [0 0 0 0 0 1 0 0 1 0 0 0 1 1 0 0 0]

            [0 0 0 0 1 0 0 1 0 0 0 1 1 0 0 0 1]
        CRC [1 0 0 0 1 0 0 0 0 0 0 1 0 0 0 0 1]
        XOR [1 0 0 0 0 0 0 1 0 0 0 0 1 0 0 0 0]

error

信息错误

对方离开，关闭连接

【Python实现】CRC-CCITT —— 用socket演示

Python相关栏目本月热门文章