chiyouts
/
iTi-Flask-IOT
forked from iti-framework/iTi-Flask
1
0
Fork 0
Code Pull Requests Activity

添加物联网模块基础接口

Browse Source
iot
DESKTOP-1JS6RSM\Admin 3 months ago
parent 13395ea3e3
commit 6753312961
15 changed files with 1161 additions and 0 deletions
Show Stats Download Patch File Download Diff File

4
iti/applications/models/__init__.py
Unescape Escape View File

@ -10,3 +10,7 @@ from .sys.sys_dept import SysDept, SysDeptSchema
from .sys.sys_rel_user_dept import sys_user_dept
from .sys.sys_menu import SysMenu, SysMenuSchema, SysMenuMetaSchema
from .sys.sys_file import SysFile, SysFileSchema, SysFileDirectory, SysFileDirectorySchema
from .iot.iot_workshop import IotWorkshop, IotWorkshopSchema
from .iot.iot_device import IotDevice, IotDeviceSchema
from .iot.iot_endpoint import IotEndpoint, IotEndpointSchema
from .iot.iot_node import IotNode, IotNodeSchema

44
iti/applications/models/iot/iot_device.py
Unescape Escape View File

@ -0,0 +1,44 @@
from iti.applications.extensions import db
from iti.applications.common.crud import IdModelMixin, TimeModelMixin
from iti.applications.common.utils import BaseSchema
from apiflask.fields import String, Integer, DateTime
import uuid
class IotDevice(db.Model, IdModelMixin, TimeModelMixin):
"""
设备信息表
"""
__tablename__ = "iot_device"
id = db.Column(
db.String(36),
primary_key=True,
default=lambda: str(uuid.uuid4().hex),
comment="标识",
)
workshop_id = db.Column(db.String(36), nullable=False, unique=True, comment="车间ID")
device_name = db.Column(db.String(255), nullable=False, unique=True, comment="设备名称")
device_number = db.Column(db.String(20), nullable=False, comment="设备编号")
description = db.Column(db.Text, nullable=True, comment="设备描述")
brand_name = db.Column(db.String(255), nullable=False, comment="品牌名称")
specification_model = db.Column(db.String(255), nullable=False, comment="规格型号")
status = db.Column(db.Integer, nullable=False, default=0, comment="状态 0:已停机 1:生产中 2:维修中")
class IotDeviceSchema(BaseSchema):
"""
设备信息表响应结构
"""
class Meta:
name = "IotDevice"
id = String()
workshop_id = String()
device_name = String()
device_number = String()
description = String()
brand_name = String()
specification_model = String()
status = Integer()
created_at = DateTime(format="%Y-%m-%d %H:%M:%S")
updated_at = DateTime(format="%Y-%m-%d %H:%M:%S")

49
iti/applications/models/iot/iot_endpoint.py
Unescape Escape View File

@ -0,0 +1,49 @@
from iti.applications.extensions import db
from iti.applications.common.crud import IdModelMixin, TimeModelMixin
from iti.applications.common.utils import BaseSchema
from apiflask.fields import String, Integer, DateTime
import uuid
class IotEndpoint(db.Model, IdModelMixin, TimeModelMixin):
"""
采集端信息表
"""
__tablename__ = "iot_endpoint"
id = db.Column(
db.String(36),
primary_key=True,
default=lambda: str(uuid.uuid4().hex),
comment="标识",
)
device_id = db.Column(db.String(36), nullable=False, unique=True, comment="设备ID")
endpoint_name = db.Column(db.String(255), nullable=False, unique=True, comment="采集端名称")
endpoint_number = db.Column(db.String(20), nullable=False, comment="采集端编号")
description = db.Column(db.Text, nullable=True, comment="采集端描述")
ip = db.Column(db.String(255), nullable=False, comment="采集端IP")
port = db.Column(db.String(255), nullable=False, comment="采集端端口")
brand_name = db.Column(db.String(255), nullable=False, comment="品牌名称")
specification_model = db.Column(db.String(255), nullable=False, comment="规格型号")
status = db.Column(db.Integer, nullable=False, default=0, comment="状态 0:停用 1:运行中 2:维修中")
class IotEndpointSchema(BaseSchema):
"""
采集端信息表响应结构
"""
class Meta:
name = "IotEndpoint"
id = String()
device_id = String()
endpoint_name = String()
endpoint_number = String()
description = String()
ip = String()
port = String()
brand_name = String()
specification_model = String()
status = Integer()
created_at = DateTime(format="%Y-%m-%d %H:%M:%S")
updated_at = DateTime(format="%Y-%m-%d %H:%M:%S")

