Skip to content
GitLab
An error occurred while loading the file. Please try again.
Switch branch/tag
Find file BlameHistoryPermalink
level1.py 10.27 KiB
Open in Web IDE
1 
from typing import Optional, Tuple, Literal, IO, Union
2 
from .common import request, Result, utils, constants
3 
import os
4
5 
DateTimeTuple = Tuple[str, str]
6
7 
def find(project_id: Optional[str] = None,
8 
        obs_id: Optional[str] = None,
9 
        level0_id: Optional[str] = None,
10 
        module_id: Literal['MSC', 'IFS', 'MCI', 'HSTDM', 'CPIC'] = 'MSC',
11 
        detector_no: Optional[str] = None,
12 
        data_type: Optional[str] = None,
13 
        filter: Optional[str] = None,
14 
        obs_time: Optional[DateTimeTuple] = None,
15 
        create_time: Optional[DateTimeTuple] = None,
16 
        qc1_status: Optional[int] = None,
17 
        prc_status: Optional[int] = None,
18 
        file_name: Optional[str] = None,
19 
        ra_obj: Optional[int] = None,
20 
        dec_obj: Optional[int] = None,
21 
        radius: Optional[float] = None,
22 
        object_name: Optional[str] = None,
23 
        rss_id: Optional[str] = None,
24 
        cube_id: Optional[str] = None,
25 
        dataset: Optional[str] = None,
26 
        batch_id: Optional[str] = None,
27 
        page: int = 1,
28 
        limit: int = 0) -> Result:
29 
    """
30 
    根据给定的参数搜索1级数据文件记录
31
32 
    Args:
33 
        project_id (Optional[str], optional): 项目ID. Defaults to None.
34 
        obs_id (Optional[str], optional): 观测ID. Defaults to None.
35 
        module_id (Optional[str], optional): 模块ID，如'MSC', 'IFS'. Defaults to None.
36 
        detector_no (Optional[str], optional): 探测器编号. Defaults to None.
37 
        data_type (Optional[str], optional): 数据类型，如'csst-msc-l1-mbi'. Defaults to None.
38 
        filter (Optional[str], optional): 滤光片. Defaults to None.
39 
        obs_time (Optional[DateTimeTuple], optional): 观测时间范围. Defaults to None.
40 
        create_time (Optional[DateTimeTuple], optional): 创建时间范围. Defaults to None.
41 
        qc1_status (Optional[int], optional): QC1状态. Defaults to None.
42 
        prc_status (Optional[int], optional): 处理状态. Defaults to None.
43 
        file_name (Optional[str], optional): 文件名. Defaults to None.
44 
        ra_obj (Optional[int], optional): 目标赤经. Defaults to None.
45 
        dec_obj (Optional[int], optional): 目标赤纬. Defaults to None.
46 
        radius (Optional[float], optional): 搜索半径. Defaults to None.
47 
        object_name (Optional[str], optional): 天体名称. Defaults to None.
48 
        rss_id (Optional[str], optional): RSS ID (IFS) Defaults to None.
49 
        cube_id (Optional[str], optional): Cube ID (IFS). Defaults to None.
50 
        dataset (Optional[str], optional): 数据集名称. Defaults to None.
51 
        batch_id (Optional[str], optional): 批次ID. Defaults to None.
52 
        page (int, optional): 页码. Defaults to 1.
53 
        limit (int, optional): 每页数量. Defaults to 0.
54
55 
    Returns:
56 
        Result: 搜索结果对象.
57
58 
    """
