车间数字化MES项目-数据抓取与处理

由于设备端上传的文件为文本形式，因此，需要首先对文本中的信息进行处理，将其存入数据库中。

思路：python定时读取固定文件夹下的文本，进行解析，存入数据库
主要解决的问题：该部分代码运行独立于django框架，但是又需要引用django中的models.py进行数据库存储

代码实现：

# 模块引入
import os
import time
import pandas as pd
import threading

# 报警文本读取
def alarm_read():
    path = "A:\\ftptest\\AF200\\GM4242_02_06.adb"  # 设置路径
    data = pd.read_table(path, encoding='gbk', header=None)
    # 获取分列
    data = data[0].str.split('|', expand=True)
    # 获得读取的文本行数
    data_number = data.shape[0]
    # 获取数据库的行数
    data_sql = models.alarm.objects.count()
    # 判断是否有数据更新
    number_add = data_number - data_sql
    print(data_sql, data_number, number_add)
    # 如果有数据更新，则更新数据存入数据库
    if number_add:
        # 数据行与列切片，得到更新的报警数据
        data = data.iloc[-number_add:, [0, 1, 2, 5]]
        data.columns = ['alarm_code', 'alarm_start', 'alarm_end', 'alarm_content']
        # 转换时间戳
        new_time = []
        new_time1 = []
        for i, j in zip(data['alarm_start'].values, data['alarm_end'].values):
            timeArray, timeArray1 = time.localtime(int(i)), time.localtime(int(j))
            otherStyleTime, otherStyleTime1 = time.strftime("%Y-%m-%d %H:%M:%S", timeArray), time.strftime(
                "%Y-%m-%d %H:%M:%S", timeArray1)
            new_time.append(otherStyleTime)
            new_time1.append(otherStyleTime1)
        data["alarm_start"] = new_time
        data["alarm_end"] = new_time1
        for index, row in data.iterrows():
            # print(row['alarm_start'])
            models.alarm.objects.create(alarm_code=row['alarm_code'][1:], alarm_start=row['alarm_start'],
                                        alarm_end=row['alarm_end'], alarm_content=row['alarm_content'])
        return number_add
    else:
        print("无数据更新")

在这个模块里用到了django的models，需要引入，注意其需要在main函数中引入

if __name__ == '__main__':
    # 加载Django环境，mainproject是我的Django项目名称
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", 'mainproject.settings')
    # 引入Django模块
    import django
    django.setup()
    import sqlite3
    from firstweb import models
    # 初始化Django环境
    time_read()

time_read函数定时执行alarm_read函数

class CallingCounter(object):
    def __init__(self, func):
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        return self.func(*args, **kwargs)


@CallingCounter
def time_read():
    # 定义全局变量
    global timer
    # 5秒调用函数一次,递归调用
    alarm_read()
    timer = threading.Timer(10, time_read)
    # 启用定时器
    timer.start()
    # 执行次数
    print('执行次数', time_read.count)

结果：

image.png