55
iti/applications/models/iot/iot_node.py
Unescape Escape View File

@ -0,0 +1,55 @@
from iti.applications.extensions import db
from iti.applications.common.crud import IdModelMixin, TimeModelMixin
from iti.applications.common.utils import BaseSchema
from apiflask.fields import String, Integer, DateTime
import uuid
class IotNode(db.Model, IdModelMixin, TimeModelMixin):
"""
节点信息表
"""
__tablename__ = "iot_node"
id = db.Column(
db.String(36),
primary_key=True,
default=lambda: str(uuid.uuid4().hex),
comment="标识",
)
endpoint_id = db.Column(db.String(36), nullable=False, unique=True, comment="采集端ID")
title = db.Column(db.String(255), nullable=False, unique=True, comment="节点ID")
mark = db.Column(db.String(255), nullable=False, comment="采集标识")
mark_type = db.Column(db.Integer, nullable=True, comment="采集类型 1:只读 2:只写 3:读写")
tag_label = db.Column(db.String(255), nullable=False, comment="变量别名，用于数据存储标记")
data_type = db.Column(db.String(255), nullable=False, comment="值类型 text: 文本 int: 整型 float: 浮点型 boolean:布尔型")
is_warning = db.Column(db.Integer, nullable=False, comment="预警类型 0:无预警 1:预警")
warning_effective_config = db.Column(db.String(255), nullable=False, comment="预警触发表达式")
is_calling = db.Column(db.Integer, nullable=False, comment="报警类型 0:无报警 1:报警")
calling_effective_config = db.Column(db.String(255), nullable=False, comment="报警触发表达式")
method_content = db.Column(db.String(255), nullable=False, comment="方法节点")
status = db.Column(db.Integer, nullable=False, default=0, comment="状态 0:禁用 1:启用")
class IotNodeSchema(BaseSchema):
"""
节点信息表响应结构
"""
class Meta:
name = "IotNode"
id = String()
endpoint_id = String()
title = String()
mark = String()
mark_type = Integer()
tag_label = String()
data_type = String()
is_warning = Integer()
warning_effective_config = String()
is_calling = Integer()
calling_effective_config = String()
method_content = String()
status = Integer()
created_at = DateTime(format="%Y-%m-%d %H:%M:%S")
updated_at = DateTime(format="%Y-%m-%d %H:%M:%S")

42
iti/applications/models/iot/iot_workshop.py
Unescape Escape View File

@ -0,0 +1,42 @@
from iti.applications.extensions import db
from iti.applications.common.crud import IdModelMixin, TimeModelMixin, RemarkModelMixin
from iti.applications.common.utils import BaseSchema
from apiflask.fields import String, Integer, DateTime
import uuid
class IotWorkshop(db.Model, IdModelMixin, TimeModelMixin, RemarkModelMixin):
"""
车间信息表
"""
__tablename__ = "iot_workshop"
id = db.Column(
db.String(36),
primary_key=True,
default=lambda: str(uuid.uuid4().hex),
comment="标识",
)
workshop_name = db.Column(db.String(255), nullable=False, unique=True, comment="车间名称")
workshop_number = db.Column(db.String(50), nullable=False, comment="车间编号")
total_area = db.Column(db.String(50), nullable=True, comment="总面积(单位:平方米)")
director_name = db.Column(db.String(30), nullable=False, comment="负责人姓名")
director_phone = db.Column(db.String(15), nullable=False, comment="负责人电话")
status = db.Column(db.Integer, nullable=False, default=0, comment="状态 0:已停用 1:生产中")
class IotWorkshopSchema(BaseSchema):
"""
车间信息表响应结构
"""
class Meta:
name = "IotWorkshop"
id = String()
workshop_name = String()
workshop_number = String()
total_area = String()
director_name = String()
director_phone = String()
status = Integer()
remark = String()
created_at = DateTime(format="%Y-%m-%d %H:%M:%S")
updated_at = DateTime(format="%Y-%m-%d %H:%M:%S")