59
60 
    params = {
61 
        'project_id': project_id,
62 
        'level0_id': level0_id,
63 
        'obs_id': obs_id,
64 
        'module_id': module_id,
65 
        'detector_no': detector_no,
66 
        'data_type': data_type,
67 
        'filter': filter,
68 
        'qc1_status': qc1_status,
69 
        'prc_status': prc_status,
70 
        'file_name': file_name,
71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140
'ra_obj': ra_obj, 'dec_obj': dec_obj, 'radius': radius, 'object_name': object_name, 'obs_time_start': None, 'obs_time_end': None, 'create_time_start': None, 'create_time_end': None, 'rss_id': rss_id, 'cube_id': cube_id, 'dataset': dataset, 'batch_id': batch_id, 'page': page, 'limit': limit, } if obs_time is not None: params['obs_time_start'], params['obs_time_end'] = obs_time utils.is_valid_datetime_format(params['obs_time_start']) or not utils.is_valid_datetime_format(params['obs_time_end']) if create_time is not None: params['create_time_start'], params['create_time_end'] = create_time utils.is_valid_datetime_format(params['create_time_start']) or utils.is_valid_datetime_format(params['create_time_end']) return request.post("/api/level1", params) def find_by_level1_id(level1_id: str) -> Result: """ 通过 level1 的 ID 查询1级数据 Args: level1_id (str): 1级数据的ID Returns: Result: 查询结果 """ return request.get(f"/api/level1/{level1_id}") def find_by_brick_id(brick_id: int) -> Result: """ 通过 brick 的 ID 查询1级数据 Args: brick_id (int): 天区ID Returns: Result: 查询结果 """ return request.get(f"/api/level1/brick/{brick_id}") def sls_find_by_qc1_status(qc1_status: int, limit: int = 1, batch_id: str = constants.DEFAULT_BATCH_ID) -> Result: return request.post(f"/api/level1/sls/qc1_status/{qc1_status}", {'limit': limit, 'batch_id': batch_id}) def update_qc1_status(level1_id: str, data_type: str, qc1_status: int, batch_id: str = constants.DEFAULT_BATCH_ID) -> Result: """ 更新1级数据的QC0状态 Args: level1_id (str): 1级数据的ID data_type (str): 文件类型 qc1_status (int): QC0状态 batch_id (str): 批次ID Returns: Result: 更新结果 """ return request.put(f"/api/level1/qc1_status/{level1_id}", {'data_type': data_type, 'qc1_status': qc1_status, 'batch_id': batch_id}) def update_prc_status(level1_id: str, data_type: str, prc_status: int, batch_id: str = constants.DEFAULT_BATCH_ID) -> Result:
141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210
""" 更新1级数据的处理状态 Args: level1_id (str): 1级数据的ID data_type (str): 文件类型 prc_status (int): 处理状态 batch_id (str): 批次ID Returns: Result: 操作结果 """ return request.put(f"/api/level1/prc_status/{level1_id}", {'data_type': data_type, 'prc_status': prc_status, 'batch_id': batch_id}) def write(local_file: Union[IO, str], module_id: Literal['MSC', 'IFS', 'MCI', 'HSTDM', 'CPIC'], level1_id: str, data_type: str, file_name: str, dag_id: str, pmapname: str, build: int, level0_id: Optional[str] = None, dataset: str = constants.DEFAULT_DATASET, batch_id: str = constants.DEFAULT_BATCH_ID, qc1_status: int = 0, **extra_kwargs) -> Result: ''' 将本地的1级数据文件写入到DFS中其他参数如rss_id, cube_id等，可通过extra_kwargs传入 Args: local_file (Union[IO, str]): 文件路径或文件对象 module_id ['MSC', 'IFS', 'MCI', 'HSTDM', 'CPIC']其中一个，代表: 模块ID level0_id (Optional[str]): 0级数据的ID默认为 None level1_id (str): 1级数据的ID data_type (str): 数据类型 file_name (str): 1级数据文件名 dag_id (str): 管线ID pmapname (str): CCDS pmap名称 build (int): 构建号 dataset (str): 数据集名称 batch_id (str): 批次ID qc1_status (int): QC1状态 **kwargs: 额外的关键字参数，这些参数将传递给DFS Returns: Result: 操作的结果对象，包含操作是否成功以及相关的错误信息，成功返回data为1级数据对象 ''' params = { 'module_id': module_id, 'level0_id': level0_id, 'level1_id': level1_id, 'data_type': data_type, 'file_name': file_name, 'dag_id': dag_id, 'pmapname': pmapname, 'build': build, 'dataset': dataset, 'batch_id': batch_id, 'qc1_status': qc1_status } if not dataset or not batch_id: raise ValueError("dataset and batch_id is required") params.update(extra_kwargs) if local_file is None: raise ValueError("local_file is required") if isinstance(local_file, str): if not os.path.exists(local_file): raise FileNotFoundError(local_file) return request.post_file("/api/level1/file", local_file, params)
211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280
return request.post_bytesio("/api/level1/file", local_file, params) def generate_prc_msg(module_id: Literal['MSC', 'IFS', 'MCI', 'HSTDM', 'CPIC'], level1_id: str, dataset: str = constants.DEFAULT_DATASET, batch_id: str = constants.DEFAULT_BATCH_ID) -> Result: """ 生成流水线的处理消息 Args: module_id (str): 模块ID level1_id (str): 1级数据的ID dag_id (str): 流水管线ID，默认为空字符串 dataset (str): 数据集 batch_id (str): 批次ID Returns: Result: 处理消息生成的结果，是否成功以及相关的错误信息 """ params = { 'dataset': dataset, 'batch_id': batch_id, 'level1_id': level1_id, } return request.put(f"/api/level1/prc/{module_id}", params) def process_list(level1_id: str) -> Result: """ 通过 level1 的 ID 查询1级数据处理记录 Args: level1_id (str): 1级数据的ID Returns: Result: 查询结果 """ return request.get(f"/api/level1/prc/{level1_id}") def add_process(level1_id: str, dag_id: str, dag_run_id: str, dataset: str = constants.DEFAULT_DATASET, batch_id: str = constants.DEFAULT_BATCH_ID, prc_time: str = utils.get_current_time(), prc_status: int = -1024, prc_module: str = "", message: str = "") -> Result: """ 添加1级数据处理记录 Args: level1_id (str): 1级数据的ID dag_id (str): 管线ID dag_run_id (str): 运行ID dataset (str): 数据集 batch_id (str): 批次ID prc_time (str): 处理时间,格式为"YYYY-MM-DD HH:MM:SS" prc_status (int): 处理状态 prc_module (str): 处理模块 message (str): 处理消息 Returns: Result: 成功后，Result.data为写入记录，失败message为失败原因 """ params = { 'level1_id': level1_id,
281 282 283 284 285 286 287 288 289 290 291
'dag_id': dag_id, 'dag_run_id': dag_run_id, 'dataset': dataset, 'batch_id': batch_id, 'prc_time': prc_time, 'prc_status': prc_status, 'prc_module': prc_module, 'message': message, } utils.is_valid_datetime_format(prc_time) return request.post("/api/level1/prc", params)