项目初始化

api/modbus_api.py

@@ -0,0 +1,32 @@
+from fastapi import APIRouter
+from api.schemas.modbus_schemas import ModbusIn, ModbusOut
+from common import reponse_code as rc
+from modbus.modbus_util import write_register, read_register_float_value
+
+
+modbus = APIRouter()
+
+
+@modbus.post("/write")
+def write_float_value(modbusIn: ModbusIn):
+    if modbusIn.type == 'float':
+        value = float(modbusIn.value)
+    elif modbusIn.type == 'short':
+        value = int(modbusIn.value)
+    success, msg = write_register(host=modbusIn.host, port=modbusIn.port,
+                                  address=modbusIn.address, value=value)
+    if success:
+        return rc.response_success("修改成功")
+    else:
+        return rc.response_error(msg)
+
+
+@modbus.post("/read")
+def write_float_value(modbusOut: ModbusOut):
+    success, result_dict = read_register_float_value(host=modbusOut.host, port=modbusOut.port, address=modbusOut.address)
+    if success:
+        return rc.response_success(msg="读取成功", data=result_dict)
+    else:
+        return rc.response_error('读取失败')
+
+

api/schemas/modbus_schemas.py

@@ -0,0 +1,16 @@
+from pydantic import BaseModel, Field
+from typing import Optional
+
+
+class ModbusIn(BaseModel):
+    host: Optional[str] = Field(..., description="地址")
+    port: Optional[int] = Field(..., description="端口")
+    address: Optional[int] = Field(..., description="内存地址")
+    value: Optional[str] = Field(..., description="写入的值")
+    type: Optional[str] = Field(..., description="写入值的类型")
+
+
+class ModbusOut(BaseModel):
+    host: Optional[str] = Field(..., description="地址")
+    port: Optional[int] = Field(..., description="端口")
+    address: Optional[list[int]] = Field(..., description="内存地址")