智能制造系统设计与代码实现

智能制造系统设计与代码实现pythonfrominfluxdbimportInfluxDBClientimportlogginglogging.basicConfig(level=logging.INFO)logger=logging.getLogger('InfluxDBWriter')classInfluxDBWriter:def__init__(self,host='localhost',port=8086,username='admin',password='admin',database='smart_factory'):self.host=hostself.port=portself.username=usernameself.password=passwordself.database=databaseself.client=Noneself.connect()defconnect(self):"""连接到InfluxDB并创建数据库（如果不存在）"""try:self.client=InfluxDBClient(host=self.host,port=self.port,username=self.username,password=self.password,database=self.database)检查数据库是否存在，不存在则创建databases=self.client.get_list_database()db_names=[db['name']fordbindatabases]ifself.databasenotindb_names:self.client.create_database(self.database)(f"创建数据库:{self.database}")(f"成功连接到InfluxDB:{self.host}:{self.port},数据库:{self.database}")exceptExceptionase:logger.error(f"连接InfluxDB失败:{e}")self.client=Nonedefwrite_data(self,data_point):"""将数据点写入InfluxDB"""ifnotself.client:logger.error("InfluxDB客户端未连接，无法写入数据")returnFalsetry:构造InfluxDB数据点格式measurement:通常对应设备类型或数据大类，例如"machine_data"tags:设备ID、节点名称等用于查询和分组的键值对fields:具体的测量值，数值型time:时间戳(可选，InfluxDB会自动添加，这里使用采集时的时间戳)json_body=[{"measurement":"machine_data","tags":{"device":data_point['device'],"node_name":data_point['node_name'],"node_id":data_point['node_id']},"time":data_point['timestamp'],"fields":{"value":float(data_point['valu