MVPTeam/test
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request '[KC-103] 추론 실행 배치 수정' (#229) from feat/infer_dev_260107 into develop

Browse Source 
Reviewed-on: https://kamco.gitea.gs.dabeeo.com/dabeeo/kamco-dabeeo-backoffice/pulls/229
This commit is contained in:
teddy.kim
2026-01-14 15:26:23 +09:00
parent 9eb06d2c2e 1ae69e5665
commit 3daf150d82
4 changed files with 360 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/com/kamco/cd/kamcoback/inference/dto/InferenceResultDto.java
View File

@@ -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;
}
}

5
src/main/java/com/kamco/cd/kamcoback/inference/service/InferenceResultService.java
View File

@@ -141,6 +141,7 @@ public class InferenceResultService {
throw new CustomApiException("NOT_FOUND_COMPARE_YEAR", HttpStatus.NOT_FOUND);
}
// 사용할 영상파일 년도 기록 및 추론에 포함되는지 설정
for (MngListDto target : targetList) {
for (Map<String, Object> 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");

160
src/main/java/com/kamco/cd/kamcoback/postgres/core/InferenceResultCoreService.java
View File

@@ -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<Long> setBatchId,
Consumer<ZonedDateTime> setStart,
Consumer<ZonedDateTime> setEnd,
Consumer<Integer> setPending,
Consumer<Integer> setRunning,
Consumer<Integer> setCompleted,
Consumer<Integer> 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 <T> void applyIfNotNull(T value, Consumer<T> setter) {
if (value != null) {
setter.accept(value);
}
}
public List<InferenceServerStatusDto> 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);
}

384
src/main/java/com/kamco/cd/kamcoback/scheduler/service/MapSheetInferenceJobService.java
View File

@@ -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<String> 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<String> 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<String> 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<Map<String, Object>> list =
om.readValue(body, new TypeReference<List<Map<String, Object>>>() {});
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";
}
}