README.md · 小马/DTCloud

# DTCloud

#### 介绍
中亿丰数字的DTCloud数字中台是一个面向数字孪生企业和工业物联网的快速应用开发平台。该平台采用BIM/CIM+AIoT的技术架构和数据管理技术，旨在实现可视数化的数字孪生管理，助力打造智能企业的实施及相关应用。

DTCloud数字中台提供BIM/CIM可视化展示模块、IoT数据中台、AI数据分析模块，支持智慧大屏、移动端APP及web页面等多终端，平台提供海量开发接口和项目模块，实现智慧场景解决方案的快速应用开发。

DTCloud数字中台具有以下特点：

数据集中管理：前端采用3D空间图形展示数据，数据集中管理，支持实时感知企业、园区的运行状态，提前预知风险，优化运行效率。
    
    AI分析和仿真模拟分析：后台提供AI分析和仿真模拟分析，实现物理和数字企业之间的信息同步及管控，提升能效、安全、健康及智能化管理水平。
    
    智慧场景解决方案的快速应用开发：DTCloud中台集成工地，园区及企业场景中50多个大门类的300多种接口，提供海量开发接口和项目模块，实现智慧场景解决方案的快速应用开发。
    
    低门槛、智能化产品：数据中台产品可以支撑建设过程中的智慧建造应用和建成后的智慧运维应用，是一款可以应用于企业全生命周期管理的低门槛、智能化产品。

[![DTCloud317](dtcloud/addons/web/static/img/dtcloud4-1.png)](https://e.gitee.com/open_product/repos/dtcloud360/dtcloud/tree/DTCloud-4.18)
[![DTCloud317](dtcloud/addons/web/static/img/dtcloud4-2.png)](https://e.gitee.com/open_product/repos/dtcloud360/dtcloud/tree/DTCloud-4.18)

备注：DTCloud需要Python3.10以上的版本，推荐3.10.9版本，默认账号密码是 admin/admin

> #### 本地服务管理：(兼容.conf以及.yaml格式配置文件)
> * 启动：python dtcloud.py -c dtcloud.yaml
> * 启动queue_job：python dtcloud.py jobqueue -c dtcloud.yaml

> #### 接口编写简化标准如下：

```python
# -*- coding: utf-8 -*-
# &&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&
# DTCloud360
# QQ:35350428
# 邮件:35350428@qq.com
# 手机：13584935775
# 作者：yu.qian
# 公司网址： http://www.dtcloud360.com/
# Copyright 中亿丰数字科技集团有限公司 2012-2025
# 日期：
# 文件:main.py
# 备注:基于Springboot&PEP8规范的控制器的接口类
# &&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&

# 标准库
import io
import logging
import pandas as pd

# 第三方库
from dtcloud import http

# 当前项目自定义的模块
from commons.web.decorators import render_json, render_file
from commons.web.objects import SuccessData
from commons.core import CRUDBase

# from dtcloud.addons.BaseException.api_error import AppValidationError, AppResourceError, ValidationError

_logger = logging.getLogger(__name__)

def save_data_to_excel_by_pandas():
    """
    pandas写入excel
    """
    # excel的表头
    struct_columns = ["日期", "网络购票人数", "线下购票人数", "团队购票人数", "总购票人数"]
    # 数据拼接的结构必须为[[]] ，内部列表的长度必须跟表头长度一致
    datas = [
        ["2022/10/27", 42, 35, 23, 160]
    ]
    df_detail = pd.DataFrame(datas, columns=struct_columns)
    bio = io.BytesIO()

with pd.ExcelWriter(bio, engine='xlsxwriter', engine_kwargs={'options': {'strings_to_urls': False}}) as writer:
        # 答卷数据
        df_detail.to_excel(writer, sheet_name='data', index=False)
        df_detail_shape = df_detail.shape[1]
        workbook = writer.book
        format_border = workbook.add_format({'border_color': '#e1e4eb'})  # 设置边框格式
        worksheets = writer.sheets
        worksheet1 = worksheets['data']
        # 这里是核心，根据条件来设置格式
        worksheet1.conditional_format(0, 0, 0, df_detail_shape, {'type': 'no_blanks', 'format': format_border})
        writer.save()
        bio.seek(0)
        data = bio.read()
    return data

class DtMockController(http.Controller, CRUDBase):
    """
    Mock控制器

"""

@http.route("/api/project/getProjectData/<int:role_id>", type="http", methods=['GET'], auth="none", sitemap=False,
                csrf=False,
                cors="*",
                description='获取Mock数据', group='获取Mock数据')
    @render_json # 返回标准json格式装饰器
    def get_mock_data(self, **kw):
        """
        获取dt.sensitive资源列表
        :param kw:
        :return:
        """
        a = {"mock": "这是一条mock数据"}
        return SuccessData(a=a, msg='获取dt.sensitive资源列表成功')

@http.route("/api/getFileData", type="http", methods=['GET'], auth="none", sitemap=False,
                csrf=False,
                cors="*",
                description='获取File类型文件', group='获取File类型文件')
    @render_file # 返回文件流装饰器
    def get_file_data(self, **kw):
        data = save_data_to_excel_by_pandas()
        return '公司信息.xlsx', data
```
> #### queue_job常用方法介绍：
```python
# 1、将作业排入队列的方法是在记录或模型上使用 with_delay()
def button_done(self):
    self.with_delay().print_confirmation_document(self.state)
    self.write({"state": "done"})
    return True

# 2、定义依赖关系
def button_chain_done(self):
    self.ensure_one()
    job1 = self.browse(1).delayable().generate_thumbnail((50, 50))
    job2 = self.browse(1).delayable().generate_thumbnail((50, 50))
    job3 = self.browse(1).delayable().generate_thumbnail((50, 50))
    
    """
    在job2上调用on_done(job3)方法，意味着只有在job3完成后，job2才会被执行。
    同样，job1只有在job2完成后才会被执行。这创建了一个依赖链，其中job3首先被执行，然后是job2，最后是job1。
    """
    job1.on_done(job2.on_done(job3)).delay()

# 3、高阶用法介绍 chain & group
"""
	其中 链（chain）表示要按顺序执行的作业序列  组（group）表示可以并行执行的作业

使用 chain() 与使用多个嵌套具有相同的效果 on_done() ，但更具可读性。两者可以组合成一个图
	
	例如，我们可以对作业进行分组 [A]，从而阻止另一组作业 [B]。当且仅当组 [A] 的所有作业都执行时，组 [B] 的作业才会执行。
"""
def button_done(self):
    group_a = group(self.delayable().method_foo(), self.delayable().method_bar())
    group_b = group(self.delayable().method_baz(1), self.delayable().method_baz(2))
    chain(group_a, group_b).delay()
    self.write({"state": "done"})
    return True
```

### 主要开发者
扶程星云，sugar，yu.qian

小马/DTCloud

简介

发行版

贡献者

近期动态

小马/DTCloud .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

小马/DTCloud