4
iti/applications/routes/__init__.py
Unescape Escape View File

@ -1,5 +1,6 @@
from iti.applications.extensions import broadcast_execute
from iti.applications.routes.common import register_common_bp
from iti.applications.routes.iot import register_iot_bp
from iti.applications.routes.sys import register_sys_bp
from iti.applications.routes.index import bp as index_bp
from iti.applications.routes.front import bp as frontend_bp
@ -17,5 +18,8 @@ def init_routes(app):
# 系统API蓝图注册
register_sys_bp(app)
# 物联网API注册
register_iot_bp(app)
# 插件初始化
broadcast_execute(app, "event_init")

16
iti/applications/routes/iot/__init__.py
Unescape Escape View File

@ -0,0 +1,16 @@
from apiflask import APIBlueprint
from .workshop_ctl import bp as workshop_bp
from .device_ctl import bp as device_bp
from .endpoint_ctl import bp as endpoint_bp
from .node_ctl import bp as node_bp
iot_bp = APIBlueprint("iot", __name__, url_prefix="/iot")
def register_iot_bp(app):
iot_bp.register_blueprint(workshop_bp)
iot_bp.register_blueprint(device_bp)
iot_bp.register_blueprint(endpoint_bp)
iot_bp.register_blueprint(node_bp)
app.register_blueprint(iot_bp)

108
iti/applications/routes/iot/device_ctl.py
Unescape Escape View File

@ -0,0 +1,108 @@
from apiflask import APIBlueprint
from iti.applications.extensions import db, sys_log
from iti.applications.common.utils import success, page_schema, page
from iti.applications.models import (
IotDevice,
IotDeviceSchema,
)
from .schemas.device import (
DeviceQuery,
DeviceAddRequest,
DeviceUpdateRequest,
)
from iti.applications.common import ModelFilter
from iti.applications.common.exceptions.biz_exp import BizException
from flask_jwt_extended import jwt_required, current_user
from sqlalchemy import select, delete, exists
from sqlalchemy.orm import noload
from iti.applications.common import permission
bp = APIBlueprint("iot_device", __name__, url_prefix="/device", tag="设备管理")
@bp.get("/list")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:device:list")
@bp.input(DeviceQuery.Schema(partial=True), location="query")
@bp.output(IotDeviceSchema(many=True))
def list_workshop(query_data: DeviceQuery):
"""
获取设备列表
"""
return success(get_list_or_page(query_data))
@bp.get("/page")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:device:list")
@bp.input(DeviceQuery.Schema(partial=True), location="query")
@bp.output(page_schema(IotDeviceSchema(many=True)))
def page_user(query_data: DeviceQuery):
"""
分页获取设备列表
"""
return page(get_list_or_page(query_data))
@bp.post("/add")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:device:add")
@bp.input(DeviceAddRequest, location="json")
def add_workshop(json_data: dict):
"""
添加设备信息
"""
device = IotDevice(**json_data)
device.status = 0
db.session.add(device)
db.session.commit()
return success()
@bp.put("/<string:id>")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:device:update")
@bp.input(DeviceUpdateRequest(partial=True), location="json")
def update_workshop(id: str, json_data: dict):
"""
更新设备信息
"""
device = db.session.scalar(select(IotDevice).filter_by(id=id))
if not device:
raise BizException("设备信息不存在")
for key, value in json_data.items():
if value is not None:
setattr(device, key, value)
db.session.commit()
return success()
def get_list_or_page(query_data: DeviceQuery):
"""
获取设备信息列表或分页
"""
query = select(IotDevice).order_by(IotDevice.created_at.desc())
if query_data.keyword:
kw = ModelFilter.escape_like(query_data.keyword)
query = query.filter(
IotDevice.device_name.like(f"%{kw}%")
| IotDevice.device_number.like(f"%{kw}%")
)
if query_data.workshop_id:
query = query.filter(IotDevice.workshop_id == query_data.workshop_id)
if query_data.status:
query = query.filter(IotDevice.status == query_data.status)
if query_data.page and query_data.size:
return db.paginate(query, page=query_data.page, per_page=query_data.size)
else:
return db.session.scalars(query).all()

108
iti/applications/routes/iot/endpoint_ctl.py
Unescape Escape View File

