修复多通道温度覆盖与实时阈值异常漏报
按绑定角色保存采集器测量,每五秒独立判定超限,避免采样入库周期阻断预警。控制台提醒根据当前观测生成,与异常处理记录独立,并共用测试数据归属边界。补充通道覆盖、采样间隔、历史处理与时间边界回归验证。
| | |
| | | package com.battery.business.service.impl; |
| | | |
| | | import java.time.Instant; |
| | | import java.math.BigDecimal; |
| | | import java.util.ArrayList; |
| | | import java.util.LinkedHashMap; |
| | | import java.util.List; |
| | |
| | | } |
| | | List<Map<String, Object>> exceptions = new ArrayList<>(); |
| | | java.util.Set<String> offlineSources = new java.util.HashSet<>(); |
| | | // 完成或暂停后只保留历史异常事实,实时控制台不再展示开放预警。 |
| | | if (receiving && mapper != null) { |
| | | mapper.selectWarningExceptions(testId, "OPEN").forEach(item -> { |
| | | // 离线待办的处理状态不代表当前设备状态,实时提醒只由下方当前来源生成。 |
| | | if ("DEVICE_OFFLINE".equals(item.getExceptionType())) return; |
| | | Map<String, Object> warning = new LinkedHashMap<>(); |
| | | warning.put("exceptionId", item.getExceptionId()); |
| | | warning.put("exceptionType", item.getExceptionType()); |
| | | warning.put("sourceType", item.getSourceType()); |
| | | warning.put("sourceIdentity", item.getSourceIdentity()); |
| | | warning.put("metricCode", item.getMetricCode()); |
| | | warning.put("value", item.getAbnormalValue()); |
| | | warning.put("unit", item.getUnit()); |
| | | warning.put("status", item.getExceptionStatus()); |
| | | warning.put("message", "测试预警异常"); |
| | | exceptions.add(warning); |
| | | }); |
| | | } |
| | | // 实时提醒只反映当前设备和测量事实,异常管理待办是否处理不参与展示判定。 |
| | | java.util.Set<String> thresholdSources = new java.util.HashSet<>(); |
| | | if (receiving) { |
| | | for (LiveSource source : sources) { |
| | | if ("ONLINE".equals(source.deviceStatus())) appendThresholdWarnings(test, source, exceptions, thresholdSources); |
| | | if (!"OFFLINE".equals(source.deviceStatus())) continue; |
| | | String identity = offlineIdentity(test, source); |
| | | if (!offlineSources.add(source.sourceType() + ":" + identity)) continue; |
| | |
| | | String testStatus = test.getTestStatus(); |
| | | TestPlanSnapshot testPlan = detectorTestPlan(test); |
| | | return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now().toEpochMilli(), sources, exceptions, testPlan); |
| | | } |
| | | |
| | | private void appendThresholdWarnings(TTest test, LiveSource source, List<Map<String, Object>> warnings, |
| | | java.util.Set<String> seen) |
| | | { |
| | | var snapshot = source.latestSnapshot(); |
| | | if ("CABINET_CHANNEL".equals(source.sourceType())) { |
| | | if (!receivedDuringTest(test, snapshot.get("deviceTime"), snapshot.get("observedAt"))) return; |
| | | String identity = offlineIdentity(test, source); |
| | | for (String[] field : List.of(new String[] { "voltageV", "VOLTAGE", "V" }, |
| | | new String[] { "currentA", "CURRENT", "A" }, new String[] { "capacityAh", "CAPACITY", "Ah" }, |
| | | new String[] { "energyWh", "ENERGY", "Wh" })) |
| | | appendThresholdValue(test, source.sourceType(), identity, field[1], snapshot.get(field[0]), field[2], warnings, seen); |
| | | } |
| | | else if (snapshot.get("measurements") instanceof List<?> measurements) { |
| | | for (Object item : measurements) if (item instanceof Map<?, ?> measurement |
| | | && receivedDuringTest(test, measurement.get("deviceTime"), measurement.get("serverReceivedAt"))) { |
| | | appendThresholdValue(test, source.sourceType(), source.sourceId() + ":" + measurement.get("channelCode"), |
| | | String.valueOf(measurement.get("metricCode")), measurement.get("value"), |
| | | String.valueOf(measurement.get("unit")), warnings, seen); |
| | | } |
| | | } |
| | | } |
| | | |
| | | private boolean receivedDuringTest(TTest test, Object deviceTime, Object observedAt) |
| | | { |
| | | return TestDataReceptionServiceImpl.accepted(test, |
| | | deviceTime instanceof Number time ? Instant.ofEpochMilli(time.longValue()) : null, |
| | | observedAt instanceof Number time ? Instant.ofEpochMilli(time.longValue()) : null); |
| | | } |
| | | |
| | | private void appendThresholdValue(TTest test, String sourceType, String identity, String metricCode, |
| | | Object value, String unit, List<Map<String, Object>> warnings, java.util.Set<String> seen) |
| | | { |
| | | if (!(value instanceof Number number) || !Double.isFinite(number.doubleValue())) return; |
| | | BigDecimal measurement = value instanceof BigDecimal decimal ? decimal : new BigDecimal(value.toString()); |
| | | for (var rule : TestWarningExceptionService.exceededRules(test, metricCode, measurement)) { |
| | | if (!seen.add(rule.getThresholdId() + ":" + sourceType + ":" + identity + ":" + metricCode)) continue; |
| | | Map<String, Object> warning = new LinkedHashMap<>(); |
| | | warning.put("exceptionType", "THRESHOLD"); warning.put("thresholdId", rule.getThresholdId()); |
| | | warning.put("sourceType", sourceType); warning.put("sourceIdentity", identity); |
| | | warning.put("metricCode", metricCode); warning.put("value", measurement); warning.put("unit", unit); |
| | | warning.put("minValue", rule.getMinValue()); warning.put("maxValue", rule.getMaxValue()); |
| | | warning.put("status", "OPEN"); warning.put("message", "当前测量超出预警阈值"); |
| | | warnings.add(warning); |
| | | } |
| | | } |
| | | |
| | | private String offlineIdentity(TTest test, LiveSource source) |
| | |
| | | if (channel.latestSnapshot() == null) return value; |
| | | value.put("source", "检测柜"); |
| | | value.put("observedAt", channel.latestSnapshot().receivedAt().toEpochMilli()); |
| | | value.put("deviceTime", channel.latestSnapshot().deviceTime() == null ? null : channel.latestSnapshot().deviceTime().toEpochMilli()); |
| | | // BTS 核心采样字段是快照的一等事实,不能只依赖动态 measurements 展示。 |
| | | value.put("currentA", channel.latestSnapshot().currentA()); |
| | | value.put("voltageV", channel.latestSnapshot().voltageV()); |
| | |
| | | value.put("source", "采集器"); value.put("sampleIntervalMs", collector.sampleIntervalMs()); |
| | | // 在业务响应边界转换毫秒时间,设备时间与服务器接收时间分别保留。 |
| | | value.put("measurements", collector.measurements().stream().filter(item -> battery.getCollectorChannels() != null |
| | | && battery.getCollectorChannels().stream().anyMatch(binding -> binding.getChannelCode().equals(item.channelCode()))).map(item -> { |
| | | && battery.getCollectorChannels().stream().anyMatch(binding -> binding.getChannelCode().equals(item.channelCode()) |
| | | && TestWarningExceptionService.matchesCollectorMetric(binding.getMeasurementCode(), item.metricCode()))).map(item -> { |
| | | Map<String, Object> measurement = new LinkedHashMap<>(); |
| | | measurement.put("metricCode", item.metricCode()); |
| | | measurement.put("channelCode", item.channelCode()); |
| | |
| | | private final TestWarningExceptionService warningExceptions; |
| | | private final Clock clock; |
| | | private final Map<String, Instant> lastSnapshotAt = new java.util.concurrent.ConcurrentHashMap<>(); |
| | | private final Map<Long, Instant> lastOfflineCheckAt = new java.util.concurrent.ConcurrentHashMap<>(); |
| | | private final Map<Long, Instant> lastRealtimeCheckAt = new java.util.concurrent.ConcurrentHashMap<>(); |
| | | |
| | | @Autowired |
| | | public TestDataReceptionServiceImpl(TTestMapper mapper, ITTestService tests, @Nullable TestDataObservationAdapter adapter, |
| | |
| | | { |
| | | for (TTest test : receivingTests()) { |
| | | Instant now = clock.instant(); |
| | | Instant previous = lastOfflineCheckAt.get(test.getTestId()); |
| | | // 离线检查独立于测量采样槽位,并与暂停、停止和删除共享测试事务锁。 |
| | | Instant previous = lastRealtimeCheckAt.get(test.getTestId()); |
| | | // 实时预警独立于持久化槽位,并与暂停、停止和删除共享测试事务锁。 |
| | | if (previous == null || !now.isBefore(previous.plusSeconds(5))) { |
| | | warningExceptions.recordOfflineSources(test, Date.from(now)); |
| | | lastOfflineCheckAt.put(test.getTestId(), now); |
| | | for (TTestBattery battery : batteries(test)) saveRealtime(test, battery); |
| | | lastRealtimeCheckAt.put(test.getTestId(), now); |
| | | } |
| | | for (TTestBattery battery : batteries(test)) { saveRealtime(test, battery); saveDeviceAggregates(test, battery); } |
| | | for (TTestBattery battery : batteries(test)) saveDeviceAggregates(test, battery); |
| | | } |
| | | } |
| | | |
| | |
| | | if (adapter == null) return; |
| | | TTestChannelBinding binding = battery.getBinding(); if (binding == null || binding.getDevid() == null || binding.getChlid() == null) return; |
| | | Instant now = clock.instant(); long interval = persistenceInterval(binding); String key = test.getTestId() + ":" + battery.getTestBatteryId(); |
| | | Instant previous = lastSnapshotAt.get(key); if (previous != null && now.isBefore(previous.plusMillis(interval))) return; |
| | | CabinetRealtime cabinet = latestCabinet(binding); |
| | | if (cabinet != null && !accepted(test, cabinet.deviceTime(), cabinet.serverReceivedAt())) cabinet = null; |
| | | CollectorRealtime collector = latestCollector(battery); |
| | | Map<String, CollectorMeasurement> mapped = mappedCollectorMetrics(test, battery, collector); |
| | | if (cabinet == null && (collector == null || !collector.online() || mapped.isEmpty())) return; |
| | | // 当前成功测量先判定预警;槽位尚未到期或历史行被删除,都不能阻断实时告警。 |
| | | if (cabinet != null) { |
| | | Date observedAt = Date.from(cabinet.serverReceivedAt()); |
| | | evaluateCabinet(test, binding, "VOLTAGE", BigDecimal.valueOf(cabinet.voltageV()), "V", observedAt); |
| | | evaluateCabinet(test, binding, "CURRENT", BigDecimal.valueOf(cabinet.currentA()), "A", observedAt); |
| | | evaluateCabinet(test, binding, "CAPACITY", BigDecimal.valueOf(cabinet.capacityAh()), "Ah", observedAt); |
| | | evaluateCabinet(test, binding, "ENERGY", BigDecimal.valueOf(cabinet.energyWh()), "Wh", observedAt); |
| | | } |
| | | for (CollectorMeasurement measurement : mapped.values().stream().distinct().toList()) { |
| | | warningExceptions.evaluate(test, "COLLECTOR", battery.getCollectorId() + ":" + measurement.channelCode(), |
| | | measurement.metricCode(), measurement.value(), measurement.unit(), Date.from(measurement.serverReceivedAt())); |
| | | } |
| | | Instant previous = lastSnapshotAt.get(key); if (previous != null && now.isBefore(previous.plusMillis(interval))) return; |
| | | Instant collectorTime = mapped.values().stream().map(CollectorMeasurement::deviceTime).max(Instant::compareTo).orElse(null); |
| | | Instant slot = Instant.ofEpochMilli((now.toEpochMilli() / interval) * interval); |
| | | // 删除修订随数据库保留;重启后同一槽位也不能重新出现。 |
| | |
| | | row.setCollectorSequenceId(mapped.values().stream().map(CollectorMeasurement::sequenceId).max(Long::compareTo).orElse(null)); row.setCollectorDeviceTimeUtc(collectorTime == null ? null : Date.from(collectorTime)); row.setCollectorDeviceTimeRaw(mapped.values().stream().filter(value -> value.deviceTime().equals(collectorTime)).map(CollectorMeasurement::deviceTimeRaw).findFirst().orElse(null)); |
| | | if (cabinet != null) { row.setCycleNo(cabinet.cycleNo()); row.setStepNo(cabinet.stepNo()); row.setStepType(cabinet.stepType()); row.setStepName(cabinet.stepName()); row.setVoltageV(BigDecimal.valueOf(cabinet.voltageV())); row.setCurrentA(BigDecimal.valueOf(cabinet.currentA())); row.setCapacityAh(BigDecimal.valueOf(cabinet.capacityAh())); row.setEnergyWh(BigDecimal.valueOf(cabinet.energyWh())); } |
| | | row.setTemperature(metricValue(mapped, "TEMPERATURE")); row.setPositivePotential(metricValue(mapped, "POSITIVE_POTENTIAL")); row.setNegativePotential(metricValue(mapped, "NEGATIVE_POTENTIAL")); row.setCollectorChannelCode(mapped.values().stream().map(CollectorMeasurement::channelCode).findFirst().orElse(null)); |
| | | // 只有本次观测成功取得的值进入阈值判断;离线、请求失败和缺失值在前面被过滤。 |
| | | Date cabinetObservedAt = cabinet == null || cabinet.serverReceivedAt() == null ? nowDate(now) : Date.from(cabinet.serverReceivedAt()); |
| | | if (cabinet != null) |
| | | { |
| | | evaluateCabinet(test, binding, "VOLTAGE", row.getVoltageV(), "V", cabinetObservedAt); |
| | | evaluateCabinet(test, binding, "CURRENT", row.getCurrentA(), "A", cabinetObservedAt); |
| | | evaluateCabinet(test, binding, "CAPACITY", row.getCapacityAh(), "Ah", cabinetObservedAt); |
| | | evaluateCabinet(test, binding, "ENERGY", row.getEnergyWh(), "Wh", cabinetObservedAt); |
| | | } |
| | | for (CollectorMeasurement measurement : mapped.values()) |
| | | { |
| | | if (measurement.value() != null) |
| | | warningExceptions.evaluate(test, "COLLECTOR", String.valueOf(battery.getCollectorId()), measurement.metricCode(), measurement.value(), measurement.unit(), |
| | | measurement.serverReceivedAt() == null ? nowDate(now) : Date.from(measurement.serverReceivedAt())); |
| | | } |
| | | Map<String, Object> extension = new LinkedHashMap<>(); |
| | | for (Map.Entry<String, CollectorMeasurement> e : mapped.entrySet()) if (!isFixedMetric(e.getKey())) |
| | | { |
| | |
| | | { |
| | | if (value != null) warningExceptions.evaluate(test, "CABINET", sourceIdentity(binding), metricCode, value, unit, occurredAt); |
| | | } |
| | | |
| | | private static Date nowDate(Instant instant) { return Date.from(instant); } |
| | | |
| | | private void saveDeviceAggregates(TTest test, TTestBattery battery) |
| | | { |
| | |
| | | |
| | | private CabinetRealtime latestCabinet(TTestChannelBinding b) { try { return adapter.latestCabinet(new ChannelIdentity(b.getAdapterCode(), b.getDevtype(), b.getDevid(), b.getSubdevid(), b.getChlid())); } catch (RuntimeException ex) { return null; } } |
| | | private CollectorRealtime latestCollector(TTestBattery b) { if (b.getCollectorId() == null) return null; try { return adapter.latestCollector(mapper.selectCollectorCode(b.getCollectorId())); } catch (RuntimeException ex) { return null; } } |
| | | private Map<String, CollectorMeasurement> mappedCollectorMetrics(TTest test, TTestBattery b, CollectorRealtime s) { Map<String, CollectorMeasurement> out = new LinkedHashMap<>(); if (s == null || !s.online() || s.measurements() == null || b.getCollectorChannels() == null) return out; for (CollectorMeasurement m : s.measurements()) if (accepted(test, m.deviceTime(), m.serverReceivedAt()) && b.getCollectorChannels().stream().anyMatch(x -> java.util.Objects.equals(x.getChannelCode(), m.channelCode()) && metricMatches(x.getMeasurementCode(), m.metricCode()))) out.put(m.metricCode().toUpperCase(Locale.ROOT), m); return out; } |
| | | private boolean metricMatches(String c, String a) { if (c == null || a == null) return false; c = c.toUpperCase(Locale.ROOT); a = a.toUpperCase(Locale.ROOT); return c.equals(a) || (c.contains("TEMPERATURE") && a.equals("TEMPERATURE")) || (c.contains("POSITIVE_POTENTIAL") && a.equals("POSITIVE_POTENTIAL")) || (c.contains("NEGATIVE_POTENTIAL") && a.equals("NEGATIVE_POTENTIAL")); } |
| | | private static boolean isFixedMetric(String code) { return "TEMPERATURE".equals(code) || "POSITIVE_POTENTIAL".equals(code) || "NEGATIVE_POTENTIAL".equals(code); } |
| | | private static BigDecimal metricValue(Map<String, CollectorMeasurement> values, String code) { return values.get(code) == null ? null : values.get(code).value(); } |
| | | private Map<String, CollectorMeasurement> mappedCollectorMetrics(TTest test, TTestBattery battery, CollectorRealtime source) |
| | | { |
| | | Map<String, CollectorMeasurement> result = new LinkedHashMap<>(); |
| | | if (source == null || !source.online() || battery.getCollectorChannels() == null) return result; |
| | | for (var binding : battery.getCollectorChannels()) for (var measurement : source.measurements()) { |
| | | if (accepted(test, measurement.deviceTime(), measurement.serverReceivedAt()) |
| | | && java.util.Objects.equals(binding.getChannelCode(), measurement.channelCode()) |
| | | && TestWarningExceptionService.matchesCollectorMetric(binding.getMeasurementCode(), measurement.metricCode())) { |
| | | // 绑定角色区分主、辅助测量;不能用设备通用指标名覆盖其他物理通道。 |
| | | result.put(binding.getMeasurementCode().toUpperCase(Locale.ROOT), measurement); |
| | | } |
| | | } |
| | | return result; |
| | | } |
| | | private static boolean isFixedMetric(String code) { return List.of("TEMPERATURE", "POSITIVE_POTENTIAL", "NEGATIVE_POTENTIAL", "PRIMARY_TEMPERATURE", "PRIMARY_POSITIVE_POTENTIAL", "PRIMARY_NEGATIVE_POTENTIAL").contains(code); } |
| | | private static BigDecimal metricValue(Map<String, CollectorMeasurement> values, String code) { |
| | | var measurement = values.get("PRIMARY_" + code); |
| | | if (measurement == null) measurement = values.get(code); |
| | | return measurement == null ? null : measurement.value(); |
| | | } |
| | | private static String sourceRecordId(CabinetRealtime cabinet, Map<String, CollectorMeasurement> measurements) |
| | | { |
| | | List<String> parts = new java.util.ArrayList<>(); |
| | |
| | | private static String sourceIdentity(TTestChannelBinding b) { return String.join(":", String.valueOf(b.getAdapterCode()), String.valueOf(b.getDevtype()), String.valueOf(b.getDevid()), String.valueOf(b.getSubdevid()), String.valueOf(b.getChlid())); } |
| | | private static String timezone(TTestChannelBinding b) { return b.getAdapterTimezoneId() == null || b.getAdapterTimezoneId().isBlank() ? "Asia/Hong_Kong" : b.getAdapterTimezoneId(); } |
| | | private static String blankToNull(String value) { return value == null || value.isBlank() ? null : value; } |
| | | private static boolean accepted(TTest test, Instant deviceTime, Instant serverReceivedAt) |
| | | // 实时提醒与持久化共用归属边界,避免同一旧观测只在首页触发异常。 |
| | | static boolean accepted(TTest test, Instant deviceTime, Instant serverReceivedAt) |
| | | { |
| | | if (deviceTime != null && test.getStartedAtUtc() != null && deviceTime.isBefore(test.getStartedAtUtc().toInstant())) return false; |
| | | return serverReceivedAt != null && (test.getReceptionBoundaryAtUtc() == null || !serverReceivedAt.isBefore(test.getReceptionBoundaryAtUtc().toInstant())); |
| | |
| | | |
| | | import java.math.BigDecimal; |
| | | import java.util.Date; |
| | | import java.util.List; |
| | | import java.util.Locale; |
| | | import com.alibaba.fastjson2.JSON; |
| | | import org.springframework.stereotype.Service; |
| | | import org.springframework.transaction.annotation.Transactional; |
| | |
| | | } |
| | | } |
| | | |
| | | /** 仅对成功接收的非空测量值判定阈值,并聚合更新开放异常及规则快照。 */ |
| | | /** 在持有测试事务锁时,对接收中的成功实时测量聚合开放异常,不受采样入库周期限制。 */ |
| | | public void evaluate(TTest test, String sourceType, String sourceIdentity, String metricCode, BigDecimal value, String unit, Date occurredAt) { |
| | | if (value == null || test == null || test.getThresholds() == null) return; |
| | | for (TTestWarningThreshold rule : test.getThresholds()) { |
| | | if (rule == null || rule.getMetricCode() == null || !matches(rule.getMetricCode(), metricCode) |
| | | || !outside(value, rule.getMinValue(), rule.getMaxValue())) continue; |
| | | for (TTestWarningThreshold rule : exceededRules(test, metricCode, value)) { |
| | | mapper.upsertWarningException(test.getTestId(), rule.getThresholdId(), sourceType, sourceIdentity, metricCode, value, unit, |
| | | JSON.toJSONString(rule), rule.getSeverity(), occurredAt == null ? new Date() : occurredAt); |
| | | } |
| | | } |
| | | |
| | | static List<TTestWarningThreshold> exceededRules(TTest test, String metricCode, BigDecimal value) { |
| | | if (value == null || test == null || test.getThresholds() == null |
| | | || !"RUNNING".equals(test.getTestStatus()) || !"RECEIVING".equals(test.getReceptionStatus())) return List.of(); |
| | | return test.getThresholds().stream().filter(rule -> rule != null |
| | | && matchesMetric(rule.getMetricCode(), metricCode) |
| | | && outside(value, rule.getMinValue(), rule.getMaxValue())).toList(); |
| | | } |
| | | |
| | | /** 由测试负责人关闭开放异常,处理内容保存在异常事实中。 */ |
| | |
| | | int rows = mapper.closeWarningException(exceptionId, actorId, actorName, new Date(), handlingNote); |
| | | return rows; |
| | | } |
| | | private boolean outside(BigDecimal value, BigDecimal min, BigDecimal max) { return (min != null && value.compareTo(min) < 0) || (max != null && value.compareTo(max) > 0); } |
| | | private boolean matches(String configured, String actual) { |
| | | if (actual == null) return false; |
| | | private static boolean outside(BigDecimal value, BigDecimal min, BigDecimal max) { return (min != null && value.compareTo(min) < 0) || (max != null && value.compareTo(max) > 0); } |
| | | static boolean matchesCollectorMetric(String configured, String actual) { |
| | | if (configured == null || actual == null) return false; |
| | | String role = configured.toUpperCase(Locale.ROOT), metric = actual.toUpperCase(Locale.ROOT); |
| | | return role.equals(metric) || List.of("TEMPERATURE", "POSITIVE_POTENTIAL", "NEGATIVE_POTENTIAL").stream() |
| | | .anyMatch(code -> metric.equals(code) && role.contains(code)); |
| | | } |
| | | |
| | | static boolean matchesMetric(String configured, String actual) { |
| | | if (configured == null || actual == null) return false; |
| | | if (configured.equalsIgnoreCase(actual)) return true; |
| | | String c = configured.toUpperCase(), a = actual.toUpperCase(); |
| | | String c = configured.toUpperCase(Locale.ROOT), a = actual.toUpperCase(Locale.ROOT); |
| | | return (c.contains("VOLTAGE") && a.contains("VOLTAGE")) || (c.contains("CURRENT") && a.contains("CURRENT")) |
| | | || (c.contains("CAPACITY") && a.contains("CAPACITY")) || (c.contains("ENERGY") && a.contains("ENERGY")) |
| | | || (c.contains("TEMPERATURE") && a.contains("TEMPERATURE")) || (c.contains("POSITIVE_POTENTIAL") && a.contains("POSITIVE_POTENTIAL")) |
| | |
| | | class TTestLiveServiceImplTest |
| | | { |
| | | @Test |
| | | void thresholdWarningsRespectDeviceTimeAndReceptionBoundary() |
| | | { |
| | | Instant now = Instant.parse("2026-09-05T12:00:00Z"); |
| | | TTest test = test("RECEIVING"); |
| | | test.setStartedAtUtc(java.util.Date.from(now)); |
| | | test.setReceptionBoundaryAtUtc(java.util.Date.from(now)); |
| | | var rule = new com.battery.business.domain.TTestWarningThreshold(); |
| | | rule.setThresholdId(11L); rule.setMetricCode("VOLTAGE"); rule.setMaxValue(new BigDecimal("3")); |
| | | test.setThresholds(List.of(rule)); |
| | | assertTrue(liveService(test, now, now.minusSeconds(1)).current(10L).exceptions().isEmpty()); |
| | | assertEquals(1, liveService(test, now, now).current(10L).exceptions().size()); |
| | | test.setReceptionBoundaryAtUtc(java.util.Date.from(now.plusSeconds(1))); |
| | | assertTrue(liveService(test, now, now).current(10L).exceptions().isEmpty()); |
| | | |
| | | var battery = test.getBatteries().get(0); battery.setCollectorId(7L); battery.setBinding(null); |
| | | var binding = new com.battery.business.domain.TTestCollectorChannel(); |
| | | binding.setChannelCode("S-01"); binding.setMeasurementCode("PRIMARY_TEMPERATURE"); |
| | | battery.setCollectorChannels(List.of(binding)); |
| | | rule.setMetricCode("TEMPERATURE"); rule.setMaxValue(new BigDecimal("33")); |
| | | var tests = mock(ITTestService.class); when(tests.selectTestById(10L)).thenReturn(test); |
| | | var mapper = mock(TTestMapper.class); when(mapper.selectCollectorCode(7L)).thenReturn("collector-7"); |
| | | var adapter = mock(TestDataObservationAdapter.class); |
| | | var service = new TTestLiveServiceImpl(tests, |
| | | new com.battery.device.api.discovery.InMemoryDeviceDiscoveryCandidateStore(), adapter, mapper); |
| | | test.setReceptionBoundaryAtUtc(java.util.Date.from(now)); |
| | | for (Instant deviceTime : List.of(now.minusSeconds(1), now)) { |
| | | var measurement = new TestDataObservationAdapter.CollectorMeasurement("S-01", 1L, deviceTime, deviceTime.toString(), |
| | | now, "TEMPERATURE", new BigDecimal("38"), "℃"); |
| | | when(adapter.latestCollector("collector-7")).thenReturn(new TestDataObservationAdapter.CollectorRealtime( |
| | | "collector-7", true, 1000, List.of(measurement), null)); |
| | | assertEquals(deviceTime.equals(now) ? 1 : 0, service.current(10L).exceptions().size()); |
| | | } |
| | | test.setReceptionBoundaryAtUtc(java.util.Date.from(now.plusSeconds(1))); |
| | | assertTrue(service.current(10L).exceptions().isEmpty()); |
| | | } |
| | | |
| | | @Test |
| | | void currentTemperatureAlarmIsIndependentOfHistoricalTodoHandling() throws Exception |
| | | { |
| | | try (var simulator = new com.battery.device.simulator.BtsDevelopmentSimulator(0)) { |
| | | TTest test = test("RECEIVING"); |
| | | var battery = test.getBatteries().get(0); battery.setCollectorId(7L); |
| | | var primary = new com.battery.business.domain.TTestCollectorChannel(); |
| | | primary.setChannelCode("S-01"); primary.setMeasurementCode("PRIMARY_TEMPERATURE"); |
| | | var secondary = new com.battery.business.domain.TTestCollectorChannel(); |
| | | secondary.setChannelCode("S-03"); secondary.setMeasurementCode("SECONDARY_TEMPERATURE_2"); |
| | | battery.setCollectorChannels(List.of(primary, secondary)); |
| | | var rule = new com.battery.business.domain.TTestWarningThreshold(); |
| | | rule.setThresholdId(11L); rule.setMetricCode("TEMPERATURE"); |
| | | rule.setMinValue(new BigDecimal("8")); rule.setMaxValue(new BigDecimal("33")); |
| | | test.setThresholds(List.of(rule)); |
| | | var tests = mock(ITTestService.class); when(tests.selectTestById(10L)).thenReturn(test); |
| | | var mapper = mock(TTestMapper.class); when(mapper.selectCollectorCode(7L)).thenReturn("dev-sim-collector-1"); |
| | | var service = new TTestLiveServiceImpl(tests, |
| | | new com.battery.device.api.discovery.InMemoryDeviceDiscoveryCandidateStore(), simulator, mapper); |
| | | simulator.configureCollectorChannel(1, "S-01", List.of( |
| | | new com.battery.device.simulator.BtsSimulatorStateStore.CollectorMetricState("temperature", new BigDecimal("38"), "℃"))); |
| | | var alarms = service.current(10L).exceptions().stream().filter(item -> "THRESHOLD".equals(item.get("exceptionType"))).toList(); |
| | | assertEquals(1, alarms.size()); |
| | | assertEquals(new BigDecimal("38"), alarms.get(0).get("value")); |
| | | assertEquals("7:S-01", alarms.get(0).get("sourceIdentity")); |
| | | assertFalse(alarms.get(0).containsKey("exceptionId")); |
| | | |
| | | var history = new TTestWarningException(); history.setExceptionStatus("OPEN"); |
| | | history.setExceptionType("THRESHOLD"); history.setMetricCode("temperature"); |
| | | history.setAbnormalValue(new BigDecimal("38")); |
| | | when(mapper.selectWarningExceptions(10L, "OPEN")).thenReturn(List.of(history)); |
| | | simulator.configureCollectorChannel(1, "S-01", List.of( |
| | | new com.battery.device.simulator.BtsSimulatorStateStore.CollectorMetricState("temperature", new BigDecimal("25"), "℃"))); |
| | | assertTrue(service.current(10L).exceptions().stream().noneMatch(item -> "THRESHOLD".equals(item.get("exceptionType")))); |
| | | assertEquals("OPEN", history.getExceptionStatus()); |
| | | } |
| | | } |
| | | |
| | | @Test |
| | | void consoleReadsConfiguredSimulatorPlanAtCurrentChannelPosition() throws Exception |
| | | { |
| | | try (var simulator = new com.battery.device.simulator.BtsDevelopmentSimulator(0)) { |
| | |
| | | private static TTestLiveServiceImpl liveService(TTest test, Instant now, boolean online, |
| | | DeviceWorkStatus workStatus, TTestMapper mapper) |
| | | { |
| | | return liveService(test, now, online, workStatus, mapper, now); |
| | | } |
| | | |
| | | private static TTestLiveServiceImpl liveService(TTest test, Instant now, Instant deviceTime) |
| | | { |
| | | return liveService(test, now, true, DeviceWorkStatus.WORKING, null, deviceTime); |
| | | } |
| | | |
| | | private static TTestLiveServiceImpl liveService(TTest test, Instant now, boolean online, |
| | | DeviceWorkStatus workStatus, TTestMapper mapper, Instant deviceTime) |
| | | { |
| | | DeviceDiscoveryCandidateStore candidates = mock(DeviceDiscoveryCandidateStore.class); |
| | | TestDataObservationAdapter simulator = mock(TestDataObservationAdapter.class); |
| | | ITTestService tests = mock(ITTestService.class); |
| | |
| | | RealtimeSnapshot snapshot = new RealtimeSnapshot(identity, workStatus, 1, 1, "cc", null, |
| | | new BigDecimal("1.200"), new BigDecimal("3.700"), new BigDecimal("2.500"), |
| | | new BigDecimal("9.250"), 12L, 12L, "0", now, |
| | | java.util.Map.of()); |
| | | java.util.Map.of(), com.battery.device.api.model.DeviceConnectionStatus.ONLINE, deviceTime, List.of(), null); |
| | | DeviceDiscoveryCandidateStore.ChannelCandidate channel = new DeviceDiscoveryCandidateStore.ChannelCandidate( |
| | | identity, online, workStatus, now, snapshot); |
| | | DeviceDiscoveryCandidateStore.CabinetCandidate cabinet = new DeviceDiscoveryCandidateStore.CabinetCandidate( |
| | |
| | | private static final Clock CLOCK = Clock.fixed(CAPTURED_AT, ZoneOffset.UTC); |
| | | |
| | | @Test |
| | | void detectsRealtimeTemperatureExceededBetweenPersistenceSlots() |
| | | { |
| | | Fixture fixture = fixture(); |
| | | var threshold = new com.battery.business.domain.TTestWarningThreshold(); |
| | | threshold.setThresholdId(11L); threshold.setMetricCode("TEMPERATURE"); |
| | | threshold.setMinValue(new BigDecimal("8")); threshold.setMaxValue(new BigDecimal("33")); |
| | | threshold.setUnit("℃"); threshold.setSeverity("WARNING"); |
| | | fixture.tests.selectTestById(1L).setThresholds(List.of(threshold)); |
| | | Clock clock = mock(Clock.class); |
| | | when(clock.instant()).thenReturn(CAPTURED_AT); |
| | | var service = new TestDataReceptionServiceImpl(fixture.mapper, fixture.tests, fixture.simulator, clock); |
| | | when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true, |
| | | List.of(collector("S-01", 10L, "temperature", "25.1", "℃")))); |
| | | when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1); |
| | | service.captureReceivedData(); |
| | | |
| | | when(clock.instant()).thenReturn(CAPTURED_AT.plusSeconds(5)); |
| | | when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true, |
| | | List.of(collector("S-01", 11L, "temperature", "38", "℃")))); |
| | | service.captureReceivedData(); |
| | | |
| | | verify(fixture.mapper).upsertWarningException(eq(1L), eq(11L), eq("COLLECTOR"), any(), |
| | | eq("temperature"), eq(new BigDecimal("38")), eq("℃"), any(), eq("WARNING"), any(Date.class)); |
| | | verify(fixture.mapper, times(1)).insertRealtimeRecord(any()); |
| | | } |
| | | |
| | | @Test |
| | | void primaryOvertemperatureIsNotOverwrittenByNormalSecondaryTemperature() |
| | | { |
| | | Fixture fixture = fixture(); |
| | | var threshold = new com.battery.business.domain.TTestWarningThreshold(); |
| | | threshold.setThresholdId(11L); threshold.setMetricCode("TEMPERATURE"); |
| | | threshold.setMinValue(new BigDecimal("8")); threshold.setMaxValue(new BigDecimal("33")); |
| | | threshold.setUnit("℃"); threshold.setSeverity("WARNING"); |
| | | var test = fixture.tests.selectTestById(1L); |
| | | test.setThresholds(List.of(threshold)); |
| | | test.getBatteries().get(0).setCollectorChannels(List.of( |
| | | mapping("PRIMARY_TEMPERATURE", "S-01"), mapping("SECONDARY_TEMPERATURE_2", "S-03"), |
| | | mapping("SECONDARY_TEMPERATURE_3", "S-05"))); |
| | | when(fixture.simulator.latestCollector("collector-9")).thenReturn(collectorSnapshot(true, List.of( |
| | | collector("S-01", 10L, "temperature", "38", "℃"), |
| | | collector("S-03", 10L, "temperature", "25.3", "℃"), |
| | | collector("S-05", 10L, "temperature", "25.5", "℃")))); |
| | | when(fixture.mapper.insertRealtimeRecord(any())).thenReturn(1); |
| | | |
| | | fixture.service().captureReceivedData(); |
| | | |
| | | verify(fixture.mapper).upsertWarningException(eq(1L), eq(11L), eq("COLLECTOR"), any(), |
| | | eq("temperature"), eq(new BigDecimal("38")), eq("℃"), any(), eq("WARNING"), any(Date.class)); |
| | | var row = ArgumentCaptor.forClass(TTestRealtimeRecord.class); |
| | | verify(fixture.mapper).insertRealtimeRecord(row.capture()); |
| | | assertEquals(new BigDecimal("38"), row.getValue().getTemperature()); |
| | | assertTrue(row.getValue().getExtensionMeasurementsJson().contains("SECONDARY_TEMPERATURE_3")); |
| | | } |
| | | |
| | | @Test |
| | | void recordsOfflineSourcesInBackgroundEvenWithoutSuccessfulSamples() { |
| | | Fixture fixture = fixture(); |
| | | when(fixture.mapper.selectOfflineTestSources(1L)).thenReturn(List.of( |
| | |
| | | |
| | | service.evaluate(test, "COLLECTOR", "adapter:9:S-01", "temperature", new BigDecimal("31"), "℃", new Date()); |
| | | service.evaluate(test, "COLLECTOR", "adapter:9:S-01", "temperature", new BigDecimal("20"), "℃", new Date()); |
| | | service.evaluate(test, "COLLECTOR", "adapter:9:S-01", "temperature", new BigDecimal("0"), "℃", new Date()); |
| | | service.evaluate(test, "COLLECTOR", "adapter:9:S-01", "temperature", new BigDecimal("30"), "℃", new Date()); |
| | | service.evaluate(test, "COLLECTOR", "adapter:9:S-01", "temperature", null, "℃", new Date()); |
| | | |
| | | verify(mapper, times(1)).upsertWarningException(eq(1L), eq(11L), eq("COLLECTOR"), eq("adapter:9:S-01"), |
| | | eq("temperature"), eq(new BigDecimal("31")), eq("℃"), anyString(), eq("WARNING"), any(Date.class)); |
| | | } |
| | | |
| | | @Test |
| | | void nonReceivingTestsDoNotCreateThresholdTodos() { |
| | | var mapper = mock(TTestMapper.class); |
| | | var service = new TestWarningExceptionService(mapper); |
| | | var test = testWithThreshold("TEMPERATURE", new BigDecimal("8"), new BigDecimal("33")); |
| | | for (String reception : List.of("PAUSED", "ENDED", "NOT_STARTED")) { |
| | | test.setReceptionStatus(reception); |
| | | service.evaluate(test, "COLLECTOR", "9:S-01", "temperature", new BigDecimal("38"), "℃", new Date()); |
| | | } |
| | | verifyNoInteractions(mapper); |
| | | } |
| | | |
| | | @Test |
| | |
| | | TTest test = new TTest(); |
| | | test.setTestId(1L); |
| | | test.setThresholds(List.of(threshold)); |
| | | test.setTestStatus("RUNNING"); |
| | | test.setReceptionStatus("RECEIVING"); |
| | | return test; |
| | | } |
| | | } |