PLC通讯协议选型与上位机开发（下）：Python/C#实战开发指南
·
前言
上一篇我们对比了四种主流PLC通讯协议的优缺点和适用场景。这一篇进入实战环节：
- 如何用Python/C#开发PLC上位机
- 数据采集、实时监控、报警管理的实现
- 性能优化和异常处理技巧
项目背景：
- 开发工具：Python 3.10+ / C# .NET 6+
- 通讯协议：Modbus TCP、OPC UA
- 应用场景：设备监控、数据采集、MES对接
一、Python上位机开发
1.1 项目结构
code复制
plc_monitor/
├── config/
│ └── plc_config.yaml # PLC配置文件
├── core/
│ ├── __init__.py
│ ├── modbus_client.py # Modbus通讯模块
│ ├── opcua_client.py # OPC UA通讯模块
│ └── data_manager.py # 数据管理模块
├── ui/
│ ├── __init__.py
│ ├── main_window.py # 主窗口
│ ├── dashboard.py # 仪表盘
│ └── alarm_panel.py # 报警面板
├── storage/
│ └── database.py # 数据库操作
├── main.py # 程序入口
└── requirements.txt # 依赖包
1.2 配置文件
yaml复制
# plc_config.yaml
plc:
name: "1号生产线PLC"
protocol: "modbus_tcp" # modbus_tcp 或 opcua
ip: "192.168.1.100"
port: 502
timeout: 3 # 超时时间（秒）
retry: 3 # 重试次数
# 数据点配置
points:
- name: "温度"
address: 100
type: "float" # float/int/bool
unit: "℃"
alarm_high: 80.0
alarm_low: 10.0
- name: "压力"
address: 102
type: "float"
unit: "MPa"
alarm_high: 1.0
- name: "运行状态"
address: 200
type: "bool"
description: "设备运行中"
# 采集配置
sampling:
interval: 1000 # 采集周期（毫秒）
history_days: 30 # 历史数据保留天数
1.3 Modbus通讯模块
python复制
# core/modbus_client.py
from pymodbus.client import ModbusTcpClient
from pymodbus.payload import BinaryPayloadDecoder
from pymodbus.constants import Endian
import threading
import time
import logging
logger = logging.getLogger(__name__)
class ModbusClient:
def __init__(self, config):
self.config = config
self.client = None
self.lock = threading.Lock()
self.connected = False
def connect(self):
"""建立连接"""
try:
self.client = ModbusTcpClient(
self.config['ip'],
port=self.config['port'],
timeout=self.config['timeout']
)
if self.client.connect():
self.connected = True
logger.info(f"连接成功: {self.config['ip']}")
return True
except Exception as e:
logger.error(f"连接失败: {e}")
return False
def read_float(self, address):
"""读取浮点数"""
with self.lock:
try:
result = self.client.read_holding_registers(
address=address, count=2
)
if result.isError():
raise Exception(f"读取失败: {result}")
decoder = BinaryPayloadDecoder.fromRegisters(
result.registers,
byteorder=Endian.BIG,
wordorder=Endian.BIG
)
return decoder.decode_32bit_float()
except Exception as e:
logger.error(f"读取浮点数失败: {e}")
self._reconnect()
return None
def read_bool(self, address, bit=0):
"""读取布尔值"""
with self.lock:
try:
result = self.client.read_coils(
address=address, count=1
)
if result.isError():
raise Exception(f"读取失败: {result}")
return result.bits[bit]
except Exception as e:
logger.error(f"读取布尔值失败: {e}")
self._reconnect()
return None
def write_float(self, address, value):
"""写入浮点数"""
with self.lock:
try:
from pymodbus.payload import BinaryPayloadBuilder
builder = BinaryPayloadBuilder(
byteorder=Endian.BIG,
wordorder=Endian.BIG
)
builder.add_32bit_float(value)
payload = builder.to_registers()
result = self.client.write_registers(
address=address,
values=payload
)
return not result.isError()
except Exception as e:
logger.error(f"写入浮点数失败: {e}")
self._reconnect()
return False
def _reconnect(self):
"""自动重连"""
self.connected = False
for i in range(self.config['retry']):
logger.info(f"尝试重连 ({i+1}/{self.config['retry']})")
if self.connect():
return True
time.sleep(1)
return False
def disconnect(self):
"""断开连接"""
if self.client:
self.client.close()
self.connected = False
1.4 数据采集与存储
python复制
# core/data_manager.py
import time
import threading
import sqlite3
from datetime import datetime, timedelta
class DataManager:
def __init__(self, client, config):
self.client = client
self.config = config
self.data_cache = {}
self.running = False
self.init_database()
def init_database(self):
"""初始化数据库"""
self.conn = sqlite3.connect('plc_data.db', check_same_thread=False)
self.conn.execute('''
CREATE TABLE IF NOT EXISTS history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
point_name TEXT NOT NULL,
value REAL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
self.conn.execute('''
CREATE TABLE IF NOT EXISTS alarms (
id INTEGER PRIMARY KEY AUTOINCREMENT,
point_name TEXT NOT NULL,
alarm_type TEXT,
value REAL,
threshold REAL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
acknowledged BOOLEAN DEFAULT 0
)
''')
self.conn.commit()
def start_sampling(self):
"""开始数据采集"""
self.running = True
self.sampling_thread = threading.Thread(target=self._sampling_loop)
self.sampling_thread.daemon = True
self.sampling_thread.start()
def _sampling_loop(self):
"""采集循环"""
interval = self.config['sampling']['interval'] / 1000.0
while self.running:
for point in self.config['points']:
value = self._read_point(point)
if value is not None:
self.data_cache[point['name']] = value
self._save_history(point['name'], value)
self._check_alarm(point, value)
time.sleep(interval)
def _read_point(self, point):
"""读取单个数据点"""
if point['type'] == 'float':
return self.client.read_float(point['address'])
elif point['type'] == 'bool':
return self.client.read_bool(point['address'])
return None
def _save_history(self, name, value):
"""保存历史数据"""
try:
self.conn.execute(
'INSERT INTO history (point_name, value) VALUES (?, ?)',
(name, value)
)
self.conn.commit()
except Exception as e:
print(f"保存历史数据失败: {e}")
def _check_alarm(self, point, value):
"""检查报警"""
if value is None:
return
alarms = []
if 'alarm_high' in point and value > point['alarm_high']:
alarms.append(('HIGH', point['alarm_high']))
if 'alarm_low' in point and value < point['alarm_low']:
alarms.append(('LOW', point['alarm_low']))
for alarm_type, threshold in alarms:
self._trigger_alarm(point['name'], alarm_type, value, threshold)
def _trigger_alarm(self, name, alarm_type, value, threshold):
"""触发报警"""
try:
self.conn.execute(
'''INSERT INTO alarms (point_name, alarm_type, value, threshold)
VALUES (?, ?, ?, ?)''',
(name, alarm_type, value, threshold)
)
self.conn.commit()
print(f"⚠️ 报警: {name} {alarm_type} = {value} (阈值: {threshold})")
except Exception as e:
print(f"保存报警失败: {e}")
def get_history(self, point_name, hours=24):
"""查询历史数据"""
since = datetime.now() - timedelta(hours=hours)
cursor = self.conn.execute(
'SELECT value, timestamp FROM history '
'WHERE point_name = ? AND timestamp > ? '
'ORDER BY timestamp',
(point_name, since)
)
return cursor.fetchall()
def get_active_alarms(self):
"""获取未确认的报警"""
cursor = self.conn.execute(
'SELECT * FROM alarms WHERE acknowledged = 0 '
'ORDER BY timestamp DESC'
)
return cursor.fetchall()
def acknowledge_alarm(self, alarm_id):
"""确认报警"""
self.conn.execute(
'UPDATE alarms SET acknowledged = 1 WHERE id = ?',
(alarm_id,)
)
self.conn.commit()
def cleanup_old_data(self):
"""清理过期数据"""
days = self.config['sampling']['history_days']
since = datetime.now() - timedelta(days=days)
self.conn.execute(
'DELETE FROM history WHERE timestamp < ?', (since,)
)
self.conn.commit()
def stop(self):
"""停止采集"""
self.running = False
self.conn.close()
1.5 简单的命令行监控界面
python复制
# main.py
import yaml
import time
from core.modbus_client import ModbusClient
from core.data_manager import DataManager
def main():
# 加载配置
with open('config/plc_config.yaml', 'r') as f:
config = yaml.safe_load(f)
# 初始化客户端
client = ModbusClient(config['plc'])
if not client.connect():
print("无法连接PLC，退出程序")
return
# 初始化数据管理器
manager = DataManager(client, config)
manager.start_sampling()
print("PLC监控系统已启动")
print("按 Ctrl+C 停止")
print("-" * 50)
try:
while True:
# 显示实时数据
print("\033[2J\033[H") # 清屏
print("=" * 50)
print("PLC实时监控")
print("=" * 50)
for point in config['points']:
value = manager.data_cache.get(point['name'], 'N/A')
unit = point.get('unit', '')
print(f"{point['name']}: {value} {unit}")
# 显示报警
alarms = manager.get_active_alarms()
if alarms:
print("\n" + "=" * 50)
print("⚠️ 活跃报警:")
for alarm in alarms:
print(f" {alarm[1]} {alarm[2]} = {alarm[3]} (阈值: {alarm[4]})")
time.sleep(1)
except KeyboardInterrupt:
print("\n正在停止...")
manager.stop()
client.disconnect()
print("已停止")
if __name__ == '__main__':
main()
二、C#上位机开发
2.1 项目结构
code复制
PLCMonitor/
├── Models/
│ ├── PlcConfig.cs # PLC配置模型
│ ├── DataPoint.cs # 数据点模型
│ └── AlarmRecord.cs # 报警记录模型
├── Services/
│ ├── IPlcClient.cs # PLC客户端接口
│ ├── ModbusTcpClient.cs # Modbus实现
│ ├── OpcUaClient.cs # OPC UA实现
│ └── DataService.cs # 数据服务
├── ViewModels/
│ ├── MainViewModel.cs # 主窗口ViewModel
│ ├── DashboardViewModel.cs # 仪表盘ViewModel
│ └── AlarmViewModel.cs # 报警ViewModel
├── Views/
│ ├── MainWindow.xaml # 主窗口
│ ├── Dashboard.xaml # 仪表盘
│ └── AlarmPanel.xaml # 报警面板
├── Database/
│ └── DatabaseHelper.cs # 数据库操作
└── Program.cs # 程序入口
2.2 PLC客户端接口
csharp复制
// Services/IPlcClient.cs
namespace PLCMonitor.Services
{
public interface IPlcClient : IDisposable
{
bool IsConnected { get; }
event EventHandler<ConnectionChangedEventArgs> ConnectionChanged;
Task<bool> ConnectAsync();
Task DisconnectAsync();
Task<float> ReadFloatAsync(int address);
Task<bool> ReadBoolAsync(int address, int bit = 0);
Task<bool> WriteFloatAsync(int address, float value);
Task<Dictionary<int, float>> ReadMultipleFloatsAsync(
int startAddress, int count);
}
public class ConnectionChangedEventArgs : EventArgs
{
public bool IsConnected { get; set; }
public string Message { get; set; }
}
}
2.3 Modbus TCP实现
csharp复制
// Services/ModbusTcpClient.cs
using NModbus;
using System.Net.Sockets;
namespace PLCMonitor.Services
{
public class ModbusTcpClient : IPlcClient
{
private readonly string _ip;
private readonly int _port;
private TcpClient _tcpClient;
private IModbusMaster _master;
private readonly byte _slaveId = 1;
public bool IsConnected => _tcpClient?.Connected ?? false;
public event EventHandler<ConnectionChangedEventArgs> ConnectionChanged;
public ModbusTcpClient(string ip, int port = 502)
{
_ip = ip;
_port = port;
}
public async Task<bool> ConnectAsync()
{
try
{
_tcpClient = new TcpClient();
await _tcpClient.ConnectAsync(_ip, _port);
var factory = new ModbusFactory();
_master = factory.CreateMaster(_tcpClient);
_master.Transport.ReadTimeout = 3000;
_master.Transport.WriteTimeout = 3000;
ConnectionChanged?.Invoke(this,
new ConnectionChangedEventArgs
{
IsConnected = true,
Message = "连接成功"
});
return true;
}
catch (Exception ex)
{
ConnectionChanged?.Invoke(this,
new ConnectionChangedEventArgs
{
IsConnected = false,
Message = $"连接失败: {ex.Message}"
});
return false;
}
}
public async Task<float> ReadFloatAsync(int address)
{
try
{
ushort[] registers = await _master.ReadHoldingRegistersAsync(
_slaveId, (ushort)address, 2);
// 大端序解析
byte[] bytes = new byte[4];
bytes[0] = (byte)(registers[1] >> 8);
bytes[1] = (byte)(registers[1] & 0xFF);
bytes[2] = (byte)(registers[0] >> 8);
bytes[3] = (byte)(registers[0] & 0xFF);
return BitConverter.ToSingle(bytes, 0);
}
catch (Exception ex)
{
throw new PlcReadException($"读取浮点数失败: {ex.Message}", ex);
}
}
public async Task<bool> ReadBoolAsync(int address, int bit = 0)
{
try
{
bool[] results = await _master.ReadCoilsAsync(
_slaveId, (ushort)address, 1);
return results[bit];
}
catch (Exception ex)
{
throw new PlcReadException($"读取布尔值失败: {ex.Message}", ex);
}
}
public async Task<bool> WriteFloatAsync(int address, float value)
{
try
{
byte[] bytes = BitConverter.GetBytes(value);
ushort[] registers = new ushort[2];
// 大端序
registers[0] = (ushort)((bytes[2] << 8) | bytes[3]);
registers[1] = (ushort)((bytes[0] << 8) | bytes[1]);
await _master.WriteMultipleRegistersAsync(
_slaveId, (ushort)address, registers);
return true;
}
catch (Exception ex)
{
throw new PlcWriteException($"写入浮点数失败: {ex.Message}", ex);
}
}
public async Task<Dictionary<int, float>> ReadMultipleFloatsAsync(
int startAddress, int count)
{
var result = new Dictionary<int, float>();
try
{
ushort[] registers = await _master.ReadHoldingRegistersAsync(
_slaveId, (ushort)startAddress, (ushort)(count * 2));
for (int i = 0; i < count; i++)
{
byte[] bytes = new byte[4];
bytes[0] = (byte)(registers[i * 2 + 1] >> 8);
bytes[1] = (byte)(registers[i * 2 + 1] & 0xFF);
bytes[2] = (byte)(registers[i * 2] >> 8);
bytes[3] = (byte)(registers[i * 2] & 0xFF);
result[startAddress + i * 2] =
BitConverter.ToSingle(bytes, 0);
}
}
catch (Exception ex)
{
throw new PlcReadException($"批量读取失败: {ex.Message}", ex);
}
return result;
}
public async Task DisconnectAsync()
{
_tcpClient?.Close();
_tcpClient?.Dispose();
_tcpClient = null;
_master = null;
ConnectionChanged?.Invoke(this,
new ConnectionChangedEventArgs
{
IsConnected = false,
Message = "已断开"
});
}
public void Dispose()
{
DisconnectAsync().Wait();
}
}
public class PlcReadException : Exception
{
public PlcReadException(string message, Exception inner)
: base(message, inner) { }
}
public class PlcWriteException : Exception
{
public PlcWriteException(string message, Exception inner)
: base(message, inner) { }
}
}
2.4 WPF数据绑定
csharp复制
// ViewModels/MainViewModel.cs
using CommunityToolkit.Mvvm.ComponentModel;
using CommunityToolkit.Mvvm.Input;
using PLCMonitor.Services;
using System.Collections.ObjectModel;
namespace PLCMonitor.ViewModels
{
public partial class MainViewModel : ObservableObject
{
private readonly IPlcClient _plcClient;
private readonly DataService _dataService;
private DispatcherTimer _timer;
[ObservableProperty]
private bool _isConnected;
[ObservableProperty]
private string _connectionStatus = "未连接";
public ObservableCollection<DataPointViewModel> DataPoints { get; }
public ObservableCollection<AlarmViewModel> Alarms { get; }
public MainViewModel(IPlcClient plcClient, DataService dataService)
{
_plcClient = plcClient;
_dataService = dataService;
DataPoints = new ObservableCollection<DataPointViewModel>();
Alarms = new ObservableCollection<AlarmViewModel>();
_plcClient.ConnectionChanged += OnConnectionChanged;
// 定时刷新
_timer = new DispatcherTimer { Interval = TimeSpan.FromSeconds(1) };
_timer.Tick += async (s, e) => await RefreshDataAsync();
}
[RelayCommand]
private async Task ConnectAsync()
{
if (IsConnected)
{
await _plcClient.DisconnectAsync();
}
else
{
await _plcClient.ConnectAsync();
_timer.Start();
}
}
private async Task RefreshDataAsync()
{
if (!IsConnected) return;
foreach (var point in DataPoints)
{
try
{
if (point.Type == "float")
{
point.Value = await _plcClient.ReadFloatAsync(point.Address);
}
else if (point.Type == "bool")
{
point.Value = await _plcClient.ReadBoolAsync(point.Address);
}
point.Status = "正常";
}
catch (Exception ex)
{
point.Status = $"错误: {ex.Message}";
}
}
}
private void OnConnectionChanged(object sender, ConnectionChangedEventArgs e)
{
IsConnected = e.IsConnected;
ConnectionStatus = e.Message;
}
}
}
2.5 XAML界面
xml复制
<!-- Views/MainWindow.xaml -->
<Window x:Class="PLCMonitor.Views.MainWindow"
xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
Title="PLC监控系统" Height="600" Width="800">
<Grid>
<Grid.RowDefinitions>
<RowDefinition Height="Auto"/>
<RowDefinition Height="*"/>
<RowDefinition Height="Auto"/>
</Grid.RowDefinitions>
<!-- 顶部状态栏 -->
<Border Grid.Row="0" Background="#2196F3" Padding="10">
<StackPanel Orientation="Horizontal">
<TextBlock Text="PLC监控系统"
Foreground="White"
FontSize="20"
FontWeight="Bold"/>
<TextBlock Text="{Binding ConnectionStatus}"
Foreground="White"
Margin="20,0,0,0"
VerticalAlignment="Center"/>
<Button Content="{Binding IsConnected,
Converter={StaticResource BoolToConnectText}}"
Command="{Binding ConnectCommand}"
Margin="20,0,0,0"
Padding="10,5"/>
</StackPanel>
</Border>
<!-- 数据表格 -->
<DataGrid Grid.Row="1"
ItemsSource="{Binding DataPoints}"
AutoGenerateColumns="False"
IsReadOnly="True"
AlternatingRowBackground="#F5F5F5">
<DataGrid.Columns>
<DataGridTextColumn Header="名称"
Binding="{Binding Name}"
Width="120"/>
<DataGridTextColumn Header="地址"
Binding="{Binding Address}"
Width="80"/>
<DataGridTextColumn Header="值"
Binding="{Binding DisplayValue}"
Width="100"/>
<DataGridTextColumn Header="单位"
Binding="{Binding Unit}"
Width="60"/>
<DataGridTextColumn Header="状态"
Binding="{Binding Status}"
Width="120"/>
</DataGrid.Columns>
</DataGrid>
<!-- 底部报警栏 -->
<Border Grid.Row="2"
Background="#FFEB3B"
Padding="10"
Visibility="{Binding Alarms.Count,
Converter={StaticResource CountToVisibility}}">
<StackPanel>
<TextBlock Text="⚠️ 活跃报警" FontWeight="Bold"/>
<ItemsControl ItemsSource="{Binding Alarms}">
<ItemsControl.ItemTemplate>
<DataTemplate>
<TextBlock>
<Run Text="{Binding PointName}"/>
<Run Text=" "/>
<Run Text="{Binding AlarmType}"/>
<Run Text=" = "/>
<Run Text="{Binding Value}"/>
<Run Text=" (阈值: "/>
<Run Text="{Binding Threshold}"/>
<Run Text=")"/>
</TextBlock>
</DataTemplate>
</ItemsControl.ItemTemplate>
</ItemsControl>
</StackPanel>
</Border>
</Grid>
</Window>
三、性能优化技巧
3.1 批量读取
不要一个点一个点读，批量读取效率高10倍：
python复制
# 差：逐个读取
for point in points:
value = client.read_float(point['address'])
# 好：批量读取
addresses = [p['address'] for p in points]
values = client.read_multiple_floats(addresses[0], len(addresses))
3.2 异步编程
Python用asyncio，C#用async/await：
python复制
# Python异步
import asyncio
from pymodbus.client import AsyncModbusTcpClient
async def read_data():
client = AsyncModbusTcpClient('192.168.1.100')
await client.connect()
# 并发读取多个数据点
tasks = [
client.read_holding_registers(address=i*2, count=2)
for i in range(10)
]
results = await asyncio.gather(*tasks)
return results
csharp复制
// C#异步
public async Task<List<float>> ReadAllDataAsync()
{
var tasks = _dataPoints.Select(p => ReadFloatAsync(p.Address));
var results = await Task.WhenAll(tasks);
return results.ToList();
}
3.3 连接池
频繁创建销毁连接很浪费，用连接池：
python复制
# 连接池实现
class ConnectionPool:
def __init__(self, config, pool_size=5):
self.config = config
self.pool = queue.Queue(maxsize=pool_size)
self.lock = threading.Lock()
# 初始化连接池
for _ in range(pool_size):
client = ModbusClient(config)
client.connect()
self.pool.put(client)
def get_client(self):
return self.pool.get()
def return_client(self, client):
self.pool.put(client)
def execute(self, func, *args, **kwargs):
client = self.get_client()
try:
return func(client, *args, **kwargs)
finally:
self.return_client(client)
四、异常处理最佳实践
4.1 重试机制
python复制
import time
from functools import wraps
def retry(max_retries=3, delay=1):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
print(f"重试 {attempt+1}/{max_retries}: {e}")
time.sleep(delay)
return None
return wrapper
return decorator
@retry(max_retries=3, delay=1)
def read_with_retry(client, address):
return client.read_float(address)
4.2 断线重连
python复制
class AutoReconnectClient:
def __init__(self, config):
self.config = config
self.client = None
self.reconnect_thread = None
self.should_reconnect = True
def connect(self):
self.client = ModbusTcpClient(self.config)
return self.client.connect()
def start_reconnect_monitor(self):
"""启动断线重连监控"""
def monitor():
while self.should_reconnect:
if not self.client.connected:
print("连接断开，尝试重连...")
self.connect()
time.sleep(5)
self.reconnect_thread = threading.Thread(target=monitor)
self.reconnect_thread.daemon = True
self.reconnect_thread.start()
4.3 数据校验
python复制
def validate_data(value, point_config):
"""数据校验"""
if value is None:
return False, "读取失败"
# 范围检查
if 'min' in point_config and value < point_config['min']:
return False, f"值低于最小值 {point_config['min']}"
if 'max' in point_config and value > point_config['max']:
return False, f"值超过最大值 {point_config['max']}"
# 变化率检查（防跳变）
if 'last_value' in point_config:
change = abs(value - point_config['last_value'])
if change > point_config.get('max_change', float('inf')):
return False, f"变化率异常: {change}"
return True, "正常"
五、部署与运维
5.1 日志配置
python复制
import logging
from logging.handlers import RotatingFileHandler
def setup_logging():
"""配置日志"""
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# 文件日志（自动轮转）
file_handler = RotatingFileHandler(
'plc_monitor.log',
maxBytes=10*1024*1024, # 10MB
backupCount=5
)
file_handler.setLevel(logging.INFO)
# 控制台日志
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
# 格式
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
5.2 开机自启动
csharp复制
// Windows服务方式
using Microsoft.Extensions.Hosting;
public class PlcMonitorService : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
// 运行监控逻辑
await RunMonitorAsync(stoppingToken);
await Task.Delay(
TimeSpan.FromSeconds(1), stoppingToken);
}
}
}
// 注册服务
Host.CreateDefaultBuilder(args)
.UseWindowsService(options =>
{
options.ServiceName = "PLC Monitor Service";
})
.ConfigureServices(services =>
{
services.AddHostedService<PlcMonitorService>();
})
.Build()
.Run();
六、总结
开发清单
| 阶段 | 任务 | 工具 |
|---|---|---|
| 需求分析 | 确定数据点、通讯协议 | Excel |
| 环境搭建 | 安装Python/C#环境 | pip/nuget |
| 通讯测试 | 验证PLC连接 | Modbus Poll |
| 数据采集 | 实现读写功能 | pymodbus/NModbus |
| 界面开发 | 实现监控界面 | tkinter/WPF |
| 数据存储 | 历史数据记录 | SQLite/MySQL |
| 报警管理 | 异常检测和通知 | 自定义 |
| 部署上线 | 打包、安装、自启动 | PyInstaller/SCM |
常见问题
| 问题 | 原因 | 解决方案 |
|---|---|---|
| 连接超时 | IP/端口错误、防火墙 | ping测试、检查端口 |
| 数据错误 | 字节序不对 | 确认PLC品牌的大端/小端 |
| 频繁断线 | 网络不稳定、交换机问题 | 换交换机、加网线 |
| 性能差 | 逐个读取、同步阻塞 | 批量读取、异步编程 |
作者： 梁某
经验来源： 库卡机器人现场调试
技术栈： Python + C# + Modbus + OPC UA
声明： 本文所有代码均来自实际项目，可直接复用。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
3
0- 0
已为社区贡献3条内容
所有评论(0)