@ -0,0 +1,108 @@
from apiflask import APIBlueprint
from iti.applications.extensions import db, sys_log
from iti.applications.common.utils import success, page_schema, page
from iti.applications.models import (
IotEndpoint,
IotEndpointSchema,
)
from .schemas.endpoint import (
EndpointQuery,
EndpointAddRequest,
EndpointUpdateRequest,
)
from iti.applications.common import ModelFilter
from iti.applications.common.exceptions.biz_exp import BizException
from flask_jwt_extended import jwt_required, current_user
from sqlalchemy import select, delete, exists
from sqlalchemy.orm import noload
from iti.applications.common import permission
bp = APIBlueprint("iot_endpoint", __name__, url_prefix="/endpoint", tag="采集端管理")
@bp.get("/list")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:endpoint:list")
@bp.input(EndpointQuery.Schema(partial=True), location="query")
@bp.output(IotEndpointSchema(many=True))
def list_workshop(query_data: EndpointQuery):
"""
获取采集端列表
"""
return success(get_list_or_page(query_data))
@bp.get("/page")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:endpoint:list")
@bp.input(EndpointQuery.Schema(partial=True), location="query")
@bp.output(page_schema(IotEndpointSchema(many=True)))
def page_user(query_data: EndpointQuery):
"""
分页获取采集端列表
"""
return page(get_list_or_page(query_data))
@bp.post("/add")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:endpoint:add")
@bp.input(EndpointAddRequest, location="json")
def add_workshop(json_data: dict):
"""
添加采集端信息
"""
device = IotEndpoint(**json_data)
device.status = 0
db.session.add(device)
db.session.commit()
return success()
@bp.put("/<string:id>")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:endpoint:update")
@bp.input(EndpointUpdateRequest(partial=True), location="json")
def update_workshop(id: str, json_data: dict):
"""
更新采集端信息
"""
endpoint = db.session.scalar(select(IotEndpoint).filter_by(id=id))
if not endpoint:
raise BizException("采集端信息不存在")
for key, value in json_data.items():
if value is not None:
setattr(endpoint, key, value)
db.session.commit()
return success()
def get_list_or_page(query_data: EndpointQuery):
"""
获取采集端信息列表或分页
"""
query = select(IotEndpoint).order_by(IotEndpoint.created_at.desc())
if query_data.keyword:
kw = ModelFilter.escape_like(query_data.keyword)
query = query.filter(
IotEndpoint.endpoint_name.like(f"%{kw}%")
| IotEndpoint.endpoint_number.like(f"%{kw}%")
)
if query_data.device_id:
query = query.filter(IotEndpoint.device_id == query_data.device_id)
if query_data.status:
query = query.filter(IotEndpoint.status == query_data.status)
if query_data.page and query_data.size:
return db.paginate(query, page=query_data.page, per_page=query_data.size)
else:
return db.session.scalars(query).all()

102
iti/applications/routes/iot/node_ctl.py
Unescape Escape View File

@ -0,0 +1,102 @@
from apiflask import APIBlueprint
from iti.applications.extensions import db, sys_log
from iti.applications.common.utils import success, page_schema, page
from iti.applications.models import (
IotNode,
IotNodeSchema,
)
from .schemas.node import (
NodeQuery,
NodeAddRequest,
NodeUpdateRequest,
)
from iti.applications.common import ModelFilter
from iti.applications.common.exceptions.biz_exp import BizException
from flask_jwt_extended import jwt_required, current_user
from sqlalchemy import select, delete, exists
from sqlalchemy.orm import noload
from iti.applications.common import permission
bp = APIBlueprint("iot_node", __name__, url_prefix="/node", tag="采集节点管理")
@bp.get("/list")
# @jwt_required()
# @bp.doc(security="JWT")
# @permission("iot:node:list")
@bp.input(NodeQuery.Schema(partial=True), location="query")
@bp.output(IotNodeSchema(many=True))
def list_workshop(query_data: NodeQuery):
"""
获取采集节点列表
"""
return success(get_list_or_page(query_data))
@bp.get("/page")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:node:list")
@bp.input(NodeQuery.Schema(partial=True), location="query")
@bp.output(page_schema(IotNodeSchema(many=True)))
def page_user(query_data: NodeQuery):
"""
分页获取采集节点列表
"""
return page(get_list_or_page(query_data))
@bp.post("/add")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:node:add")
@bp.input(NodeAddRequest, location="json")
def add_workshop(json_data: dict):
"""
添加采集节点信息
"""
node = IotNode(**json_data)
node.status = 0
db.session.add(node)
db.session.commit()
return success()
@bp.put("/<string:id>")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:node:update")
@bp.input(NodeUpdateRequest(partial=True), location="json")
def update_workshop(id: str, json_data: dict):
"""
更新采集节点信息
"""
node = db.session.scalar(select(IotNode).filter_by(id=id))
if not node:
raise BizException("节点信息不存在")
for key, value in json_data.items():
if value is not None:
setattr(node, key, value)
db.session.commit()
return success()
def get_list_or_page(query_data: NodeQuery):
"""
获取采集节点列表或分页
"""
query = select(IotNode).order_by(IotNode.created_at.desc())
if query_data.endpoint_id:
query = query.filter(IotNode.endpoint_id == query_data.endpoint_id)
if query_data.status:
query = query.filter(IotNode.status == query_data.status)
if query_data.page and query_data.size:
return db.paginate(query, page=query_data.page, per_page=query_data.size)
else:
return db.session.scalars(query).all()

