plan.py

import os
from typing import Optional, IO, Tuple, Literal, Union
from .common import request, Result

DateTimeTuple = Tuple[str, str]

def find(mode: Optional[str] = None,
        obsid: Optional[str] = None,
        backend: Literal['MSC', 'IFS', 'MCI', 'HSTDM', 'CPIC'] = 'MSC', 
        obstype: Optional[str] = None,
        obstime: Optional[DateTimeTuple] = None,
        page: int = 1,
        limit: int = 0) -> Result:
    """
    根据给定的参数在DFS中搜索编排数据。
    
    Args:
        mode (Optional[str], optional): 观测模式. Defaults to None.
        obsid (Optional[str], optional): 观测ID，支持模糊搜索. Defaults to None.
        backend (Optional[str], optional): 模块ID，如'MSC', 'IFS'. Defaults to None.
        obstype (Optional[str], optional): 观测类型，如'SCIE'. Defaults to None.
        obstime (Optional[DateTimeTuple], optional): 观测时间范围. Defaults to None.
        page (int, optional): 页码. Defaults to 1.
        limit (int, optional): 每页数量. Defaults to 0，不限制.
    
    Returns:
        Result: 搜索结果对象.
    
    """

    params = {
        'mode': mode,
        'obsid': obsid,
        'backend': backend,
        'obstype': obstype,
        'obs_time_start': None,
        'obs_time_end': None,
        'create_time_start': None,
        'create_time_end': None,
        'page': page,
        'limit': limit,
    }
    
    if obstime is not None:
        params['obs_time_start'], params['obs_time_end'] = obstime
    
    return request.post("/api/plan", params)

def get_by_id(_id: int) -> Result:
    """
    通过计划ID获取编排数据。
    
    Args:
        _id (int): 计划ID。
    
    Returns:
        Result: 查询结果。
    
    """
    return request.get(f"/api/plan/id/{_id}")

def find_by_opid(opid: str) -> Result:
    """
    通过 plan 的 opid 查询编排数据。
    
    Args:
        opid (str): opid。
    
    Returns:
        Result: 查询结果。