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,
        project_id: Optional[str] = None,
        obs_id: Optional[str] = None,
        instrument: Literal['MSC', 'IFS', 'MCI', 'HSTDM', 'CPIC'] = 'MSC', 
        file_type: Optional[str] = None,
        obs_type: Optional[str] = None,
        object_name: Optional[str] = None,
        obstime: Optional[DateTimeTuple] = None,
        page: int = 1,
        limit: int = 0) -> Result:
    """
    根据给定的参数在DFS中搜索编排数据。
    
    Args:
        mode (Optional[str], optional): 观测模式，如CM1-1、OM1. Defaults to None.
        project_id (Optional[str], optional): 项目ID. Defaults to None.
        obs_id (Optional[str], optional): 观测ID，支持模糊搜索. Defaults to None.
        instrument (Optional[str], optional): 模块ID，如'MSC', 'IFS'. Defaults to None.
        file_type (Optional[str], optional): 文件类型，如'SCI'. Defaults to None.
        obs_type (Optional[str], optional): 观测类型，如主巡天宽场、TOO观测、定标星场等等. Defaults to None.
        object_name (Optional[str], optional): 目标名称. 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,
        'project_id': project_id,
        'obs_id': obs_id,
        'instrument': instrument,
        'file_type': file_type,
        'obs_type': obs_type,
        'object_name': object_name,
        '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: 查询结果。
    
    """
    return request.get(f"/api/plan/{opid}")

def write_file(local_file: Union[IO, str], **kwargs) -> Result:
    """
    将本地json文件或json数据流写入DFS中。
    
    Args:
        local_file (str]): 文件路径。
        **kwargs: 额外的关键字参数，这些参数将传递给DFS。
    
    Returns:
        Result: 操作的结果对象，包含操作是否成功以及相关的错误信息，成功返回数据对象。
    
    """    
    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/plan/file", local_file, kwargs)
    return request.post_bytesio("/api/plan/file", local_file, kwargs)

def new(data: dict) -> Result:
    """
    新建编排数据
    
    Args:
        data (dict): 编排数据的字典表示，如：{'id': 394, 'opid': 'xxx', 'instrument': 'MSC', ...}
    
    Returns:
        Result: 成功后，Result.data为写入记录，失败message为失败原因。
    
    """
    return request.post("/api/plan/new", data)