117
iti/applications/routes/iot/schemas/device.py
Unescape Escape View File

@ -0,0 +1,117 @@
from dataclasses import field
from marshmallow_dataclass import dataclass
from marshmallow import validates_schema, ValidationError
from iti.applications.common.utils.schema import BaseSchema, Pagination
from typing import ClassVar, Optional
from apiflask import fields
@dataclass(base_schema=BaseSchema)
class DeviceQuery(Pagination):
"""
设备信息查询请求
"""
keyword: Optional[str] = field(
default=None,
metadata={
"required": False,
"metadata": {
"description": "关键字 [设备名称|设备编号] 模糊查询"
},
},
)
workshop_id: str = field(
default=None,
metadata={
"required": False,
"metadata": {"example": "ase1f2", "description": "车间ID"},
},
)
status: int = field(
default=None,
metadata={
"required": False,
"metadata": {"example": 0, "description": "状态"},
},
)
Schema: ClassVar[BaseSchema] = BaseSchema # For the type check
class DeviceAddRequest(BaseSchema):
"""
新增设备信息
"""
workshop_id = fields.String(
required=False,
metadata={"example": "车间ID", "description": "车间ID"},
load_default=None,
)
device_name = fields.String(
required=False,
metadata={"example": "设备名称", "descriptrion": "设备名称"},
load_default=None,
)
device_number = fields.String(
required=False,
metadata={"example": "设备编号", "description": "设备编号"},
load_default=None,
)
description = fields.String(
required=False,
metadata={"example": "设备描述", "description": "设备描述"},
load_default=None,
)
brand_name = fields.String(
required=False,
metadata={"example": "品牌名称", "description": "品牌名称"},
load_default=None,
)
specification_model = fields.String(
required=False,
metadata={"example": "规格型号", "description": "规格型号"},
load_default=None,
)
class DeviceUpdateRequest(BaseSchema):
"""
更新设备信息
"""
workshop_id = fields.String(
required=False,
metadata={"example": "车间ID", "description": "车间ID"},
load_default=None,
)
device_name = fields.String(
required=False,
metadata={"example": "设备名称", "descriptrion": "设备名称"},
load_default=None,
)
device_number = fields.String(
required=False,
metadata={"example": "设备编号", "description": "设备编号"},
load_default=None,
)
description = fields.String(
required=False,
metadata={"example": "设备描述", "description": "设备描述"},
load_default=None,
)
brand_name = fields.String(
required=False,
metadata={"example": "品牌名称", "description": "品牌名称"},
load_default=None,
)
specification_model = fields.String(
required=False,
metadata={"example": "规格型号", "description": "规格型号"},
load_default=None,
)
status = fields.Integer(
required=False,
metadata={"example": 1, "description": "状态(0-已停机,1-生产中,2-维修中)"},
load_default=None,
)

137
iti/applications/routes/iot/schemas/endpoint.py
Unescape Escape View File

