跳到主要内容

任务类

Class: Job​

CloudPSS 任务类。

job.id​

任务ID。

job.args​

任务参数。

job.job_status​

任务状态。

job.createTime​

任务创建时间。

job.startTime​

任务开始时间。

job.endTime​

任务结束时间。

job.context​

任务执行上下文信息。

job.input​

任务执行输入流 ID，用于发送数据给当前任务。

job.output​

任务执行输出流 ID，用于接收任务执行的输出结果。

job.result​

任务结果，用于兼容旧版本的使用方法。

Job.fetch(id)​

  • 静态方法
  • id: String 任务的 id
  • Returns: Job 返回当前任务实例

获取指定 id 的 job 信息，如果 id 不存在直接抛异常。

job = Job.fetch('job/1')

Job.fetchMany(name=None, cursor=[], pageSize=10)​

  • 静态方法
  • name: String 查询名称，模糊查询
  • cursor: List 分页游标
  • pageSize: Number 分页大小
  • Returns: Dict 按分页信息返回 job 列表

获取用户有权限的 job 列表。

data = Job.fetchMany()
"""
    返回的数据格式 {
        'cursor': ['1705377106000'], # 当前返回数据游标
        'total': 4893,  # 任务总数
        'count': 10,    # 当前返回数据量
        'items':[
            {'id': 'job/1', 'name': 'job1', 'description': 'job1'}
            ....
        ]
    }
"""

# 查询下个游标后的数据 
cursor = data['cursor']
data2 = Job.fetchMany(cursor=cursor)

# 模糊查询 
data2 = Job.fetchMany(name='job1')
print('获取任务列表', data)

Job.create(revisionHash, job, config, name=None, rid="", policy=None, **kwargs)​

  • 静态方法
  • revisionHash: String 项目版本的 hash
  • job: Dict 计算方案
  • config: Dict 参数方案
  • name: String 任务名称
  • rid: String 任务的 rid
  • policy: Dict 仿真策略
  • kwargs: Dict 仿真参数
  • Returns: Job 返回一个运行实例

创建一个运行任务。

job = Job.create(revisionHash, job, config, name=None, rid="", policy=None, **kwargs)

Job.dump(job, file, format="yaml", compress="gzip")​

  • 静态方法
  • job: Job 任务实例
  • file: String 保存文件路径
  • format: String 保存文件格式， 默认为 yaml 格式
  • compress: String 保存文件压缩格式，默认为 gzip 格式

下载 job。

Job.dump(job, file, format="yaml", compress="gzip")

Job.load(file, format="yaml")​

  • 静态方法
  • file: String 保存文件路径
  • format: String 保存文件格式， 默认为 yaml 格式
  • Returns: Job 返回一个仿真任务

加载 job。

Job.load(file, format="yaml")

job.read(receiver=None, **kwargs)​

接收当前运行实例的输出。

Job.read(receiver=None, dev=False, **kwargs)

job.write(sender=None, **kwargs)​

发送为当前运行实例输入。

Job.write(sender=None, dev=False, **kwargs)

job.view(viewType)​

  • 实例方法
  • viewType: Result 结果类，分为：
    • EMTView 电磁暂态结果
    • PowerFlowView 潮流结果，
    • IESView 综合能源结果
    • IESLabSimulationView 综合能源建模仿真结果
    • IESLabTypicalDayView 综合能源典型日结果
  • Returns: Result 返回一个结果实例

获取当前运行实例的输出。

job = Job.fetch(id)
view = job.view(EMTView)

job.status()​

  • Returns: Number 返回任务状态
    • 0: 运行中
    • 1: 运行完成
    • 2: 运行失败

任务状态（旧版本兼容）。

import os
import time
import cloudpss

if __name__ == "__main__":
    cloudpss.setToken(token)     ## 输入 token
    os.environ['CLOUDPSS_API_URL'] = 'https://cloudpss.net/'
    job = cloudpss.Job.fetch(id)
    st = time.time()
    while not job.status():
        time.sleep(0.1)
    print('计算完成', time.time() - st)

job.abort(timeout=3)​

  • timeout: Number 超时 3 秒

中断当前运行实例。

job = Job.fetch(id)
job.abort(3)