diff --git a/src/main/java/com/kamco/cd/kamcoback/inference/dto/InferenceResultDto.java b/src/main/java/com/kamco/cd/kamcoback/inference/dto/InferenceResultDto.java index fecbb7e1..6d6970cb 100644 --- a/src/main/java/com/kamco/cd/kamcoback/inference/dto/InferenceResultDto.java +++ b/src/main/java/com/kamco/cd/kamcoback/inference/dto/InferenceResultDto.java @@ -458,5 +458,9 @@ public class InferenceResultDto { private ZonedDateTime modelEndDttm; private Long updateUid; private String runningModelType; + private Integer pendingJobs; + private Integer runningJobs; + private Integer completedJobs; + private Integer failedJobs; } } diff --git a/src/main/java/com/kamco/cd/kamcoback/inference/service/InferenceResultService.java b/src/main/java/com/kamco/cd/kamcoback/inference/service/InferenceResultService.java index 19ba493c..8cff1043 100644 --- a/src/main/java/com/kamco/cd/kamcoback/inference/service/InferenceResultService.java +++ b/src/main/java/com/kamco/cd/kamcoback/inference/service/InferenceResultService.java @@ -141,6 +141,7 @@ public class InferenceResultService { throw new CustomApiException("NOT_FOUND_COMPARE_YEAR", HttpStatus.NOT_FOUND); } + // 사용할 영상파일 년도 기록 및 추론에 포함되는지 설정 for (MngListDto target : targetList) { for (Map map : totalNumList) { if (target.getMapSheetNum().equals(map.get("mapSheetNum").toString())) { @@ -272,14 +273,14 @@ public class InferenceResultService { throw new CustomApiException("BAD_REQUEST", HttpStatus.BAD_REQUEST); } - // 1) 요청 로그 (debug 권장) + // 1) 요청 로그 try { log.debug("Inference request dto={}", objectMapper.writeValueAsString(dto)); } catch (JsonProcessingException e) { log.warn("Failed to serialize inference dto", e); } - // 2) local 환경 임시 처리 (NPE 방어) + // 2) local 환경 임시 처리 if ("local".equals(profile)) { if (dto.getPred_requests_areas() == null) { throw new IllegalStateException("pred_requests_areas is null"); diff --git a/src/main/java/com/kamco/cd/kamcoback/postgres/core/InferenceResultCoreService.java b/src/main/java/com/kamco/cd/kamcoback/postgres/core/InferenceResultCoreService.java index 1f1489ba..c68a6276 100644 --- a/src/main/java/com/kamco/cd/kamcoback/postgres/core/InferenceResultCoreService.java +++ b/src/main/java/com/kamco/cd/kamcoback/postgres/core/InferenceResultCoreService.java @@ -30,6 +30,7 @@ import java.time.ZonedDateTime; import java.util.ArrayList; import java.util.List; import java.util.UUID; +import java.util.function.Consumer; import lombok.RequiredArgsConstructor; import lombok.extern.log4j.Log4j2; import org.springframework.dao.DataAccessException; @@ -235,91 +236,95 @@ public class InferenceResultCoreService { .toList(); } + /** + * learn 테이블 update + * + * @param request 추론 실행 정보 + */ public void update(SaveInferenceAiDto request) { MapSheetLearnEntity entity = mapSheetLearnRepository .getInferenceResultByUuid(request.getUuid()) - .orElseThrow(() -> new EntityNotFoundException()); + .orElseThrow(EntityNotFoundException::new); + // M1/M2/M3 영역 업데이트 if (request.getType() != null) { - switch (request.getType()) { - case "M1" -> { - if (request.getBatchId() != null) { - entity.setM1ModelBatchId(request.getBatchId()); - } - - if (request.getModelStartDttm() != null) { - entity.setM1ModelStartDttm(request.getModelStartDttm()); - } - - if (request.getModelEndDttm() != null) { - entity.setM1ModelEndDttm(request.getModelEndDttm()); - } - } - case "M2" -> { - if (request.getBatchId() != null) { - entity.setM2ModelBatchId(request.getBatchId()); - } - - if (request.getModelStartDttm() != null) { - entity.setM2ModelStartDttm(request.getModelStartDttm()); - } - - if (request.getModelEndDttm() != null) { - entity.setM2ModelEndDttm(request.getModelEndDttm()); - } - } - case "M3" -> { - if (request.getBatchId() != null) { - entity.setM3ModelBatchId(request.getBatchId()); - } - - if (request.getModelStartDttm() != null) { - entity.setM3ModelStartDttm(request.getModelStartDttm()); - } - - if (request.getModelEndDttm() != null) { - entity.setM3ModelEndDttm(request.getModelEndDttm()); - } - } - } + applyModelUpdate(entity, request); } - if (request.getRunningModelType() != null) { - entity.setRunningModelType(request.getRunningModelType()); - } - - if (request.getInferStartDttm() != null) { - entity.setInferStartDttm(request.getInferStartDttm()); - } - - if (request.getInferEndDttm() != null) { - entity.setInferEndDttm(request.getInferEndDttm()); - } - - if (request.getModelComparePath() != null) { - entity.setModelComparePath(request.getModelComparePath()); - } - - if (request.getModelTargetPath() != null) { - entity.setModelTargetPath(request.getModelTargetPath()); - } - - if (request.getDetectEndCnt() != null) { - entity.setDetectEndCnt(request.getDetectEndCnt()); - } - - if (request.getStatus() != null) { - entity.setStatus(request.getStatus()); - } - - if (request.getUpdateUid() != null) { - entity.setUpdatedUid(request.getUpdateUid()); - } + // 공통 영역 업데이트 + applyIfNotNull(request.getRunningModelType(), entity::setRunningModelType); + applyIfNotNull(request.getInferStartDttm(), entity::setInferStartDttm); + applyIfNotNull(request.getInferEndDttm(), entity::setInferEndDttm); + applyIfNotNull(request.getModelComparePath(), entity::setModelComparePath); + applyIfNotNull(request.getModelTargetPath(), entity::setModelTargetPath); + applyIfNotNull(request.getDetectEndCnt(), entity::setDetectEndCnt); + applyIfNotNull(request.getStatus(), entity::setStatus); + applyIfNotNull(request.getUpdateUid(), entity::setUpdatedUid); entity.setUpdatedDttm(ZonedDateTime.now()); } + private void applyModelUpdate(MapSheetLearnEntity entity, SaveInferenceAiDto request) { + switch (request.getType()) { + case "M1" -> + applyModelFields( + request, + entity::setM1ModelBatchId, + entity::setM1ModelStartDttm, + entity::setM1ModelEndDttm, + entity::setM1PendingJobs, + entity::setM1RunningJobs, + entity::setM1CompletedJobs, + entity::setM1FailedJobs); + case "M2" -> + applyModelFields( + request, + entity::setM2ModelBatchId, + entity::setM2ModelStartDttm, + entity::setM2ModelEndDttm, + entity::setM2PendingJobs, + entity::setM2RunningJobs, + entity::setM2CompletedJobs, + entity::setM2FailedJobs); + case "M3" -> + applyModelFields( + request, + entity::setM3ModelBatchId, + entity::setM3ModelStartDttm, + entity::setM3ModelEndDttm, + entity::setM3PendingJobs, + entity::setM3RunningJobs, + entity::setM3CompletedJobs, + entity::setM3FailedJobs); + default -> throw new IllegalArgumentException("Unknown type: " + request.getType()); + } + } + + private void applyModelFields( + SaveInferenceAiDto request, + Consumer setBatchId, + Consumer setStart, + Consumer setEnd, + Consumer setPending, + Consumer setRunning, + Consumer setCompleted, + Consumer setFailed) { + applyIfNotNull(request.getBatchId(), setBatchId); + applyIfNotNull(request.getModelStartDttm(), setStart); + applyIfNotNull(request.getModelEndDttm(), setEnd); + applyIfNotNull(request.getPendingJobs(), setPending); + applyIfNotNull(request.getRunningJobs(), setRunning); + applyIfNotNull(request.getCompletedJobs(), setCompleted); + applyIfNotNull(request.getFailedJobs(), setFailed); + } + + private void applyIfNotNull(T value, Consumer setter) { + if (value != null) { + setter.accept(value); + } + } + public List getInferenceServerStatusList() { return mapSheetLearnRepository.getInferenceServerStatusList(); } @@ -346,7 +351,14 @@ public class InferenceResultCoreService { return inferenceBatchSheet; } - public InferenceProgressDto getInferenceAiResultById(Long id, String type, UUID modelUuid) { + /** + * 추론 실행 api 파라미터 조회 + * + * @param id + * @param modelUuid + * @return + */ + public InferenceProgressDto getInferenceAiResultById(Long id, UUID modelUuid) { return mapSheetLearnRepository.getInferenceAiResultById(id, modelUuid); } diff --git a/src/main/java/com/kamco/cd/kamcoback/scheduler/service/MapSheetInferenceJobService.java b/src/main/java/com/kamco/cd/kamcoback/scheduler/service/MapSheetInferenceJobService.java index 36ad1059..42138051 100644 --- a/src/main/java/com/kamco/cd/kamcoback/scheduler/service/MapSheetInferenceJobService.java +++ b/src/main/java/com/kamco/cd/kamcoback/scheduler/service/MapSheetInferenceJobService.java @@ -11,7 +11,6 @@ import com.kamco.cd.kamcoback.inference.dto.InferenceProgressDto; import com.kamco.cd.kamcoback.inference.dto.InferenceResultDto.SaveInferenceAiDto; import com.kamco.cd.kamcoback.inference.dto.InferenceResultDto.Status; import com.kamco.cd.kamcoback.inference.dto.InferenceSendDto; -import com.kamco.cd.kamcoback.inference.dto.InferenceSendDto.pred_requests_areas; import com.kamco.cd.kamcoback.postgres.core.InferenceResultCoreService; import com.kamco.cd.kamcoback.scheduler.dto.JobStatusDto; import jakarta.transaction.Transactional; @@ -52,113 +51,226 @@ public class MapSheetInferenceJobService { @Scheduled(fixedDelay = 60_000) @Transactional public void runBatch() { - - if ("local".equalsIgnoreCase(profile)) { + if (isLocalProfile()) { return; } try { - InferenceBatchSheet batchSheet = - inferenceResultCoreService.getInferenceResultByStatus(Status.IN_PROGRESS.getId()); - - if (batchSheet == null) { + // 진행중 배치 조회 + InferenceBatchSheet sheet = findInProgressSheet(); + if (sheet == null) { return; } - HttpHeaders headers = new HttpHeaders(); - headers.setContentType(MediaType.APPLICATION_JSON); - headers.setAccept(List.of(MediaType.APPLICATION_JSON)); - - Long batchId = 0L; - - if (batchSheet.getM3BatchId() != null) { - batchId = batchSheet.getM3BatchId(); - } else if (batchSheet.getM2BatchId() != null) { - batchId = batchSheet.getM2BatchId(); - } else if (batchSheet.getM1BatchId() != null) { - batchId = batchSheet.getM1BatchId(); - } - - if (batchId == 0L) { + // 배치 아이디 가져오기 + Long batchId = resolveBatchId(sheet); + if (batchId == null || batchId == 0L) { return; } - String url = batchUrl + "/" + batchId; - - ExternalCallResult result = - externalHttpClient.call(url, HttpMethod.GET, null, headers, String.class); - - int status = result.statusCode(); - if (status < 200 || status >= 300) { + // 추론실행 상태 정보 가져오기 + JobStatusDto job = fetchJobStatus(batchId); + if (job == null) { return; } - String json = result.body(); - JobStatusDto dto = objectMapper.readValue(json, JobStatusDto.class); - - int totalJobs = dto.getTotalJobs(); - int completedJobs = dto.getCompletedJobs(); - int failedJobs = dto.getFailedJobs(); - - // 성공, 실패 값 더해서 total 과 같으면 완료 - String inferStatus = setStatus(totalJobs, completedJobs, failedJobs); - - if ("COMPLETED".equals(inferStatus)) { - String type = batchSheet.getRunningModelType(); - - if (type.equals("M1")) { - // M1 완료되었으면 M2 실행 - startInference( - batchSheet.getId(), batchSheet.getUuid(), "M2", batchSheet.getM2ModelUuid()); - // 종료시간 - updateProcessingEndTimeByModel(batchSheet.getUuid(), ZonedDateTime.now(), "M1"); - } else if (type.equals("M2")) { - // M2 완료되었으면 M3 실행 - startInference( - batchSheet.getId(), batchSheet.getUuid(), "M3", batchSheet.getM3ModelUuid()); - // 종료시간 - updateProcessingEndTimeByModel(batchSheet.getUuid(), ZonedDateTime.now(), "M2"); - } else if (type.equals("M3")) { - // 완료 - SaveInferenceAiDto saveInferenceAiDto = new SaveInferenceAiDto(); - saveInferenceAiDto.setUuid(batchSheet.getUuid()); - saveInferenceAiDto.setStatus(Status.END.getId()); - saveInferenceAiDto.setInferEndDttm(ZonedDateTime.now()); - saveInferenceAiDto.setType(type); - inferenceResultCoreService.update(saveInferenceAiDto); - // 종료시간 - updateProcessingEndTimeByModel(batchSheet.getUuid(), ZonedDateTime.now(), "M3"); - } + if (isCompleted(job)) { + // 완료 처리 + onCompleted(sheet, job); } else { - SaveInferenceAiDto saveInferenceAiDto = new SaveInferenceAiDto(); - saveInferenceAiDto.setUuid(batchSheet.getUuid()); - saveInferenceAiDto.setStatus(Status.IN_PROGRESS.getId()); - saveInferenceAiDto.setDetectEndCnt((long) (completedJobs + failedJobs)); - inferenceResultCoreService.update(saveInferenceAiDto); + // 진행중 처리 + onProcessing(sheet, job); } } catch (JsonProcessingException e) { - Thread.currentThread().interrupt(); - log.error("배치 중 인터럽트 발생", e); + // JSON 파싱 오류는 interrupt 대상 아님 + log.error("배치 중 JSON 파싱 오류", e); + } catch (Exception e) { + log.error("배치 처리 중 예외", e); } } - private void startInference(Long id, UUID uuid, String type, UUID modelUuid) { + /** + * 진행중 배치 조회 + * + * @return + */ + private InferenceBatchSheet findInProgressSheet() { + return inferenceResultCoreService.getInferenceResultByStatus(Status.IN_PROGRESS.getId()); + } - InferenceProgressDto progressDto = - inferenceResultCoreService.getInferenceAiResultById(id, type, modelUuid); + /** + * batchId 결정 + * + * @param sheet + * @return + */ + private Long resolveBatchId(InferenceBatchSheet sheet) { + // M3 > M2 > M1 + if (sheet.getM3BatchId() != null) { + return sheet.getM3BatchId(); + } + if (sheet.getM2BatchId() != null) { + return sheet.getM2BatchId(); + } + if (sheet.getM1BatchId() != null) { + return sheet.getM1BatchId(); + } + return 0L; + } - String inferenceType = ""; + /** + * 추론실행 상태 정보 가져오기 + * + * @param batchId + * @return + * @throws JsonProcessingException + */ + private JobStatusDto fetchJobStatus(Long batchId) throws JsonProcessingException { + String url = batchUrl + "/" + batchId; - if (type.equals("M1")) { - inferenceType = "G1"; - } else if (type.equals("M2")) { - inferenceType = "G2"; - } else if (type.equals("M3")) { - inferenceType = "G3"; + ExternalCallResult result = + externalHttpClient.call(url, HttpMethod.GET, null, jsonHeaders(), String.class); + + int status = result.statusCode(); + if (status < 200 || status >= 300) { + return null; } - pred_requests_areas predRequestsAreas = new pred_requests_areas(); + return objectMapper.readValue(result.body(), JobStatusDto.class); + } + + private HttpHeaders jsonHeaders() { + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + headers.setAccept(List.of(MediaType.APPLICATION_JSON)); + return headers; + } + + /** + * 완료 판단 + * + * @param dto + * @return + */ + private boolean isCompleted(JobStatusDto dto) { + return dto.getTotalJobs() <= (dto.getCompletedJobs() + dto.getFailedJobs()); + } + + /** + * 완료 처리 + * + * @param sheet + * @param job + */ + private void onCompleted(InferenceBatchSheet sheet, JobStatusDto job) { + String currentType = sheet.getRunningModelType(); + ZonedDateTime now = ZonedDateTime.now(); + + // 현재 모델 종료 업데이트 + updateProcessingEndTimeByModel(job, sheet.getUuid(), now, currentType); + + // M3이면 전체 종료 + if ("M3".equals(currentType)) { + endAll(sheet, now); + return; + } + + // 다음 모델 실행 (M1->M2, M2->M3) + String nextType = nextModelType(currentType); + UUID nextModelUuid = resolveModelUuid(sheet, nextType); + + // 다음 모델 추론 실행 + startInference(sheet.getId(), sheet.getUuid(), nextType, nextModelUuid); + } + + /** + * 추론 종료 할때 update + * + * @param sheet + * @param now + */ + private void endAll(InferenceBatchSheet sheet, ZonedDateTime now) { + SaveInferenceAiDto save = new SaveInferenceAiDto(); + save.setUuid(sheet.getUuid()); + save.setStatus(Status.END.getId()); + save.setInferEndDttm(now); + save.setType("M3"); // 마지막 모델 기준 + inferenceResultCoreService.update(save); + } + + /** + * 다음 실행할 모델 타입 조회 + * + * @param currentType + * @return + */ + private String nextModelType(String currentType) { + if ("M1".equals(currentType)) { + return "M2"; + } + if ("M2".equals(currentType)) { + return "M3"; + } + throw new IllegalArgumentException("Unknown runningModelType: " + currentType); + } + + /** + * 모델 정보 UUID 가져오기 + * + * @param sheet + * @param type + * @return + */ + private UUID resolveModelUuid(InferenceBatchSheet sheet, String type) { + if ("M1".equals(type)) { + return sheet.getM1ModelUuid(); + } + if ("M2".equals(type)) { + return sheet.getM2ModelUuid(); + } + if ("M3".equals(type)) { + return sheet.getM3ModelUuid(); + } + throw new IllegalArgumentException("Unknown type: " + type); + } + + /** + * 진행중 처리 + * + * @param sheet + * @param job + */ + private void onProcessing(InferenceBatchSheet sheet, JobStatusDto job) { + SaveInferenceAiDto save = new SaveInferenceAiDto(); + save.setUuid(sheet.getUuid()); + save.setStatus(Status.IN_PROGRESS.getId()); + save.setPendingJobs(job.getPendingJobs()); + save.setRunningJobs(job.getRunningJobs()); + save.setCompletedJobs(job.getCompletedJobs()); + save.setFailedJobs(job.getFailedJobs()); + inferenceResultCoreService.update(save); + } + + /** + * 다음 모델 추론 실행 + * + * @param id + * @param uuid + * @param type + * @param modelUuid + */ + private void startInference(Long id, UUID uuid, String type, UUID modelUuid) { + + // 추론 실행 api 파라미터 조회 + InferenceProgressDto progressDto = + inferenceResultCoreService.getInferenceAiResultById(id, modelUuid); + + // ai 에 맞는 모델 명으로 변경 + String inferenceType = modelToInferenceType(type); + + InferenceSendDto.pred_requests_areas predRequestsAreas = + new InferenceSendDto.pred_requests_areas(); predRequestsAreas.setInput1_year(progressDto.getPred_requests_areas().getInput1_year()); predRequestsAreas.setInput2_year(progressDto.getPred_requests_areas().getInput2_year()); predRequestsAreas.setInput1_scene_path( @@ -178,7 +290,7 @@ public class MapSheetInferenceJobService { m.setCd_model_type(inferenceType); m.setPriority(progressDto.getPriority()); - // 추론 다음모델 실행 + // 추론 실행 api 호출 Long batchId = ensureAccepted(m); SaveInferenceAiDto saveInferenceAiDto = new SaveInferenceAiDto(); @@ -191,71 +303,109 @@ public class MapSheetInferenceJobService { inferenceResultCoreService.update(saveInferenceAiDto); } + /** + * ai 에 맞는 모델 명으로 변경 + * + * @param type 모델 타입 + * @return String + */ + private String modelToInferenceType(String type) { + if ("M1".equals(type)) { + return "G1"; + } + if ("M2".equals(type)) { + return "G2"; + } + if ("M3".equals(type)) { + return "G3"; + } + throw new IllegalArgumentException("Unknown type: " + type); + } + + /** + * api 호출 + * + * @param dto + * @return + */ private Long ensureAccepted(InferenceSendDto dto) { - log.info("dto null? {}", dto == null); - ObjectMapper om = new ObjectMapper(); - try { - log.info("dto json={}", om.writeValueAsString(dto)); - } catch (Exception e) { - log.error(e.getMessage()); + if (dto == null) { + log.warn("not InferenceSendDto dto"); + throw new CustomApiException("BAD_REQUEST", HttpStatus.BAD_REQUEST); } - HttpHeaders headers = new HttpHeaders(); - headers.setContentType(MediaType.APPLICATION_JSON); - headers.setAccept(List.of(MediaType.APPLICATION_JSON)); + // 1) 요청 로그 + try { + log.debug("Inference request dto={}", objectMapper.writeValueAsString(dto)); + } catch (JsonProcessingException e) { + log.warn("Failed to serialize inference dto", e); + } - // TODO 추후 삭제 + // 2) local 환경 임시 처리 if ("local".equals(profile)) { if (dto.getPred_requests_areas() == null) { - dto.setPred_requests_areas(new InferenceSendDto.pred_requests_areas()); + throw new IllegalStateException("pred_requests_areas is null"); } - dto.getPred_requests_areas().setInput1_scene_path("/kamco-nfs/requests/2023_local.geojson"); dto.getPred_requests_areas().setInput2_scene_path("/kamco-nfs/requests/2024_local.geojson"); } + // 3) HTTP 호출 + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + headers.setAccept(List.of(MediaType.APPLICATION_JSON)); + ExternalCallResult result = externalHttpClient.call(inferenceUrl, HttpMethod.POST, dto, headers, String.class); - int status = result.statusCode(); - String body = result.body(); - - if (status < 200 || status >= 300) { + if (result.statusCode() < 200 || result.statusCode() >= 300) { + log.error("Inference API failed. status={}, body={}", result.statusCode(), result.body()); throw new CustomApiException("BAD_GATEWAY", HttpStatus.BAD_GATEWAY); } - Long batchId = 0L; - + // 4) 응답 파싱 try { List> list = - om.readValue(body, new TypeReference>>() {}); + objectMapper.readValue(result.body(), new TypeReference<>() {}); - Integer batchIdInt = (Integer) list.get(0).get("batch_id"); - batchId = batchIdInt.longValue(); if (list.isEmpty()) { throw new IllegalStateException("Inference response is empty"); } - } catch (Exception e) { - log.error(e.getMessage()); - } + Object batchIdObj = list.get(0).get("batch_id"); + if (batchIdObj == null) { + throw new IllegalStateException("batch_id not found in response"); + } - return batchId; + return Long.valueOf(batchIdObj.toString()); + + } catch (Exception e) { + log.error("Failed to parse inference response. body={}", result.body(), e); + throw new CustomApiException("INVALID_INFERENCE_RESPONSE", HttpStatus.BAD_GATEWAY); + } } - private void updateProcessingEndTimeByModel(UUID uuid, ZonedDateTime dateTime, String type) { + /** + * 실행중인 profile + * + * @return + */ + private boolean isLocalProfile() { + return "local".equalsIgnoreCase(profile); + } + + /** 모델별 추론 종료 update */ + private void updateProcessingEndTimeByModel( + JobStatusDto dto, UUID uuid, ZonedDateTime dateTime, String type) { SaveInferenceAiDto saveInferenceAiDto = new SaveInferenceAiDto(); saveInferenceAiDto.setUuid(uuid); saveInferenceAiDto.setUpdateUid(0L); saveInferenceAiDto.setModelEndDttm(dateTime); saveInferenceAiDto.setType(type); + saveInferenceAiDto.setPendingJobs(dto.getPendingJobs()); + saveInferenceAiDto.setRunningJobs(dto.getRunningJobs()); + saveInferenceAiDto.setCompletedJobs(dto.getCompletedJobs()); + saveInferenceAiDto.setFailedJobs(dto.getFailedJobs()); inferenceResultCoreService.update(saveInferenceAiDto); } - - private String setStatus(int totalJobs, int completedJobs, int failedJobs) { - if (totalJobs <= (completedJobs + failedJobs)) { - return "COMPLETED"; - } - return "PROCESSING"; - } }