@ -0,0 +1,137 @@
from dataclasses import field
from marshmallow_dataclass import dataclass
from marshmallow import validates_schema, ValidationError
from iti.applications.common.utils.schema import BaseSchema, Pagination
from typing import ClassVar, Optional
from apiflask import fields
@dataclass(base_schema=BaseSchema)
class EndpointQuery(Pagination):
"""
采集端信息查询请求
"""
keyword: Optional[str] = field(
default=None,
metadata={
"required": False,
"metadata": {
"description": "关键字 [采集端名称|采集端编号|采集端IP] 模糊查询"
},
},
)
device_id: str = field(
default=None,
metadata={
"required": False,
"metadata": {"example": "ase1f2", "description": "设备ID"},
},
)
status: int = field(
default=None,
metadata={
"required": False,
"metadata": {"example": 0, "description": "状态"},
},
)
Schema: ClassVar[BaseSchema] = BaseSchema # For the type check
class EndpointAddRequest(BaseSchema):
"""
采集端新增信息
"""
device_id = fields.String(
required=False,
metadata={"example": "设备ID", "description": "设备ID"},
load_default=None,
)
endpoint_name = fields.String(
required=False,
metadata={"example": "采集端名称", "descriptrion": "采集端名称"},
load_default=None,
)
endpoint_number = fields.String(
required=False,
metadata={"example": "采集端编号", "description": "采集端编号"},
load_default=None,
)
description = fields.String(
required=False,
metadata={"example": "采集端描述", "description": "采集端描述"},
load_default=None,
)
ip = fields.String(
required=False,
metadata={"example": "采集端IP", "description": "采集端IP"},
load_default=None,
)
port = fields.String(
required=False,
metadata={"example": "采集端端口", "description": "采集端端口"},
load_default=None,
)
brand_name = fields.String(
required=False,
metadata={"example": "品牌名称", "description": "品牌名称"},
load_default=None,
)
specification_model = fields.String(
required=False,
metadata={"example": "规格型号", "description": "规格型号"},
load_default=None,
)
class EndpointUpdateRequest(BaseSchema):
"""
更新采集端信息
"""
device_id = fields.String(
required=False,
metadata={"example": "设备ID", "description": "设备ID"},
load_default=None,
)
endpoint_name = fields.String(
required=False,
metadata={"example": "采集端名称", "descriptrion": "采集端名称"},
load_default=None,
)
endpoint_number = fields.String(
required=False,
metadata={"example": "采集端编号", "description": "采集端编号"},
load_default=None,
)
description = fields.String(
required=False,
metadata={"example": "采集端描述", "description": "采集端描述"},
load_default=None,
)
ip = fields.String(
required=False,
metadata={"example": "采集端IP", "description": "采集端IP"},
load_default=None,
)
port = fields.String(
required=False,
metadata={"example": "采集端端口", "description": "采集端端口"},
load_default=None,
)
brand_name = fields.String(
required=False,
metadata={"example": "品牌名称", "description": "品牌名称"},
load_default=None,
)
specification_model = fields.String(
required=False,
metadata={"example": "规格型号", "description": "规格型号"},
load_default=None,
)
status = fields.Integer(
required=False,
metadata={"example": 1, "description": "状态(0-已停用,1-生产中,2-维修中)"},
load_default=None,
)

158
iti/applications/routes/iot/schemas/node.py
Unescape Escape View File

