将结果写回ymir平台#

def write_monitor_logger(percent: float,
                         state: ExecutorState = ExecutorState.ES_RUNNING,
                         return_code: ExecutorReturnCode = ExecutorReturnCode.RC_EXEC_NO_ERROR) -> None:
    """write the execution progress percentage of the task to monitor file.

    Parameters
    ----------
    percent : float
         the execution progress percentage of the task, range in [0, 1]
    """
    env_config = env.get_current_env()
    with open(env_config.output.monitor_file, 'w') as f:
        f.write(f"{env_config.task_id}\t{time.time()}\t{percent:.2f}\t{state}\t{int(return_code)}\n")

@versionadded(
    version="2.1.0",
    reason="support segmentation metrics and custom metrics",
)
def write_training_result(stage_name: str,
                          files: List[str],
                          evaluation_result: Dict[str, Union[float, int]] = {},
                          timestamp: Optional[int] = None,
                          attachments: Optional[Dict[str, List[str]]] = None,
                          evaluate_config: Optional[dict] = None) -> None:
    """write training result to training result file

    if exist result in training result file, new result will append to file
    """
    write_model_stage(stage_name=stage_name,
                      files=files,
                      evaluation_result=evaluation_result,
                      timestamp=timestamp,
                      attachments=attachments,
                      evaluate_config=evaluate_config)

def write_mining_result(mining_result: List[Tuple[str, float]]) -> None:
    """write mining result to mining_result_file

    Parameters
    ----------
    mining_result : List[Tuple[str, float]]
        - image_path: str
        - score: float
    """
    # sort desc by score
    sorted_mining_result = sorted(mining_result, reverse=True, key=(lambda v: v[1]))

    env_config = env.get_current_env()
    with open(env_config.output.mining_result_file, "w") as f:
        for img_path, score in sorted_mining_result:
            f.write(f"{img_path}\t{score}\n")

@versionchanged(
    version="2.0.0",
    reason="support detection and segmentation infer result",
)
def write_infer_result(infer_result: Union[Dict, Dict[str, List[Annotation]]],
                       algorithm: Optional[str] = 'detection') -> None:
    """
    supported_algorithms = ['classification', 'detection', 'segmentation']
    for detection infer result, use ymir format
    for segmentation infer result, write coco-format to infer_result_file directly
    """
    env_config = env.get_current_env()
    supported_algorithms = ['classification', 'detection', 'segmentation']
    assert algorithm in supported_algorithms, f'unknown {algorithm}, not in {supported_algorithms}'

    if algorithm == 'detection':
        protocol_version = env_config.protocol_version
        # from ymir1.3.0, keyword change from annotations to boxes
        if Version(protocol_version) >= Version("1.0.0"):
            keyword = "boxes"
        else:
            keyword = "annotations"

        detection_result = {}
        for asset_path, annotations in infer_result.items():
            asset_basename = os.path.basename(asset_path)
            detection_result[asset_basename] = {keyword: [annotation.dict() for annotation in annotations]}

        result = {"detection": detection_result}
        with open(env_config.output.infer_result_file, "w") as f:
            f.write(json.dumps(result))
    elif algorithm == 'segmentation':
        with open(env_config.output.infer_result_file, 'w') as f:
            f.write(json.dumps(infer_result))
    else:
        raise Exception(f'not implement {algorithm} infer result')