@ -0,0 +1,158 @@
from dataclasses import field
from marshmallow_dataclass import dataclass
from marshmallow import validates_schema, ValidationError
from iti.applications.common.utils.schema import BaseSchema, Pagination
from typing import ClassVar, Optional
from apiflask import fields
@dataclass(base_schema=BaseSchema)
class NodeQuery(Pagination):
"""
节点信息查询请求
"""
endpoint_id: str = field(
default=None,
metadata={
"required": False,
"metadata": {"example": "ase1f2", "description": "采集端ID"},
},
)
status: int = field(
default=None,
metadata={
"required": False,
"metadata": {"example": 0, "description": "状态"},
},
)
Schema: ClassVar[BaseSchema] = BaseSchema # For the type check
class NodeAddRequest(BaseSchema):
"""
节点新增信息
"""
endpoint_id = fields.String(
required=False,
metadata={"example": "采集端ID", "description": "采集端ID"},
load_default=None,
)
title = fields.String(
required=False,
metadata={"example": "节点ID", "descriptrion": "节点ID"},
load_default=None,
)
mark = fields.String(
required=False,
metadata={"example": "采集标识", "description": "采集标识"},
load_default=None,
)
mark_type = fields.Integer(
required=False,
metadata={"example": "读写标识", "description": "采集类型 1:只读 2:只写 3:读写"},
load_default=1,
)
tag_label = fields.String(
required=False,
metadata={"example": "变量别名", "description": "变量别名"},
load_default=None,
)
data_type = fields.String(
required=False,
metadata={"example": "text", "description": "值类型 text: 文本 int: 整型 float: 浮点型 boolean:布尔型"},
load_default=None,
)
is_warning = fields.Integer(
required=False,
metadata={"example": "是否预警", "description": "预警类型"},
load_default=0,
)
warning_effective_config = fields.String(
required=False,
metadata={"example": "预警触发表达式", "description": "预警触发表达式"},
load_default=None,
)
is_calling = fields.Integer(
required=False,
metadata={"example": "是否报警", "description": "报警类型"},
load_default=0,
)
calling_effective_config = fields.String(
required=False,
metadata={"example": "报警触发表达式", "description": "报警触发表达式"},
load_default=None,
)
method_content = fields.String(
required=False,
metadata={"example": "方法节点", "description": "方法节点"},
load_default=None,
)
class NodeUpdateRequest(BaseSchema):
"""
节点更新信息
"""
endpoint_id = fields.String(
required=False,
metadata={"example": "采集端ID", "description": "采集端ID"},
load_default=None,
)
title = fields.String(
required=False,
metadata={"example": "节点ID", "descriptrion": "节点ID"},
load_default=None,
)
mark = fields.String(
required=False,
metadata={"example": "采集标识", "description": "采集标识"},
load_default=None,
)
mark_type = fields.Integer(
required=False,
metadata={"example": "读写标识", "description": "采集类型 1:只读 2:只写 3:读写"},
load_default=1,
)
tag_label = fields.String(
required=False,
metadata={"example": "变量别名", "description": "变量别名"},
load_default=None,
)
data_type = fields.String(
required=False,
metadata={"example": "text", "description": "值类型 text: 文本 int: 整型 float: 浮点型 boolean:布尔型"},
load_default=None,
)
is_warning = fields.Integer(
required=False,
metadata={"example": "是否预警", "description": "预警类型"},
load_default=0,
)
warning_effective_config = fields.String(
required=False,
metadata={"example": "预警触发表达式", "description": "预警触发表达式"},
load_default=None,
)
is_calling = fields.Integer(
required=False,
metadata={"example": "是否报警", "description": "报警类型"},
load_default=0,
)
calling_effective_config = fields.String(
required=False,
metadata={"example": "报警触发表达式", "description": "报警触发表达式"},
load_default=None,
)
method_content = fields.String(
required=False,
metadata={"example": "方法节点", "description": "方法节点"},
load_default=None,
)
status = fields.Integer(
required=False,
metadata={"example": 1, "description": "状态(0-禁用,1-启用)"},
load_default=None,
)

111
iti/applications/routes/iot/schemas/workshop.py
Unescape Escape View File

@ -0,0 +1,111 @@
from dataclasses import field
from marshmallow_dataclass import dataclass
from marshmallow import validates_schema, ValidationError
from iti.applications.common.utils.schema import BaseSchema, Pagination
from typing import ClassVar, Optional
from apiflask import fields
@dataclass(base_schema=BaseSchema)
class WorkshopQuery(Pagination):
"""
车间信息查询请求
"""
keyword: Optional[str] = field(
default=None,
metadata={
"required": False,
"metadata": {
"description": "关键字 [车间名称|车间编号|负责人姓名] 模糊查询"
},
},
)
status: int = field(
default=None,
metadata={
"required": False,
"metadata": {"example": 0, "description": "状态"},
},
)
Schema: ClassVar[BaseSchema] = BaseSchema # For the type check
class WorkshopAddRequest(BaseSchema):
"""
新增车间信息
"""
workshop_name = fields.String(
required=False,
metadata={"example": "车间名称", "description": "车间名称"},
load_default=None,
)
workshop_number = fields.String(
required=False,
metadata={"example": "车间编号", "description": "车间编号"},
load_default=None,
)
total_area = fields.String(
required=False,
metadata={"example": "1234", "description": "总面积(单位:平方米)"},
load_default=None,
)
director_name = fields.String(
required=False,
metadata={"example": "负责人姓名", "description": "负责人姓名"},
load_default=None,
)
director_phone = fields.String(
required=False,
metadata={"example": "负责人电话", "description": "负责人电话"},
load_default=None,
)
remark = fields.String(
required=False,
metadata={"example": "备注", "description": "备注"},
load_default=None,
)
class WorkshopUpdateRequest(BaseSchema):
"""
更新车间信息
"""
workshop_name = fields.String(
required=False,
metadata={"example": "车间名称", "description": "车间名称"},
load_default=None,
)
workshop_number = fields.String(
required=False,
metadata={"example": "车间编号", "description": "车间编号"},
load_default=None,
)
total_area = fields.String(
required=False,
metadata={"example": "1234", "description": "总面积(单位:平方米)"},
load_default=None,
)
director_name = fields.String(
required=False,
metadata={"example": "负责人姓名", "description": "负责人姓名"},
load_default=None,
)
director_phone = fields.String(
required=False,
metadata={"example": "负责人电话", "description": "负责人电话"},
load_default=None,
)
status = fields.Integer(
required=False,
metadata={"example": 1, "description": "状态(0-已停用,1-生产中)"},
load_default=None,
)
remark = fields.String(
required=False,
metadata={"example": "备注", "description": "备注"},
load_default=None,
)

106
iti/applications/routes/iot/workshop_ctl.py
Unescape Escape View File

@ -0,0 +1,106 @@
from apiflask import APIBlueprint
from iti.applications.extensions import db, sys_log
from iti.applications.common.utils import success, page_schema, page
from iti.applications.models import (
IotWorkshop,
IotWorkshopSchema,
)
from .schemas.workshop import (
WorkshopQuery,
WorkshopAddRequest,
WorkshopUpdateRequest,
)
from iti.applications.common import ModelFilter
from iti.applications.common.exceptions.biz_exp import BizException
from flask_jwt_extended import jwt_required, current_user
from sqlalchemy import select, delete, exists
from sqlalchemy.orm import noload
from iti.applications.common import permission
bp = APIBlueprint("iot_workshop", __name__, url_prefix="/workshop", tag="车间管理")
@bp.get("/list")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:workshop:list")
@bp.input(WorkshopQuery.Schema(partial=True), location="query")
@bp.output(IotWorkshopSchema(many=True))
def list_workshop(query_data: WorkshopQuery):
"""
获取车间列表
"""
return success(get_list_or_page(query_data))
@bp.get("/page")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:workshop:list")
@bp.input(WorkshopQuery.Schema(partial=True), location="query")
@bp.output(page_schema(IotWorkshopSchema(many=True)))
def page_user(query_data: WorkshopQuery):
"""
分页获取车间列表
"""
return page(get_list_or_page(query_data))
@bp.post("/add")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:workshop:add")
@bp.input(WorkshopAddRequest, location="json")
def add_workshop(json_data: dict):
"""
添加车间信息
"""
workshop = IotWorkshop(**json_data)
workshop.status = 0
db.session.add(workshop)
db.session.commit()
return success()
@bp.put("/<string:id>")
@jwt_required()
@bp.doc(security="JWT")
@permission("iot:workshop:update")
@bp.input(WorkshopUpdateRequest(partial=True), location="json")
def update_workshop(id: str, json_data: dict):
"""
更新车间信息
"""
workshop = db.session.scalar(select(IotWorkshop).filter_by(id=id))
if not workshop:
raise BizException("车间信息不存在")
for key, value in json_data.items():
if value is not None:
setattr(workshop, key, value)
db.session.commit()
return success()
def get_list_or_page(query_data: WorkshopQuery):
"""
获取车间信息列表或分页
"""
query = select(IotWorkshop).order_by(IotWorkshop.created_at.desc())
if query_data.keyword:
kw = ModelFilter.escape_like(query_data.keyword)
query = query.filter(
IotWorkshop.workshop_name.like(f"%{kw}%")
| IotWorkshop.workshop_number.like(f"%{kw}%")
| IotWorkshop.director_name.like(f"%{kw}%")
)
if query_data.status:
query = query.filter(IotWorkshop.status == query_data.status)
if query_data.page and query_data.size:
return db.paginate(query, page=query_data.page, per_page=query_data.size)
else:
return db.session.scalars(query).all()
Write Preview
Loading…
Cancel
Save