Merge remote-tracking branch 'origin/feature/issue-48'

# Conflicts:
#	frontend/src/router/index.ts
#	wm-production/pom.xml
This commit is contained in:
2026-06-15 08:33:46 +08:00
235 changed files with 23417 additions and 225 deletions
@@ -1,11 +1,26 @@
package com.water.data_engine;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
/**
* 数据引擎应用主类
* Issue #41: 实时流数据采集(MQTT/Kafka Consumer)
*/
@SpringBootApplication
public class DataEngineApplication {
public static void main(String[] args) {
SpringApplication.run(DataEngineApplication.class, args);
}
@Bean
public ObjectMapper objectMapper() {
ObjectMapper mapper = new ObjectMapper();
mapper.registerModule(new JavaTimeModule());
return mapper;
}
}
@@ -0,0 +1,20 @@
package com.water.data_engine.config;
import com.water.data_engine.service.TDengineService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
@Component
public class DataEngineInitializer implements CommandLineRunner {
@Autowired
private TDengineService tdengineService;
@Override
public void run(String... args) throws Exception {
// 系统启动时初始化 TDengine 数据库和表
tdengineService.initializeDatabase();
System.out.println("数据引擎初始化完成");
}
}
@@ -1,62 +1,45 @@
package com.water.data_engine.config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import java.util.HashMap;
import java.util.Map;
/**
* Kafka 配置
* 用于实时数据流采集和传输
*/
@EnableKafka
@Configuration
public class KafkaConfig {
@Value("${spring.kafka.bootstrap-servers:${KAFKA_SERVERS:127.0.0.1}:9092}")
@Value("${spring.kafka.bootstrap.servers}")
private String bootstrapServers;
@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.ACKS_CONFIG, "1");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
return new DefaultKafkaProducerFactory<>(props);
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
return new KafkaTemplate<>(producerFactory);
}
@Value("${spring.kafka.consumer.group-id}")
private String groupId;
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "wm-data-engine");
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
return new DefaultKafkaConsumerFactory<>(props);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
ConsumerFactory<String, String> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.setConcurrency(3);
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
}
}
@@ -0,0 +1,56 @@
package com.water.data_engine.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
/**
* MQTT 配置类
*/
@Data
@Configuration
@ConfigurationProperties(prefix = "mqtt")
public class MqttConfig {
/**
* MQTT Broker URL
*/
private String brokerUrl;
/**
* 客户端 ID
*/
private String clientId;
/**
* 用户名
*/
private String username;
/**
* 密码
*/
private String password;
/**
* 连接超时时间(秒)
*/
private int timeout;
/**
* 心跳间隔(秒)
*/
private int keepAlive;
/**
* 主题配置
*/
private TopicConfig topic;
@Data
public static class TopicConfig {
private String iotTelemetry;
private String iotCommand;
private String qualityData;
}
}
@@ -0,0 +1,58 @@
package com.water.data_engine.config;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
/**
* MQTT 连接配置工厂
*/
@Slf4j
@Configuration
@RequiredArgsConstructor
public class MqttConnectionFactory {
private final MqttConfig mqttConfig;
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[]{mqttConfig.getBrokerUrl()});
options.setUserName(mqttConfig.getUsername());
options.setPassword(mqttConfig.getPassword().toCharArray());
options.setConnectionTimeout(mqttConfig.getTimeout());
options.setKeepAliveInterval(mqttConfig.getKeepAlive());
options.setCleanSession(false);
options.setAutomaticReconnect(true);
factory.setConnectionOptions(options);
return factory;
}
@Bean
public MqttClient mqttClient() throws Exception {
MqttClient client = new MqttClient(
mqttConfig.getBrokerUrl(),
mqttConfig.getClientId(),
new MemoryPersistence()
);
try {
client.connect();
log.info("MQTT 客户端连接成功: {}", mqttConfig.getClientId());
} catch (Exception e) {
log.error("MQTT 客户端连接失败: {}", e.getMessage());
throw e;
}
return client;
}
}
@@ -1,21 +1,15 @@
package com.water.data_engine.config;
import com.baomidou.mybatisplus.annotation.DbType;
import com.baomidou.mybatisplus.core.handlers.MetaObjectHandler;
import com.baomidou.mybatisplus.extension.plugins.MybatisPlusInterceptor;
import com.baomidou.mybatisplus.extension.plugins.inner.PaginationInnerInterceptor;
import org.apache.ibatis.reflection.MetaObject;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.time.LocalDateTime;
/**
* MyBatis-Plus 配置
* MyBatisPlus配置
*/
@Configuration
@MapperScan("com.water.data_engine.mapper")
public class MyBatisPlusConfig {
/**
@@ -24,26 +18,8 @@ public class MyBatisPlusConfig {
@Bean
public MybatisPlusInterceptor mybatisPlusInterceptor() {
MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
interceptor.addInnerInterceptor(new PaginationInnerInterceptor(DbType.POSTGRE_SQL));
// 添加分页插件
interceptor.addInnerInterceptor(new PaginationInnerInterceptor(DbType.MYSQL));
return interceptor;
}
/**
* 自动填充处理器
*/
@Bean
public MetaObjectHandler metaObjectHandler() {
return new MetaObjectHandler() {
@Override
public void insertFill(MetaObject metaObject) {
this.strictInsertFill(metaObject, "createdAt", LocalDateTime.class, LocalDateTime.now());
this.strictInsertFill(metaObject, "updatedAt", LocalDateTime.class, LocalDateTime.now());
}
@Override
public void updateFill(MetaObject metaObject) {
this.strictUpdateFill(metaObject, "updatedAt", LocalDateTime.class, LocalDateTime.now());
}
};
}
}
@@ -0,0 +1,60 @@
package com.water.data_engine.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
@Configuration
@ConfigurationProperties(prefix = "tdengine")
public class TDengineConfig {
private String host;
private Integer port = 6030;
private String username;
private String password;
private String database;
// getters and setters
public String getHost() {
return host;
}
public void setHost(String host) {
this.host = host;
}
public Integer getPort() {
return port;
}
public void setPort(Integer port) {
this.port = port;
}
public String getUsername() {
return username;
}
public void setUsername(String username) {
this.username = username;
}
public String getPassword() {
return password;
}
public void setPassword(String password) {
this.password = password;
}
public String getDatabase() {
return database;
}
public void setDatabase(String database) {
this.database = database;
}
public String getJdbcUrl() {
return String.format("jdbc:TAOS://%s:%d/%s?user=%s&password=%s",
host, port, database, username, password);
}
}
@@ -0,0 +1,77 @@
package com.water.data_engine.controller;
import com.water.data_engine.entity.IotData;
import com.water.data_engine.service.TDengineService;
import com.water.data_engine.service.DataCollectService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
@Slf4j
@RestController
@RequestMapping("/api/data-engine")
public class DataEngineController {
@Autowired
private TDengineService tdengineService;
@Autowired
private DataCollectService dataCollectService;
@PostMapping("/test-write")
public Map<String, Object> testWrite(@RequestBody IotData data) {
Map<String, Object> result = new HashMap<>();
try {
// 设置测试数据
if (data.getCollectTime() == null) {
data.setCollectTime(LocalDateTime.now());
}
if (data.getStatus() == null) {
data.setStatus(1);
}
tdengineService.insertIotData(data);
result.put("success", true);
result.put("message", "测试数据写入成功");
result.put("deviceSn", data.getDeviceSn());
result.put("collectTime", data.getCollectTime());
} catch (Exception e) {
result.put("success", false);
result.put("message", "测试数据写入失败: " + e.getMessage());
}
return result;
}
@GetMapping("/status")
public Map<String, Object> getStatus() {
Map<String, Object> result = new HashMap<>();
result.put("status", "running");
result.put("tdengine", "connected");
result.put("kafka", "listening");
return result;
}
@PostMapping("/initialize")
public Map<String, Object> initialize() {
Map<String, Object> result = new HashMap<>();
try {
tdengineService.initializeDatabase();
result.put("success", true);
result.put("message", "TDengine 初始化完成");
} catch (Exception e) {
result.put("success", false);
result.put("message", "初始化失败: " + e.getMessage());
}
return result;
}
}
@@ -0,0 +1,119 @@
package com.water.data_engine.controller;
import com.water.data_engine.service.DataStatisticsService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.format.annotation.DateTimeFormat;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.time.LocalDateTime;
import java.util.Map;
/**
* 数据统计控制器
* 提供数据采集统计、质量分析等接口
*/
@Slf4j
@RestController
@RequestMapping("/api/statistics")
@Tag(name = "数据统计接口", description = "数据采集统计、质量分析")
@RequiredArgsConstructor
public class DataStatisticsController {
private final DataStatisticsService dataStatisticsService;
/**
* 获取数据采集统计信息
*/
@GetMapping("/data")
@Operation(summary = "获取数据采集统计", description = "查询指定时间范围内的数据采集统计信息")
public ResponseEntity<Map<String, Object>> getDataStatistics(
@Parameter(description = "开始时间 (格式: yyyy-MM-dd HH:mm:ss)", example = "2024-06-14 00:00:00")
@RequestParam(required = false) @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") String startTime,
@Parameter(description = "结束时间 (格式: yyyy-MM-dd HH:mm:ss)", example = "2024-06-14 23:59:59")
@RequestParam(required = false) @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") String endTime) {
Map<String, Object> stats = dataStatisticsService.getDataStatistics(startTime, endTime);
return ResponseEntity.ok(stats);
}
/**
* 获取设备数据统计
*/
@GetMapping("/device/{deviceSn}")
@Operation(summary = "获取设备数据统计", description = "查询指定设备的详细数据统计")
public ResponseEntity<Map<String, Object>> getDeviceStatistics(
@Parameter(description = "设备编号") @PathVariable String deviceSn,
@Parameter(description = "开始时间 (格式: yyyy-MM-dd HH:mm:ss)")
@RequestParam(required = false) @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") String startTime,
@Parameter(description = "结束时间 (格式: yyyy-MM-dd HH:mm:ss)")
@RequestParam(required = false) @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") String endTime) {
Map<String, Object> stats = dataStatisticsService.getDeviceStatistics(deviceSn, startTime, endTime);
return ResponseEntity.ok(stats);
}
/**
* 获取错误数据统计
*/
@GetMapping("/errors")
@Operation(summary = "获取错误数据统计", description = "查询指定时间范围内的错误数据统计")
public ResponseEntity<Map<String, Object>> getErrorStatistics(
@Parameter(description = "开始时间 (格式: yyyy-MM-dd HH:mm:ss)")
@RequestParam(required = false) @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") String startTime,
@Parameter(description = "结束时间 (格式: yyyy-MM-dd HH:mm:ss)")
@RequestParam(required = false) @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss") String endTime) {
Map<String, Object> stats = dataStatisticsService.getErrorStatistics(startTime, endTime);
return ResponseEntity.ok(stats);
}
/**
* 获取实时数据质量指标
*/
@GetMapping("/quality")
@Operation(summary = "获取数据质量指标", description = "查询实时数据质量统计")
public ResponseEntity<Map<String, Object>> getDataQuality() {
// 默认查询最近1小时的质量指标
String endTime = LocalDateTime.now().format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
String startTime = LocalDateTime.now().minusHours(1).format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> stats = dataStatisticsService.getDataStatistics(startTime, endTime);
// 计算质量分数
Integer total = (Integer) stats.get("totalRecords");
Integer success = (Integer) stats.get("successRecords");
Double avgQuality = (Double) stats.get("avgDataQuality");
Map<String, Object> quality = Map.of(
"totalRecords", total,
"successRecords", success,
"failedRecords", stats.get("failedRecords"),
"successRate", stats.get("successRate"),
"avgDataQuality", avgQuality,
"qualityGrade", calculateQualityGrade(avgQuality),
"lastUpdated", endTime
);
return ResponseEntity.ok(quality);
}
/**
* 计算质量等级
*/
private String calculateQualityGrade(double quality) {
if (quality >= 95) return "优秀";
if (quality >= 85) return "良好";
if (quality >= 75) return "一般";
if (quality >= 60) return "较差";
return "差";
}
}
@@ -0,0 +1,166 @@
package com.water.data_engine.controller;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.water.data_engine.entity.MeterReadRecord;
import com.water.data_engine.entity.MeterReadTask;
import com.water.data_engine.entity.MeterInfo;
import com.water.data_engine.entity.CustomerAccount;
import com.water.data_engine.service.MeterReadService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* 抄表管理控制器
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Slf4j
@RestController
@RequiredArgsConstructor
@RequestMapping("/api/v1/meter-read")
public class MeterReadController {
private final MeterReadService meterReadService;
/**
* 创建抄表记录
*/
@PostMapping("/records")
public ResponseEntity<MeterReadRecord> createReadRecord(@RequestBody MeterReadRecord record) {
log.info("创建抄表记录: {}", record);
MeterReadRecord result = meterReadService.createReadRecord(record);
return ResponseEntity.ok(result);
}
/**
* 获取抄表记录列表
*/
@GetMapping("/records")
public ResponseEntity<List<MeterReadRecord>> getReadRecords(
@RequestParam(required = false) String accountNo,
@RequestParam(required = false) String meterNo,
@RequestParam(required = false) String readType) {
Map<String, Object> params = new HashMap<>();
if (accountNo != null) params.put("accountNo", accountNo);
if (meterNo != null) params.put("meterNo", meterNo);
if (readType != null) params.put("readType", readType);
List<MeterReadRecord> records = meterReadService.getReadRecords(params);
return ResponseEntity.ok(records);
}
/**
* 获取抄表记录分页
*/
@GetMapping("/records/page")
public ResponseEntity<Page<MeterReadRecord>> getReadRecordPage(
@RequestParam(defaultValue = "1") Long current,
@RequestParam(defaultValue = "10") Long size,
@RequestParam(required = false) String accountNo,
@RequestParam(required = false) String meterNo) {
Page<MeterReadRecord> page = new Page<>(current, size);
Map<String, Object> params = new HashMap<>();
if (accountNo != null) params.put("accountNo", accountNo);
if (meterNo != null) params.put("meterNo", meterNo);
Page<MeterReadRecord> result = meterReadService.getReadRecordPage(page, params);
return ResponseEntity.ok(result);
}
/**
* 验证抄表记录
*/
@PostMapping("/records/{id}/verify")
public ResponseEntity<MeterReadRecord> verifyReadRecord(@PathVariable Long id, @RequestParam String verifiedBy) {
log.info("验证抄表记录: id={}, verifiedBy={}", id, verifiedBy);
MeterReadRecord result = meterReadService.verifyReadRecord(id, verifiedBy);
return ResponseEntity.ok(result);
}
/**
* 远程抄表
*/
@PostMapping("/remote-read")
public ResponseEntity<MeterReadRecord> remoteRead(@RequestParam String meterNo, @RequestParam BigDecimal readValue) {
log.info("远程抄表: meterNo={}, readValue={}", meterNo, readValue);
MeterReadRecord result = meterReadService.remoteRead(meterNo, readValue);
return ResponseEntity.ok(result);
}
/**
* 创建抄表任务
*/
@PostMapping("/tasks")
public ResponseEntity<MeterReadTask> createReadTask(@RequestBody MeterReadTask task) {
log.info("创建抄表任务: {}", task);
MeterReadTask result = meterReadService.createReadTask(task);
return ResponseEntity.ok(result);
}
/**
* 获取抄表任务列表
*/
@GetMapping("/tasks")
public ResponseEntity<List<MeterReadTask>> getReadTasks(
@RequestParam(required = false) String taskType,
@RequestParam(required = false) String status,
@RequestParam(required = false) String assignee,
@RequestParam(required = false) String area) {
Map<String, Object> params = new HashMap<>();
if (taskType != null) params.put("taskType", taskType);
if (status != null) params.put("status", status);
if (assignee != null) params.put("assignee", assignee);
if (area != null) params.put("area", area);
List<MeterReadTask> tasks = meterReadService.getReadTasks(params);
return ResponseEntity.ok(tasks);
}
/**
* 启动抄表任务
*/
@PostMapping("/tasks/{id}/start")
public ResponseEntity<Boolean> startReadTask(@PathVariable Long id) {
log.info("启动抄表任务: id={}", id);
boolean result = meterReadService.startReadTask(id);
return ResponseEntity.ok(result);
}
/**
* 完成抄表任务
*/
@PostMapping("/tasks/{id}/complete")
public ResponseEntity<Boolean> completeReadTask(@PathVariable Long id,
@RequestParam Integer actualCount,
@RequestParam(required = false) Integer abnormalCount) {
log.info("完成抄表任务: id={}, actualCount={}, abnormalCount={}", id, actualCount, abnormalCount);
boolean result = meterReadService.completeReadTask(id, actualCount, abnormalCount != null ? abnormalCount : 0);
return ResponseEntity.ok(result);
}
/**
* 获取异常抄表记录
*/
@GetMapping("/records/abnormal")
public ResponseEntity<List<MeterReadRecord>> getAbnormalRecords(
@RequestParam(required = false) String accountNo,
@RequestParam(required = false) String meterNo) {
Map<String, Object> params = new HashMap<>();
if (accountNo != null) params.put("accountNo", accountNo);
if (meterNo != null) params.put("meterNo", meterNo);
List<MeterReadRecord> records = meterReadService.getAbnormalRecords(params);
return ResponseEntity.ok(records);
}
}
@@ -0,0 +1,107 @@
package com.water.data_engine.controller;
import com.water.data_engine.service.MqttPublishService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.util.Map;
/**
* MQTT 控制器
* 提供设备控制、配置更新等 API 接口
*/
@Slf4j
@RestController
@RequestMapping("/api/mqtt")
@Tag(name = "MQTT 控制接口", description = "设备控制、配置管理")
@RequiredArgsConstructor
public class MqttController {
private final MqttPublishService mqttPublishService;
/**
* 发送设备控制命令
*/
@PostMapping("/command")
@Operation(summary = "发送设备控制命令", description = "向指定设备发送控制命令")
public ResponseEntity<Map<String, Object>> sendCommand(
@Parameter(description = "设备编号") @RequestParam String deviceSn,
@Parameter(description = "命令类型") @RequestParam String command,
@Parameter(description = "命令参数") @RequestParam(required = false) String parameters) {
boolean success = mqttPublishService.sendDeviceCommand(deviceSn, command, parameters);
Map<String, Object> response = Map.of(
"success", success,
"deviceSn", deviceSn,
"command", command,
"parameters", parameters
);
return ResponseEntity.ok(response);
}
/**
* 发送设备配置更新
*/
@PostMapping("/config")
@Operation(summary = "更新设备配置", description = "更新指定设备的配置信息")
public ResponseEntity<Map<String, Object>> sendConfig(
@Parameter(description = "设备编号") @RequestParam String deviceSn,
@Parameter(description = "配置信息") @RequestBody Map<String, Object> config) {
boolean success = mqttPublishService.sendDeviceConfig(deviceSn, config);
Map<String, Object> response = Map.of(
"success", success,
"deviceSn", deviceSn,
"config", config
);
return ResponseEntity.ok(response);
}
/**
* 批量发送设备配置
*/
@PostMapping("/config/batch")
@Operation(summary = "批量更新设备配置", description = "批量更新多个设备的配置信息")
public ResponseEntity<Map<String, Object>> batchSendConfig(
@Parameter(description = "设备配置映射") @RequestBody Map<String, Map<String, Object>> deviceConfigs) {
boolean success = mqttPublishService.batchSendConfig(deviceConfigs);
Map<String, Object> response = Map.of(
"success", success,
"deviceCount", deviceConfigs.size(),
"configs", deviceConfigs
);
return ResponseEntity.ok(response);
}
/**
* 获取 MQTT 连接状态
*/
@GetMapping("/status")
@Operation(summary = "获取 MQTT 连接状态", description = "检查 MQTT 客户端连接状态")
public ResponseEntity<Map<String, Object>> getMqttStatus() {
// 这里可以添加实际的连接状态检查逻辑
Map<String, Object> status = Map.of(
"connected", true,
"clientId", "water-data-engine",
"topics", Map.of(
"iot-telemetry", "iot/telemetry/+",
"iot-command", "iot/command/+",
"quality-data", "quality/data/+"
)
);
return ResponseEntity.ok(status);
}
}
@@ -0,0 +1,116 @@
package com.water.data_engine.controller;
import com.water.data_engine.entity.TariffLadderConfig;
import com.water.data_engine.entity.BillMain;
import com.water.data_engine.service.TariffService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* 阶梯水价计算控制器
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Slf4j
@RestController
@RequiredArgsConstructor
@RequestMapping("/api/v1/tariff")
public class TariffController {
private final TariffService tariffService;
/**
* 获取有效的阶梯水价配置
*/
@GetMapping("/config")
public ResponseEntity<TariffLadderConfig> getValidTariffConfig(
@RequestParam String waterType,
@RequestParam(required = false) String areaCode) {
log.info("获取阶梯水价配置: waterType={}, areaCode={}", waterType, areaCode);
TariffLadderConfig config = tariffService.getValidTariffConfig(waterType, areaCode);
return ResponseEntity.ok(config);
}
/**
* 计算阶梯水费
*/
@PostMapping("/calculate")
public ResponseEntity<Map<String, BigDecimal>> calculateLadderWaterFee(
@RequestParam BigDecimal consumption,
@RequestParam String waterType,
@RequestParam(required = false) String areaCode) {
log.info("计算阶梯水费: consumption={}, waterType={}, areaCode={}", consumption, waterType, areaCode);
Map<String, BigDecimal> result = tariffService.calculateLadderWaterFee(consumption, waterType, areaCode);
return ResponseEntity.ok(result);
}
/**
* 生成水费账单
*/
@PostMapping("/bill/generate")
public ResponseEntity<BillMain> generateWaterBill(
@RequestParam String accountNo,
@RequestParam LocalDate billingPeriodStart,
@RequestParam LocalDate billingPeriodEnd) {
log.info("生成水费账单: accountNo={}, period={}-{}", accountNo, billingPeriodStart, billingPeriodEnd);
BillMain bill = tariffService.generateWaterBill(accountNo, billingPeriodStart, billingPeriodEnd);
return ResponseEntity.ok(bill);
}
/**
* 获取客户账单列表
*/
@GetMapping("/bills/{accountNo}")
public ResponseEntity<List<BillMain>> getCustomerBills(@PathVariable String accountNo) {
log.info("获取客户账单列表: accountNo={}", accountNo);
List<BillMain> bills = tariffService.getCustomerBills(accountNo);
return ResponseEntity.ok(bills);
}
/**
* 获取账单详情
*/
@GetMapping("/bills/{billId}/details")
public ResponseEntity<BillMain> getBillDetails(@PathVariable Long billId) {
log.info("获取账单详情: billId={}", billId);
BillMain bill = tariffService.getBillDetails(billId);
return ResponseEntity.ok(bill);
}
/**
* 获取基本水费
*/
@GetMapping("/basic-fee")
public ResponseEntity<Map<String, BigDecimal>> calculateBasicWaterFee(
@RequestParam BigDecimal consumption,
@RequestParam String meterCaliber) {
BigDecimal basicFee = tariffService.calculateBasicWaterFee(consumption, meterCaliber);
Map<String, BigDecimal> result = new HashMap<>();
result.put("basicFee", basicFee);
result.put("consumption", consumption);
return ResponseEntity.ok(result);
}
/**
* 获取附加费
*/
@GetMapping("/surcharge")
public ResponseEntity<Map<String, BigDecimal>> calculateSurchargeFee(
@RequestParam BigDecimal basicFee,
@RequestParam BigDecimal ladderFee,
@RequestParam String waterType) {
BigDecimal surchargeFee = tariffService.calculateSurchargeFee(basicFee, ladderFee, waterType);
Map<String, BigDecimal> result = new HashMap<>();
result.put("surchargeFee", surchargeFee);
result.put("basicFee", basicFee);
result.put("ladderFee", ladderFee);
return ResponseEntity.ok(result);
}
}
@@ -0,0 +1,78 @@
package com.water.data_engine.entity;
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.time.LocalDate;
import java.time.LocalDateTime;
/**
* 账单周期实体
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Data
@EqualsAndHashCode(callSuper = true)
@TableName("bill_cycle")
public class BillCycle extends com.water.common.core.entity.BaseEntity {
/**
* 周期名称
*/
private String cycleName;
/**
* 周期代码
*/
private String cycleCode;
/**
* 周期类型: monthly/quarterly/yearly/custom
*/
private String cycleType;
/**
* 周期长度(月)
*/
private Integer cycleLength;
/**
* 开始日期
*/
private LocalDate startDate;
/**
* 结束日期
*/
private LocalDate endDate;
/**
* 抄表开始日期
*/
private LocalDate readStartDate;
/**
* 抄表结束日期
*/
private LocalDate readEndDate;
/**
* 账单生成日期
*/
private LocalDate billDate;
/**
* 缴费截止日期
*/
private LocalDate dueDate;
/**
* 是否激活
*/
private Boolean isActive = true;
/**
* 描述
*/
private String description;
}
@@ -0,0 +1,84 @@
package com.water.data_engine.entity;
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.math.BigDecimal;
import java.time.LocalDate;
/**
* 账单明细实体
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Data
@EqualsAndHashCode(callSuper = true)
@TableName("bill_detail")
public class BillDetail extends com.water.common.core.entity.BaseEntity {
/**
* 账单ID
*/
private Long billId;
/**
* 账单
*/
@TableField(exist = false)
private BillMain bill;
/**
* 明细类型: basic/ladder/surcharge
*/
private String detailType;
/**
* 项目名称
*/
private String itemName;
/**
* 单位
*/
private String unit;
/**
* 数量
*/
private BigDecimal quantity;
/**
* 单价
*/
private BigDecimal unitPrice;
/**
* 金额
*/
private BigDecimal amount;
/**
* 开始读数
*/
private BigDecimal startReading;
/**
* 结束读数
*/
private BigDecimal endReading;
/**
* 用水量
*/
private BigDecimal waterVolume;
/**
* 阶梯序号
*/
private Integer stepNumber;
/**
* 备注
*/
private String remarks;
}
@@ -0,0 +1,125 @@
package com.water.data_engine.entity;
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
/**
* 账单主表实体
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Data
@EqualsAndHashCode(callSuper = true)
@TableName("bill_main")
public class BillMain extends com.water.common.core.entity.BaseEntity {
/**
* 账单号
*/
private String billNo;
/**
* 用户编号
*/
private String accountNo;
/**
* 客户姓名
*/
private String customerName;
/**
* 周期ID
*/
private Long cycleId;
/**
* 账单周期
*/
@TableField(exist = false)
private BillCycle cycle;
/**
* 计费周期开始
*/
private LocalDate billingPeriodStart;
/**
* 计费周期结束
*/
private LocalDate billingPeriodEnd;
/**
* 期初读数
*/
private BigDecimal meterReadingStart;
/**
* 期末读数
*/
private BigDecimal meterReadingEnd;
/**
* 用水量
*/
private BigDecimal waterConsumption;
/**
* 基本水费
*/
private BigDecimal basicWaterFee;
/**
* 阶梯水费
*/
private BigDecimal ladderWaterFee;
/**
* 附加费
*/
private BigDecimal surchargeFee;
/**
* 总金额
*/
private BigDecimal totalAmount;
/**
* 状态: generated/sent/paid/overdue
*/
private String status;
/**
* 发送日期
*/
private LocalDate sentDate;
/**
* 到期日期
*/
private LocalDate dueDate;
/**
* 支付方式: cash/bank/alipay/wechat
*/
private String paymentMethod;
/**
* 支付日期
*/
private LocalDate paymentDate;
/**
* 支付金额
*/
private BigDecimal paymentAmount;
/**
* 备注
*/
private String notes;
}
@@ -0,0 +1,89 @@
package com.water.data_engine.entity;
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
/**
* 客户账户实体
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Data
@EqualsAndHashCode(callSuper = true)
@TableName("customer_account")
public class CustomerAccount extends com.water.common.core.entity.BaseEntity {
/**
* 户号
*/
private String accountNo;
/**
* 客户姓名
*/
private String customerName;
/**
* 联系电话
*/
private String phone;
/**
* 地址
*/
private String address;
/**
* 水表类型
*/
private String meterType;
/**
* 水表口径
*/
private String meterCaliber;
/**
* 用水性质: residential/commercial/industrial
*/
private String waterUsageType;
/**
* 区域代码
*/
private String areaCode;
/**
* 基本水量
*/
private BigDecimal basicWaterAmount;
/**
* 是否激活
*/
private Boolean isActive = true;
/**
* 开户日期
*/
private LocalDate openDate;
/**
* 上次抄表日期
*/
private LocalDateTime lastReadDate;
/**
* 上次抄表读数
*/
private BigDecimal lastReading;
/**
* 累计用量
*/
private BigDecimal totalConsumption;
}
@@ -0,0 +1,20 @@
package com.water.data_engine.entity;
import lombok.Data;
import java.time.LocalDateTime;
@Data
public class IotData {
private Long id;
private String deviceSn;
private String deviceType;
private Double pressure;
private Double flow;
private Double temperature;
private Double waterLevel;
private Double水质指标;
private LocalDateTime collectTime;
private Integer status;
private String location;
private String remarks;
}
@@ -0,0 +1,103 @@
package com.water.data_engine.entity;
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.math.BigDecimal;
import java.time.LocalDateTime;
/**
* 水表信息实体
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Data
@EqualsAndHashCode(callSuper = true)
@TableName("meter_info")
public class MeterInfo extends com.water.common.core.entity.BaseEntity {
/**
* 用户编号
*/
private String accountNo;
/**
* 水表编号
*/
private String meterNo;
/**
* 水表类型: mechanical/digital/smart
*/
private String meterType;
/**
* 水表口径: DN15/DN20/DN25/DN32/DN40/DN50/DN80/DN100/DN150/DN200
*/
private String meterCaliber;
/**
* 水表位置
*/
private String location;
/**
* 安装日期
*/
private LocalDateTime installDate;
/**
* 初始读数
*/
private BigDecimal initialReading;
/**
* 当前读数
*/
private BigDecimal currentReading;
/**
* 上次抄表读数
*/
private BigDecimal lastReading;
/**
* 水表状态: active/inactive/maintaining/replaced
*/
private String status;
/**
* 水表品牌
*/
private String brand;
/**
* 水表型号
*/
private String model;
/**
* 通讯协议: NB-IoT/LoRaWAN/4G/RS485/M-BUS
*/
private String protocol;
/**
* 设备地址/IMEI
*/
private String deviceAddress;
/**
* 最后在线时间
*/
private LocalDateTime lastOnlineTime;
/**
* 电池状态: normal/low/replace
*/
private String batteryStatus;
/**
* 信号强度
*/
private Integer signalStrength;
}
@@ -0,0 +1,98 @@
package com.water.data_engine.entity;
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.math.BigDecimal;
import java.time.LocalDateTime;
/**
* 抄表记录实体
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Data
@EqualsAndHashCode(callSuper = true)
@TableName("meter_read_record")
public class MeterReadRecord extends com.water.common.core.entity.BaseEntity {
/**
* 用户编号
*/
private String accountNo;
/**
* 水表编号
*/
private String meterNo;
/**
* 抄表日期
*/
private LocalDateTime readDate;
/**
* 本次读数
*/
private BigDecimal readValue;
/**
* 上次读数
*/
private BigDecimal lastReadValue;
/**
* 用水量(本次读数-上次读数)
*/
private BigDecimal readDifference;
/**
* 抄表类型: manual/auto/remote
*/
private String readType;
/**
* 抄表方式: field/phone/web/iot
*/
private String readMethod;
/**
* 抄表员姓名
*/
private String readerName;
/**
* 抄表员ID
*/
private String readerId;
/**
* 备注
*/
private String remarks;
/**
* 数据质量: normal/abnormal/verify
*/
private String dataQuality;
/**
* 照片路径
*/
private String photoPath;
/**
* 是否已验证
*/
private Boolean isVerified = false;
/**
* 验证人
*/
private String verifiedBy;
/**
* 验证时间
*/
private LocalDateTime verifiedAt;
}
@@ -0,0 +1,88 @@
package com.water.data_engine.entity;
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.time.LocalDate;
import java.time.LocalDateTime;
/**
* 抄表任务实体
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Data
@EqualsAndHashCode(callSuper = true)
@TableName("meter_read_task")
public class MeterReadTask extends com.water.common.core.entity.BaseEntity {
/**
* 任务名称
*/
private String taskName;
/**
* 任务类型: regular/remote/batch
*/
private String taskType;
/**
* 任务描述
*/
private String description;
/**
* 执行日期
*/
private LocalDate executeDate;
/**
* 计划开始时间
*/
private LocalDateTime planStartTime;
/**
* 计划结束时间
*/
private LocalDateTime planEndTime;
/**
* 任务状态: pending/progress/completed/failed
*/
private String status;
/**
* 分配抄表员
*/
private String assignee;
/**
* 任务优先级: high/medium/low
*/
private String priority;
/**
* 抄表区域
*/
private String area;
/**
* 预计抄表数量
*/
private Integer estimatedCount;
/**
* 实际抄表数量
*/
private Integer actualCount;
/**
* 完成率
*/
private BigDecimal completionRate;
/**
* 异常数量
*/
private Integer abnormalCount;
}
@@ -0,0 +1,66 @@
package com.water.data_engine.entity;
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
/**
* 阶梯水价配置实体
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Data
@EqualsAndHashCode(callSuper = true)
@TableName("tariff_ladder_config")
public class TariffLadderConfig extends com.water.common.core.entity.BaseEntity {
/**
* 配置名称
*/
private String configName;
/**
* 配置代码
*/
private String configCode;
/**
* 水类型: residential/commercial/industrial
*/
private String waterType;
/**
* 区域代码(空表示全区域)
*/
private String areaCode;
/**
* 开始日期
*/
private LocalDate startDate;
/**
* 结束日期
*/
private LocalDate endDate;
/**
* 描述
*/
private String description;
/**
* 是否激活
*/
private Boolean isActive = true;
/**
* 阶梯详情
*/
@TableField(exist = false)
private List<TariffLadderDetail> details;
}
@@ -0,0 +1,54 @@
package com.water.data_engine.entity;
import com.baomidou.mybatisplus.annotation.*;
import lombok.Data;
import lombok.EqualsAndHashCode;
import java.math.BigDecimal;
import java.time.LocalDateTime;
/**
* 阶梯水价详情实体
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Data
@EqualsAndHashCode(callSuper = true)
@TableName("tariff_ladder_detail")
public class TariffLadderDetail extends com.water.common.core.entity.BaseEntity {
/**
* 配置ID
*/
private Long configId;
/**
* 阶梯序号
*/
private Integer step;
/**
* 起始水量
*/
private BigDecimal startVolume;
/**
* 结束水量(null表示无上限)
*/
private BigDecimal endVolume;
/**
* 单价
*/
private BigDecimal unitPrice;
/**
* 是否包含起始量
*/
private Boolean includeStart = true;
/**
* 配置关联
*/
@TableField(exist = false)
private TariffLadderConfig config;
}
@@ -0,0 +1,103 @@
package com.water.data_engine.enumeration;
/**
* 数据指标类型枚举
* 用于规范物联网数据的指标定义
*/
public enum MetricType {
// 设备基础指标
DEVICE_STATUS("设备状态", "正常/异常/离线"),
DEVICE_BATTERY("电池电量", "百分比"),
DEVICE_SIGNAL("信号强度", "dBm"),
// 水表指标
WATER_FLOW("瞬时流量", "立方米/小时"),
WATER_PRESSURE("水压", "MPa"),
WATER_TEMPERATURE("水温", "℃"),
WATER_LEVEL("水位", "米"),
WATER_CONSUMPTION("累计用水量", "立方米"),
// 水质指标
WATER_TURBIDITY("浊度", "NTU"),
WATER_PH("PH值", ""),
WATER_RESIDUAL_CHLORINE("余氯", "mg/L"),
WATER_TOTAL_CHLORINE("总氯", "mg/L"),
WATER_TOTAL_HARDNESS("总硬度", "mg/L"),
// 管道指标
PIPE_PRESSURE("管道压力", "MPa"),
PIPE_FLOW("管道流量", "立方米/小时"),
PIPE_TEMPERATURE("管道温度", "℃"),
PIPE_LEAKAGE("管道泄漏", "是/否"),
// 阀门指标
VALVE_POSITION("阀门开度", "%"),
VALVE_STATUS("阀门状态", "开/关/故障"),
VALVE_PRESSURE("阀门前后压差", "MPa"),
// 水泵指标
PUMP_STATUS("水泵状态", "运行/停止/故障"),
PUMP_FLOW("水泵流量", "立方米/小时"),
PUMP_CURRENT("水泵电流", "A"),
PUMP_POWER("水泵功率", "kW"),
PUMP_TEMPERATURE("水泵温度", "℃"),
// 环境指标
AMBIENT_TEMPERATURE("环境温度", "℃"),
AMBIENT_HUMIDITY("环境湿度", "%RH"),
AMBIENT_PRESSURE("环境气压", "kPa"),
// 其他指标
ERROR_CODE("错误代码", ""),
ERROR_MESSAGE("错误信息", ""),
TIMESTAMP("采集时间戳", "毫秒");
private final String description;
private final String unit;
MetricType(String description, String unit) {
this.description = description;
this.unit = unit;
}
public String getDescription() {
return description;
}
public String getUnit() {
return unit;
}
/**
* 根据指标名称获取枚举值
*/
public static MetricType fromName(String name) {
if (name == null) return null;
for (MetricType type : values()) {
if (type.name().equalsIgnoreCase(name)) {
return type;
}
}
return null;
}
/**
* 判断是否为水质相关指标
*/
public boolean isWaterQuality() {
return this == WATER_TURBIDITY || this == WATER_PH ||
this == WATER_RESIDUAL_CHLORINE || this == WATER_TOTAL_CHLORINE ||
this == WATER_TOTAL_HARDNESS;
}
/**
* 判断是否为设备状态指标
*/
public boolean isDeviceStatus() {
return this == DEVICE_STATUS || this == DEVICE_BATTERY ||
this == DEVICE_SIGNAL || this == PUMP_STATUS ||
this == VALVE_STATUS;
}
}
@@ -0,0 +1,48 @@
package com.water.data_engine.listener;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.water.data_engine.entity.IotData;
import com.water.data_engine.service.TDengineService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
@Slf4j
@Component
public class IotDataKafkaListener {
@Autowired
private TDengineService tdengineService;
@Autowired
private ObjectMapper objectMapper;
@KafkaListener(topics = "iot-data-topic", groupId = "data-engine-group")
public void consumeIotData(String message) {
try {
log.info("接收到 Kafka 消息: {}", message);
// 解析 JSON 消息
IotData iotData = objectMapper.readValue(message, IotData.class);
// 设置默认值
if (iotData.getCollectTime() == null) {
iotData.setCollectTime(LocalDateTime.now());
}
if (iotData.getStatus() == null) {
iotData.setStatus(1); // 默认正常状态
}
// 写入 TDengine
tdengineService.insertIotData(iotData);
log.info("IoT 数据处理完成: 设备={}, 时间={}",
iotData.getDeviceSn(), iotData.getCollectTime());
} catch (Exception e) {
log.error("处理 IoT 数据失败: {}", e.getMessage(), e);
}
}
}
@@ -0,0 +1,11 @@
package com.water.data_engine.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.water.data_engine.entity.CustomerAccount;
/**
* 客户账户Mapper接口
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
public interface CustomerAccountMapper extends BaseMapper<CustomerAccount> {
}
@@ -0,0 +1,11 @@
package com.water.data_engine.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.water.data_engine.entity.MeterInfo;
/**
* 水表信息Mapper接口
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
public interface MeterInfoMapper extends BaseMapper<MeterInfo> {
}
@@ -0,0 +1,11 @@
package com.water.data_engine.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.water.data_engine.entity.MeterReadRecord;
/**
* 抄表记录Mapper接口
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
public interface MeterReadRecordMapper extends BaseMapper<MeterReadRecord> {
}
@@ -0,0 +1,11 @@
package com.water.data_engine.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.water.data_engine.entity.MeterReadTask;
/**
* 抄表任务Mapper接口
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
public interface MeterReadTaskMapper extends BaseMapper<MeterReadTask> {
}
@@ -0,0 +1,11 @@
package com.water.data_engine.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.water.data_engine.entity.TariffLadderConfig;
/**
* 阶梯水价配置Mapper接口
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
public interface TariffLadderConfigMapper extends BaseMapper<TariffLadderConfig> {
}
@@ -0,0 +1,11 @@
package com.water.data_engine.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.water.data_engine.entity.TariffLadderDetail;
/**
* 阶梯水价详情Mapper接口
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
public interface TariffLadderDetailMapper extends BaseMapper<TariffLadderDetail> {
}
@@ -39,6 +39,13 @@ public class DataCollectService {
private final SimpMessagingTemplate wsMessagingTemplate;
private final ObjectMapper mapper = new ObjectMapper();
/**
* 获取 JdbcTemplate,供其他服务使用
*/
public JdbcTemplate getJdbcTemplate() {
return jdbcTemplate;
}
// ==================== 实时流采集 ====================
/**
@@ -47,6 +54,11 @@ public class DataCollectService {
*/
public String ingestRealtime(String sourceType, String sourceId, Map<String, Object> rawData) {
try {
// 数据验证
if (!validateData(sourceType, rawData)) {
throw new RuntimeException("数据验证失败: sourceType=" + sourceType);
}
Map<String, Object> envelope = buildEnvelope(sourceType, sourceId, rawData);
String json = mapper.writeValueAsString(envelope);
@@ -68,6 +80,26 @@ public class DataCollectService {
throw new RuntimeException("数据接入失败: " + e.getMessage());
}
}
/**
* 数据验证
*/
public boolean validateData(String sourceType, Map<String, Object> rawData) {
try {
switch (sourceType.toLowerCase()) {
case "iot":
case "mqtt":
return DataValidationUtils.validateTelemetryData(rawData);
case "quality":
return DataValidationUtils.validateQualityData(rawData);
default:
return rawData != null && !rawData.isEmpty();
}
} catch (Exception e) {
log.error("数据验证异常: {}", e.getMessage());
return false;
}
}
/**
* Kafka 消费者:处理 IoT 设备遥测数据
@@ -0,0 +1,173 @@
package com.water.data_engine.service;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* 数据统计服务
* 提供数据采集统计、质量分析等功能
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class DataStatisticsService {
private final JdbcTemplate jdbcTemplate;
private final DataCollectService dataCollectService;
/**
* 获取数据采集统计信息
*/
public Map<String, Object> getDataStatistics(String startTime, String endTime) {
Map<String, Object> stats = new HashMap<>();
// 默认查询最近24小时
if (startTime == null) {
startTime = LocalDateTime.now().minusHours(24).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
}
if (endTime == null) {
endTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
}
try {
// 总采集量统计
String totalSql = "SELECT COUNT(*) as total FROM collect_record WHERE collect_time BETWEEN ? AND ?";
Integer total = jdbcTemplate.queryForObject(totalSql, Integer.class, startTime, endTime);
stats.put("totalRecords", total);
// 成功/失败统计
String successSql = "SELECT COUNT(*) as success FROM collect_record WHERE status = 'success' AND collect_time BETWEEN ? AND ?";
Integer success = jdbcTemplate.queryForObject(successSql, Integer.class, startTime, endTime);
stats.put("successRecords", success);
String failSql = "SELECT COUNT(*) as failed FROM collect_record WHERE status = 'failed' AND collect_time BETWEEN ? AND ?";
Integer failed = jdbcTemplate.queryForObject(failSql, Integer.class, startTime, endTime);
stats.put("failedRecords", failed);
// 成功率
double successRate = total > 0 ? (double) success / total * 100 : 0;
stats.put("successRate", String.format("%.2f%%", successRate));
// 按来源统计
String sourceSql = "SELECT source_type, COUNT(*) as count FROM collect_record WHERE collect_time BETWEEN ? AND ? GROUP BY source_type";
List<Map<String, Object>> sourceStats = jdbcTemplate.queryForList(sourceSql, startTime, endTime);
stats.put("sourceStats", sourceStats);
// 按小时统计趋势
String trendSql = "SELECT DATE_TRUNC('hour', collect_time) as hour, COUNT(*) as count " +
"FROM collect_record WHERE collect_time BETWEEN ? AND ? GROUP BY hour ORDER BY hour";
List<Map<String, Object>> trendStats = jdbcTemplate.queryForList(trendSql, startTime, endTime);
stats.put("hourlyTrend", trendStats);
// 数据质量评分
String qualitySql = "SELECT AVG(CASE WHEN status = 'success' THEN 100 ELSE 0 END) as avgQuality " +
"FROM collect_record WHERE collect_time BETWEEN ? AND ?";
Double avgQuality = jdbcTemplate.queryForObject(qualitySql, Double.class, startTime, endTime);
stats.put("avgDataQuality", String.format("%.2f", avgQuality));
log.info("获取数据统计成功: total={}, success={}, failed={}", total, success, failed);
} catch (Exception e) {
log.error("获取数据统计失败: {}", e.getMessage());
throw new RuntimeException("数据统计查询失败: " + e.getMessage());
}
return stats;
}
/**
* 获取设备数据统计
*/
public Map<String, Object> getDeviceStatistics(String deviceSn, String startTime, String endTime) {
Map<String, Object> stats = new HashMap<>();
if (deviceSn == null || deviceSn.trim().isEmpty()) {
throw new IllegalArgumentException("设备编号不能为空");
}
if (startTime == null) {
startTime = LocalDateTime.now().minusHours(24).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
}
if (endTime == null) {
endTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
}
try {
// 设备数据总量
String deviceSql = "SELECT COUNT(*) as total FROM collect_record WHERE source_key = ? AND collect_time BETWEEN ? AND ?";
Integer deviceTotal = jdbcTemplate.queryForObject(deviceSql, Integer.class, deviceSn, startTime, endTime);
stats.put("deviceTotal", deviceTotal);
// 设备数据趋势
String trendSql = "SELECT DATE_TRUNC('hour', collect_time) as hour, COUNT(*) as count " +
"FROM collect_record WHERE source_key = ? AND collect_time BETWEEN ? AND ? " +
"GROUP BY hour ORDER BY hour";
List<Map<String, Object>> deviceTrend = jdbcTemplate.queryForList(trendSql, deviceSn, startTime, endTime);
stats.put("deviceTrend", deviceTrend);
// 最近数据状态
String recentSql = "SELECT status, collect_time FROM collect_record " +
"WHERE source_key = ? ORDER BY collect_time DESC LIMIT 5";
List<Map<String, Object>> recentStatus = jdbcTemplate.queryForList(recentSql, deviceSn);
stats.put("recentStatus", recentStatus);
log.info("获取设备 {} 数据统计成功: total={}", deviceSn, deviceTotal);
} catch (Exception e) {
log.error("获取设备 {} 数据统计失败: {}", deviceSn, e.getMessage());
throw new RuntimeException("设备数据统计查询失败: " + e.getMessage());
}
return stats;
}
/**
* 获取错误数据统计
*/
public Map<String, Object> getErrorStatistics(String startTime, String endTime) {
Map<String, Object> stats = new HashMap<>();
if (startTime == null) {
startTime = LocalDateTime.now().minusHours(24).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
}
if (endTime == null) {
endTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
}
try {
// 错误数据总量
String errorSql = "SELECT COUNT(*) as total FROM collect_record WHERE status = 'failed' AND collect_time BETWEEN ? AND ?";
Integer errorTotal = jdbcTemplate.queryForObject(errorSql, Integer.class, startTime, endTime);
stats.put("errorTotal", errorTotal);
// 错误分布统计
String errorDistSql = "SELECT source_type, COUNT(*) as count FROM collect_record " +
"WHERE status = 'failed' AND collect_time BETWEEN ? AND ? GROUP BY source_type";
List<Map<String, Object>> errorDist = jdbcTemplate.queryForList(errorDistSql, startTime, endTime);
stats.put("errorDistribution", errorDist);
// 常见错误类型统计
String commonErrorSql = "SELECT error_msg, COUNT(*) as count FROM collect_record " +
"WHERE status = 'failed' AND collect_time BETWEEN ? AND ? " +
"GROUP BY error_msg ORDER BY count DESC LIMIT 10";
List<Map<String, Object>> commonErrors = jdbcTemplate.queryForList(commonErrorSql, startTime, endTime);
stats.put("commonErrors", commonErrors);
log.info("获取错误统计成功: total={}", errorTotal);
} catch (Exception e) {
log.error("获取错误统计失败: {}", e.getMessage());
throw new RuntimeException("错误统计查询失败: " + e.getMessage());
}
return stats;
}
}
@@ -0,0 +1,109 @@
package com.water.data_engine.service;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.water.data_engine.entity.MeterInfo;
import com.water.data_engine.entity.MeterReadRecord;
import com.water.data_engine.entity.MeterReadTask;
import com.water.data_engine.entity.CustomerAccount;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
/**
* 抄表管理服务接口
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
public interface MeterReadService {
/**
* 创建抄表记录
*/
MeterReadRecord createReadRecord(MeterReadRecord record);
/**
* 更新抄表记录
*/
MeterReadRecord updateReadRecord(MeterReadRecord record);
/**
* 获取抄表记录详情
*/
MeterReadRecord getReadRecordById(Long id);
/**
* 获取抄表记录列表
*/
List<MeterReadRecord> getReadRecords(Map<String, Object> params);
/**
* 获取抄表记录分页
*/
Page<MeterReadRecord> getReadRecordPage(Page<MeterReadRecord> page, Map<String, Object> params);
/**
* 删除抄表记录
*/
boolean deleteReadRecord(Long id);
/**
* 验证抄表记录
*/
MeterReadRecord verifyReadRecord(Long id, String verifiedBy);
/**
* 批量创建抄表记录
*/
List<MeterReadRecord> batchCreateReadRecords(List<MeterReadRecord> records);
/**
* 创建抄表任务
*/
MeterReadTask createReadTask(MeterReadTask task);
/**
* 更新抄表任务
*/
MeterReadTask updateReadTask(MeterReadTask task);
/**
* 获取抄表任务详情
*/
MeterReadTask getReadTaskById(Long id);
/**
* 获取抄表任务列表
*/
List<MeterReadTask> getReadTasks(Map<String, Object> params);
/**
* 启动抄表任务
*/
boolean startReadTask(Long taskId);
/**
* 完成抄表任务
*/
boolean completeReadTask(Long taskId, Integer actualCount, Integer abnormalCount);
/**
* 远程抄表
*/
MeterReadRecord remoteRead(String meterNo, BigDecimal readValue);
/**
* 计算用水量
*/
BigDecimal calculateWaterConsumption(BigDecimal currentReading, BigDecimal lastReading);
/**
* 验证读数合理性
*/
boolean validateReading(BigDecimal newReading, String meterNo);
/**
* 获取异常抄表记录
*/
List<MeterReadRecord> getAbnormalRecords(Map<String, Object> params);
}
@@ -0,0 +1,100 @@
package com.water.data_engine.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.HashMap;
import java.util.Map;
/**
* MQTT 消息发布服务
* 用于向 IoT 设备发送控制命令和配置信息
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class MqttPublishService {
private final MqttClient mqttClient;
private final ObjectMapper objectMapper;
/**
* 发送设备控制命令
*/
public boolean sendDeviceCommand(String deviceSn, String command, String parameters) {
try {
Map<String, Object> payload = new HashMap<>();
payload.put("deviceSn", deviceSn);
payload.put("command", command);
payload.put("parameters", parameters);
payload.put("timestamp", System.currentTimeMillis());
String topic = "iot/command/" + deviceSn;
String jsonPayload = objectMapper.writeValueAsString(payload);
MqttMessage message = new MqttMessage(jsonPayload.getBytes());
message.setQos(1);
message.setRetained(false);
mqttClient.publish(topic, message);
log.info("发送 MQTT 控制命令: device={}, command={}, topic={}", deviceSn, command, topic);
return true;
} catch (Exception e) {
log.error("发送 MQTT 控制命令失败: {}", e.getMessage());
return false;
}
}
/**
* 发送设备配置更新
*/
public boolean sendDeviceConfig(String deviceSn, Map<String, Object> config) {
try {
Map<String, Object> payload = new HashMap<>();
payload.put("deviceSn", deviceSn);
payload.put("config", config);
payload.put("timestamp", System.currentTimeMillis());
String topic = "iot/config/" + deviceSn;
String jsonPayload = objectMapper.writeValueAsString(payload);
MqttMessage message = new MqttMessage(jsonPayload.getBytes());
message.setQos(1);
message.setRetained(true);
mqttClient.publish(topic, message);
log.info("发送 MQTT 设备配置: device={}, topic={}", deviceSn, topic);
return true;
} catch (Exception e) {
log.error("发送 MQTT 设备配置失败: {}", e.getMessage());
return false;
}
}
/**
* 批量发送配置更新
*/
public boolean batchSendConfig(Map<String, Map<String, Object>> deviceConfigs) {
int successCount = 0;
int totalCount = deviceConfigs.size();
for (Map.Entry<String, Map<String, Object>> entry : deviceConfigs.entrySet()) {
String deviceSn = entry.getKey();
Map<String, Object> config = entry.getValue();
if (sendDeviceConfig(deviceSn, config)) {
successCount++;
}
}
log.info("批量发送配置完成: {}/{} 成功", successCount, totalCount);
return successCount == totalCount;
}
}
@@ -0,0 +1,185 @@
package com.water.data_engine.service;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.water.data_engine.config.MqttConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Service;
import java.util.Map;
/**
* MQTT 消息服务
* 支持物联网遥测数据、控制命令、水质数据的实时接收
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class MqttService {
private final MqttConfig mqttConfig;
private final DataCollectService dataCollectService;
private final ObjectMapper objectMapper;
private final MqttPahoClientFactory mqttClientFactory;
/**
* MQTT 消息输入通道
*/
@Bean
public MessageChannel mqttInputChannel() {
return new DirectChannel();
}
/**
* MQTT 消息消费者
*/
@Bean
public MessageProducer inbound() {
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
mqttConfig.getClientId() + "-consumer",
mqttClientFactory(),
mqttConfig.getTopic().getIotTelemetry(),
mqttConfig.getTopic().getIotCommand(),
mqttConfig.getTopic().getQualityData()
);
adapter.setCompletionTimeout(5000);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
adapter.setOutputChannel(mqttInputChannel());
return adapter;
}
/**
* 消息处理入口
*/
@ServiceActivator(inputChannel = "mqttInputChannel")
public void handleMessage(Message<?> message) throws Exception {
String topic = message.getHeaders().get("mqtt_topic").toString();
String payload = (String) message.getPayload();
log.debug("收到 MQTT 消息: topic={}, payload={}", topic, payload);
try {
switch (topic) {
case "iot/telemetry/+":
handleIotTelemetry(payload);
break;
case "iot/command/+":
handleIotCommand(payload);
break;
case "quality/data/+":
handleQualityData(payload);
break;
default:
log.warn("未知的 MQTT 主题: {}", topic);
}
} catch (Exception e) {
log.error("处理 MQTT 消息失败: topic={}, error={}", topic, e.getMessage());
throw e;
}
}
/**
* 处理物联网遥测数据
*/
private void handleIotTelemetry(String payload) throws JsonProcessingException {
@SuppressWarnings("unchecked")
Map<String, Object> data = objectMapper.readValue(payload, Map.class);
String deviceSn = (String) data.getOrDefault("deviceSn", "unknown");
@SuppressWarnings("unchecked")
List<Map<String, Object>> metrics = (List<Map<String, Object>>) data.getOrDefault("metrics", List.of());
for (Map<String, Object> metric : metrics) {
String key = (String) metric.get("key");
Object value = metric.get("value");
// 写入 TDengine
writeToTDengine(deviceSn, key, value);
// 通过 Kafka 转发到其他系统
dataCollectService.ingestRealtime("mqtt", deviceSn, data);
}
log.info("处理 IoT 遥测数据: device={}, metrics={}", deviceSn, metrics.size());
}
/**
* 处理物联网控制命令
*/
private void handleIotCommand(String payload) throws JsonProcessingException {
@SuppressWarnings("unchecked")
Map<String, Object> data = objectMapper.readValue(payload, Map.class);
String deviceSn = (String) data.getOrDefault("deviceSn", "unknown");
String command = (String) data.getOrDefault("command", "");
String parameters = (String) data.getOrDefault("parameters", "");
log.info("处理 IoT 控制命令: device={}, command={}, params={}", deviceSn, command, parameters);
// 这里可以添加具体的控制逻辑
// 例如:阀门开关、水泵启停等
}
/**
* 处理水质数据
*/
private void handleQualityData(String payload) throws JsonProcessingException {
@SuppressWarnings("unchecked")
Map<String, Object> data = objectMapper.readValue(payload, Map.class);
// 写入 PostgreSQL
String sql = """
INSERT INTO water_quality_record (test_type, test_point, point_type, area,
turbidity, ph, residual_chlorine, is_qualified, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, NOW())
""";
dataCollectService.getJdbcTemplate().update(sql,
data.get("testType"),
data.get("testPoint"),
data.get("pointType"),
data.get("area"),
data.get("turbidity"),
data.get("ph"),
data.get("residualChlorine"),
data.get("isQualified")
);
log.info("处理水质数据: point={}", data.get("testPoint"));
}
/**
* 写入 TDengine
*/
private void writeToTDengine(String deviceSn, String metricKey, Object value) {
String sql = "INSERT INTO water_iot.iot_telemetry (ts, device_sn, metric_key, metric_value, quality) VALUES (NOW, ?, ?, ?, 1)";
dataCollectService.getJdbcTemplate().update(sql, deviceSn, metricKey, value);
}
/**
* MQTT 客户端工厂
*/
public MqttPahoClientFactory getMqttClientFactory() {
return mqttClientFactory;
}
}
@@ -0,0 +1,106 @@
package com.water.data_engine.service;
import com.water.data_engine.config.TDengineConfig;
import com.water.data_engine.entity.IotData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.sql.*;
import java.time.LocalDateTime;
import java.util.List;
@Service
public class TDengineService {
@Autowired
private TDengineConfig tdengineConfig;
public Connection getConnection() throws SQLException {
return DriverManager.getConnection(tdengineConfig.getJdbcUrl());
}
public void initializeDatabase() {
String createDatabaseSql = String.format("CREATE DATABASE IF NOT EXISTS %s", tdengineConfig.getDatabase());
String useDatabaseSql = String.format("USE %s", tdengineConfig.getDatabase());
String createTableSql = "CREATE TABLE IF NOT EXISTS iot_data (" +
"id BIGINT AUTO_INCREMENT," +
"device_sn NCHAR(64) NOT NULL," +
"device_type NCHAR(32)," +
"pressure DOUBLE," +
"flow DOUBLE," +
"temperature DOUBLE," +
"water_level DOUBLE," +
"water_quality_index DOUBLE," +
"collect_time TIMESTAMP," +
"status INT," +
"location NCHAR(128)," +
"remarks NCHAR(256)," +
"PRIMARY KEY (id, device_sn, collect_time))" +
"TAGS (device_type NCHAR(32), location NCHAR(128))";
try (Connection conn = DriverManager.getConnection(
"jdbc:TAOS://" + tdengineConfig.getHost() + ":" + tdengineConfig.getPort() +
"?user=" + tdengineConfig.getUsername() + "&password=" + tdengineConfig.getPassword())) {
Statement stmt = conn.createStatement();
stmt.execute(createDatabaseSql);
stmt.execute(useDatabaseSql);
stmt.execute(createTableSql);
System.out.println("TDengine 数据库和表初始化完成");
} catch (SQLException e) {
System.err.println("初始化 TDengine 失败: " + e.getMessage());
}
}
public void insertIotData(IotData data) {
String sql = String.format("INSERT INTO iot_data VALUES (NULL, '%s', '%s', %.2f, %.2f, %.2f, %.2f, %.2f, '%s', %d, '%s', '%s')",
data.getDeviceSn(),
data.getDeviceType(),
data.getPressure(),
data.getFlow(),
data.getTemperature(),
data.getWaterLevel(),
data.get水质指标(),
data.getCollectTime().toString(),
data.getStatus(),
data.getLocation(),
data.getRemarks());
try (Connection conn = getConnection();
Statement stmt = conn.createStatement()) {
stmt.execute(sql);
System.out.println("数据已写入 TDengine: " + data.getDeviceSn());
} catch (SQLException e) {
System.err.println("写入 TDengine 失败: " + e.getMessage());
}
}
public void batchInsertIotData(List<IotData> dataList) {
try (Connection conn = getConnection()) {
conn.setAutoCommit(false);
for (IotData data : dataList) {
String sql = String.format("INSERT INTO iot_data VALUES (NULL, '%s', '%s', %.2f, %.2f, %.2f, %.2f, %.2f, '%s', %d, '%s', '%s')",
data.getDeviceSn(),
data.getDeviceType(),
data.getPressure(),
data.getFlow(),
data.getTemperature(),
data.getWaterLevel(),
data.get水质指标(),
data.getCollectTime().toString(),
data.getStatus(),
data.getLocation(),
data.getRemarks());
Statement stmt = conn.createStatement();
stmt.execute(sql);
}
conn.commit();
System.out.println("批量写入 " + dataList.size() + " 条数据到 TDengine");
} catch (SQLException e) {
System.err.println("批量写入 TDengine 失败: " + e.getMessage());
}
}
}
@@ -0,0 +1,54 @@
package com.water.data_engine.service;
import com.water.data_engine.entity.TariffLadderConfig;
import com.water.data_engine.entity.TariffLadderDetail;
import com.water.data_engine.entity.CustomerAccount;
import com.water.data_engine.entity.BillMain;
import com.water.data_engine.entity.BillDetail;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.util.List;
import java.util.Map;
/**
* 阶梯水价计算服务接口
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
public interface TariffService {
/**
* 获取有效的阶梯水价配置
*/
TariffLadderConfig getValidTariffConfig(String waterType, String areaCode);
/**
* 计算阶梯水费
*/
Map<String, BigDecimal> calculateLadderWaterFee(BigDecimal consumption, String waterType, String areaCode);
/**
* 生成水费账单
*/
BillMain generateWaterBill(String accountNo, LocalDate billingPeriodStart, LocalDate billingPeriodEnd);
/**
* 获取客户账单列表
*/
List<BillMain> getCustomerBills(String accountNo);
/**
* 获取账单详情
*/
BillMain getBillDetails(Long billId);
/**
* 计算基本水费
*/
BigDecimal calculateBasicWaterFee(BigDecimal consumption, String meterCaliber);
/**
* 计算附加费
*/
BigDecimal calculateSurchargeFee(BigDecimal basicFee, BigDecimal ladderFee, String waterType);
}
@@ -0,0 +1,181 @@
package com.water.data_engine.service.impl;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.water.data_engine.entity.MeterInfo;
import com.water.data_engine.entity.MeterReadRecord;
import com.water.data_engine.entity.MeterReadTask;
import com.water.data_engine.entity.CustomerAccount;
import com.water.data_engine.mapper.MeterInfoMapper;
import com.water.data_engine.mapper.MeterReadRecordMapper;
import com.water.data_engine.mapper.MeterReadTaskMapper;
import com.water.data_engine.mapper.CustomerAccountMapper;
import com.water.data_engine.service.MeterReadService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
/**
* 抄表管理服务实现
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class MeterReadServiceImpl implements MeterReadService {
private final MeterReadRecordMapper meterReadRecordMapper;
private final MeterReadTaskMapper meterReadTaskMapper;
private final MeterInfoMapper meterInfoMapper;
private final CustomerAccountMapper customerAccountMapper;
@Override
@Transactional(rollbackFor = Exception.class)
public MeterReadRecord createReadRecord(MeterReadRecord record) {
log.info("创建抄表记录: meterNo={}, readValue={}", record.getMeterNo(), record.getReadValue());
// 验证水表信息
MeterInfo meterInfo = meterInfoMapper.selectOne(
new LambdaQueryWrapper<MeterInfo>()
.eq(MeterInfo::getMeterNo, record.getMeterNo())
);
if (meterInfo == null) {
throw new RuntimeException("水表不存在: " + record.getMeterNo());
}
// 获取客户信息
CustomerAccount customer = customerAccountMapper.selectOne(
new LambdaQueryWrapper<CustomerAccount>()
.eq(CustomerAccount::getAccountNo, meterInfo.getAccountNo())
);
if (customer != null) {
record.setAccountNo(customer.getAccountNo());
record.setCustomerName(customer.getCustomerName());
}
// 计算用水量
BigDecimal readDifference = calculateWaterConsumption(record.getReadValue(), meterInfo.getLastReading());
record.setReadDifference(readDifference);
meterInfo.setLastReading(record.getReadValue());
meterInfo.setCurrentReading(record.getReadValue());
meterInfoMapper.updateById(meterInfo);
record.setReadDate(LocalDateTime.now());
record.setDataQuality("normal");
meterReadRecordMapper.insert(record);
return record;
}
@Override
public MeterReadRecord getReadRecordById(Long id) {
return meterReadRecordMapper.selectById(id);
}
@Override
public List<MeterReadRecord> getReadRecords(Map<String, Object> params) {
LambdaQueryWrapper<MeterReadRecord> wrapper = new LambdaQueryWrapper<>();
if (params.containsKey("accountNo")) {
wrapper.eq(MeterReadRecord::getAccountNo, params.get("accountNo"));
}
if (params.containsKey("meterNo")) {
wrapper.eq(MeterReadRecord::getMeterNo, params.get("meterNo"));
}
if (params.containsKey("readType")) {
wrapper.eq(MeterReadRecord::getReadType, params.get("readType"));
}
return meterReadRecordMapper.selectList(wrapper);
}
@Override
public Page<MeterReadRecord> getReadRecordPage(Page<MeterReadRecord> page, Map<String, Object> params) {
LambdaQueryWrapper<MeterReadRecord> wrapper = new LambdaQueryWrapper<>();
if (params.containsKey("accountNo")) {
wrapper.eq(MeterReadRecord::getAccountNo, params.get("accountNo"));
}
if (params.containsKey("meterNo")) {
wrapper.eq(MeterReadRecord::getMeterNo, params.get("meterNo"));
}
return meterReadRecordMapper.selectPage(page, wrapper);
}
@Override
@Transactional(rollbackFor = Exception.class)
public boolean deleteReadRecord(Long id) {
return meterReadRecordMapper.deleteById(id) > 0;
}
@Override
@Transactional(rollbackFor = Exception.class)
public MeterReadRecord verifyReadRecord(Long id, String verifiedBy) {
MeterReadRecord record = getReadRecordById(id);
if (record == null) {
throw new RuntimeException("抄表记录不存在: " + id);
}
record.setIsVerified(true);
record.setVerifiedBy(verifiedBy);
record.setVerifiedAt(LocalDateTime.now());
record.setDataQuality("normal");
meterReadRecordMapper.updateById(record);
return record;
}
@Override
@Transactional(rollbackFor = Exception.class)
public MeterReadRecord remoteRead(String meterNo, BigDecimal readValue) {
log.info("远程抄表: meterNo={}, readValue={}", meterNo, readValue);
MeterReadRecord record = new MeterReadRecord();
record.setMeterNo(meterNo);
record.setReadValue(readValue);
record.setReadType("remote");
record.setReadMethod("iot");
return createReadRecord(record);
}
@Override
public BigDecimal calculateWaterConsumption(BigDecimal currentReading, BigDecimal lastReading) {
if (lastReading == null) {
return currentReading;
}
return currentReading.subtract(lastReading);
}
@Override
public boolean validateReading(BigDecimal newReading, String meterNo) {
if (newReading == null || newReading.compareTo(BigDecimal.ZERO) < 0) {
return false;
}
return true;
}
@Override
public List<MeterReadRecord> getAbnormalRecords(Map<String, Object> params) {
LambdaQueryWrapper<MeterReadRecord> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(MeterReadRecord::getDataQuality, "abnormal");
if (params.containsKey("accountNo")) {
wrapper.eq(MeterReadRecord::getAccountNo, params.get("accountNo"));
}
if (params.containsKey("meterNo")) {
wrapper.eq(MeterReadRecord::getMeterNo, params.get("meterNo"));
}
return meterReadRecordMapper.selectList(wrapper);
}
}
@@ -0,0 +1,234 @@
package com.water.data_engine.service.impl;
import com.water.data_engine.entity.TariffLadderConfig;
import com.water.data_engine.entity.TariffLadderDetail;
import com.water.data_engine.entity.CustomerAccount;
import com.water.data_engine.entity.BillMain;
import com.water.data_engine.entity.BillDetail;
import com.water.data_engine.entity.MeterReadRecord;
import com.water.data_engine.entity.MeterInfo;
import com.water.data_engine.mapper.TariffLadderConfigMapper;
import com.water.data_engine.mapper.TariffLadderDetailMapper;
import com.water.data_engine.mapper.CustomerAccountMapper;
import com.water.data_engine.mapper.MeterReadRecordMapper;
import com.water.data_engine.mapper.MeterInfoMapper;
import com.water.data_engine.service.TariffService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;
/**
* 阶梯水价计算服务实现
* Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class TariffServiceImpl implements TariffService {
private final TariffLadderConfigMapper tariffLadderConfigMapper;
private final TariffLadderDetailMapper tariffLadderDetailMapper;
private final CustomerAccountMapper customerAccountMapper;
private final MeterReadRecordMapper meterReadRecordMapper;
private final MeterInfoMapper meterInfoMapper;
@Override
public TariffLadderConfig getValidTariffConfig(String waterType, String areaCode) {
LocalDate today = LocalDate.now();
// 查询当前有效的阶梯水价配置
List<TariffLadderConfig> configs = tariffLadderConfigMapper.selectList(
new com.baomidou.mybatisplus.core.conditions.query.QueryWrapper<TariffLadderConfig>()
.eq("water_type", waterType)
.eq("area_code", areaCode)
.eq("is_active", true)
.ge("start_date", today)
.le("end_date", today)
.orderByDesc("created_at")
.last("LIMIT 1")
);
if (configs.isEmpty()) {
log.warn("未找到有效的阶梯水价配置: waterType={}, areaCode={}", waterType, areaCode);
return null;
}
TariffLadderConfig config = configs.get(0);
// 加载阶梯详情
List<TariffLadderDetail> details = tariffLadderDetailMapper.selectList(
new com.baomidou.mybatisplus.core.conditions.query.QueryWrapper<TariffLadderDetail>()
.eq("config_id", config.getId())
.orderByAsc("step")
);
config.setDetails(details);
return config;
}
@Override
public Map<String, BigDecimal> calculateLadderWaterFee(BigDecimal consumption, String waterType, String areaCode) {
log.info("计算阶梯水费: consumption={}, waterType={}, areaCode={}", consumption, waterType, areaCode);
TariffLadderConfig config = getValidTariffConfig(waterType, areaCode);
if (config == null || config.getDetails() == null) {
throw new RuntimeException("未找到有效的阶梯水价配置");
}
BigDecimal ladderFee = BigDecimal.ZERO;
BigDecimal remainingConsumption = consumption;
// 按阶梯计算水费
for (TariffLadderDetail detail : config.getDetails()) {
if (remainingConsumption.compareTo(BigDecimal.ZERO) <= 0) {
break;
}
BigDecimal stepVolume = detail.getEndVolume() == null ?
remainingConsumption :
remainingConsumption.min(detail.getEndVolume().subtract(detail.getStartVolume()));
if (stepVolume.compareTo(BigDecimal.ZERO) > 0) {
BigDecimal stepFee = stepVolume.multiply(detail.getUnitPrice());
ladderFee = ladderFee.add(stepFee);
remainingConsumption = remainingConsumption.subtract(stepVolume);
}
}
Map<String, BigDecimal> result = new HashMap<>();
result.put("ladderFee", ladderFee);
result.put("consumption", consumption);
log.info("阶梯水费计算完成: ladderFee={}", ladderFee);
return result;
}
@Override
@Transactional(rollbackFor = Exception.class)
public BillMain generateWaterBill(String accountNo, LocalDate billingPeriodStart, LocalDate billingPeriodEnd) {
log.info("生成水费账单: accountNo={}, period={}-{}", accountNo, billingPeriodStart, billingPeriodEnd);
// 获取客户信息
CustomerAccount customer = customerAccountMapper.selectOne(
new com.baomidou.mybatisplus.core.conditions.query.QueryWrapper<CustomerAccount>()
.eq("account_no", accountNo)
);
if (customer == null) {
throw new RuntimeException("客户不存在: " + accountNo);
}
// 获取水表信息
MeterInfo meterInfo = meterInfoMapper.selectOne(
new com.baomidou.mybatisplus.core.conditions.query.QueryWrapper<MeterInfo>()
.eq("account_no", accountNo)
);
if (meterInfo == null) {
throw new RuntimeException("水表不存在: " + accountNo);
}
// 获取抄表记录
List<MeterReadRecord> readRecords = meterReadRecordMapper.selectList(
new com.baomidou.mybatisplus.core.conditions.query.QueryWrapper<MeterReadRecord>()
.eq("account_no", accountNo)
.ge("read_date", billingPeriodStart.atStartOfDay())
.le("read_date", billingPeriodEnd.atTime(23, 59, 59))
.orderByDesc("read_date")
.last("LIMIT 2")
);
if (readRecords.size() < 2) {
throw new RuntimeException("抄表记录不足,无法生成账单");
}
MeterReadRecord endRecord = readRecords.get(0);
MeterReadRecord startRecord = readRecords.get(1);
BigDecimal consumption = endRecord.getReadValue().subtract(startRecord.getReadValue());
// 计算各项费用
Map<String, BigDecimal> ladderResult = calculateLadderWaterFee(consumption, customer.getWaterUsageType(), customer.getAreaCode());
BigDecimal basicWaterFee = calculateBasicWaterFee(consumption, meterInfo.getMeterCaliber());
BigDecimal surchargeFee = calculateSurchargeFee(basicWaterFee, ladderResult.get("ladderFee"), customer.getWaterUsageType());
BigDecimal totalAmount = basicWaterFee.add(ladderResult.get("ladderFee")).add(surchargeFee);
// 创建主账单
BillMain billMain = new BillMain();
billMain.setBillNo("BILL" + System.currentTimeMillis());
billMain.setAccountNo(accountNo);
billMain.setCustomerName(customer.getCustomerName());
billMain.setBillingPeriodStart(billingPeriodStart);
billMain.setBillingPeriodEnd(billingPeriodEnd);
billMain.setMeterReadingStart(startRecord.getReadValue());
billMain.setMeterReadingEnd(endRecord.getReadValue());
billMain.setWaterConsumption(consumption);
billMain.setBasicWaterFee(basicWaterFee);
billMain.setLadderWaterFee(ladderResult.get("ladderFee"));
billMain.setSurchargeFee(surchargeFee);
billMain.setTotalAmount(totalAmount);
billMain.setStatus("generated");
billMain.setSentDate(LocalDate.now());
billMain.setDueDate(LocalDate.now().plusDays(30));
// 这里应该保存到数据库,简化处理,只返回对象
log.info("账单生成完成: billNo={}, totalAmount={}", billMain.getBillNo(), totalAmount);
return billMain;
}
@Override
public List<BillMain> getCustomerBills(String accountNo) {
return Collections.emptyList(); // 简化实现
}
@Override
public BillMain getBillDetails(Long billId) {
return null; // 简化实现
}
@Override
public BigDecimal calculateBasicWaterFee(BigDecimal consumption, String meterCaliber) {
// 基本水费计算,根据口径不同有不同的基础价格
BigDecimal basicRate = getBasicRateByCaliber(meterCaliber);
return consumption.multiply(basicRate);
}
@Override
public BigDecimal calculateSurchargeFee(BigDecimal basicFee, BigDecimal ladderFee, String waterType) {
// 附加费计算,通常为基本水费和阶梯水费的总和的百分比
BigDecimal surchargeRate = getSurchargeRateByType(waterType);
return basicFee.add(ladderFee).multiply(surchargeRate);
}
private BigDecimal getBasicRateByCaliber(String meterCaliber) {
// 根据水表口径获取基础价格
switch (meterCaliber) {
case "DN15": return new BigDecimal("3.5");
case "DN20": return new BigDecimal("4.5");
case "DN25": return new BigDecimal("5.5");
case "DN32": return new BigDecimal("7.0");
case "DN40": return new BigDecimal("9.0");
case "DN50": return new BigDecimal("12.0");
case "DN80": return new BigDecimal("20.0");
case "DN100": return new BigDecimal("30.0");
case "DN150": return new BigDecimal("50.0");
case "DN200": return new BigDecimal("80.0");
default: return new BigDecimal("10.0");
}
}
private BigDecimal getSurchargeRateByType(String waterType) {
// 根据用水性质获取附加费率
switch (waterType) {
case "residential": return new BigDecimal("0.10"); // 10%
case "commercial": return new BigDecimal("0.15"); // 15%
case "industrial": return new BigDecimal("0.20"); // 20%
default: return new BigDecimal("0.10");
}
}
}
@@ -0,0 +1,219 @@
package com.water.data_engine.utils;
import com.water.data_engine.enumeration.MetricType;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.regex.Pattern;
/**
* 数据验证工具类
* 用于验证物联网数据的完整性和准确性
*/
@Slf4j
public class DataValidationUtils {
// 设备编号正则表达式
private static final Pattern DEVICE_SN_PATTERN = Pattern.compile("^[A-Za-z0-9]{6,20}$");
// 数值范围验证
private static final Map<MetricType, double[]> VALID_RANGES = Map.of(
MetricType.WATER_FLOW, new double[]{0, 1000},
MetricType.WATER_PRESSURE, new double[]{0, 1.0},
MetricType.WATER_TEMPERATURE, new double[]{0, 100},
MetricType.WATER_LEVEL, new double[]{0, 100},
MetricType.WATER_CONSUMPTION, new double[]{0, 999999},
MetricType.WATER_TURBIDITY, new double[]{0, 1000},
MetricType.WATER_PH, new double[]{0, 14},
MetricType.WATER_RESIDUAL_CHLORINE, new double[]{0, 5},
MetricType.PIPE_PRESSURE, new double[]{0, 2.0},
MetricType.PIPE_FLOW, new double[]{0, 5000},
MetricType.VALVE_POSITION, new double[]{0, 100},
MetricType.PUMP_FLOW, new double[]{0, 2000},
MetricType.PUMP_CURRENT, new double[]{0, 100},
MetricType.PUMP_POWER, new double[]{0, 1000},
MetricType.AMBIENT_TEMPERATURE, new double{-40, 80},
MetricType.AMBIENT_HUMIDITY, new double[]{0, 100}
);
/**
* 验证设备编号
*/
public static boolean isValidDeviceSn(String deviceSn) {
if (deviceSn == null || deviceSn.trim().isEmpty()) {
return false;
}
return DEVICE_SN_PATTERN.matcher(deviceSn).matches();
}
/**
* 验证数据值是否在合理范围内
*/
public static boolean isValidValue(MetricType metricType, Object value) {
if (value == null) {
return false;
}
if (!VALID_RANGES.containsKey(metricType)) {
return true; // 没有范围限制的指标直接返回 true
}
try {
double numericValue = convertToDouble(value);
double[] range = VALID_RANGES.get(metricType);
return numericValue >= range[0] && numericValue <= range[1];
} catch (NumberFormatException e) {
log.warn("无法转换数据值: value={}, metricType={}", value, metricType);
return false;
}
}
/**
* 验证遥测数据包
*/
public static boolean validateTelemetryData(Map<String, Object> data) {
if (data == null || data.isEmpty()) {
log.warn("遥测数据为空");
return false;
}
// 验证设备编号
String deviceSn = (String) data.get("deviceSn");
if (!isValidDeviceSn(deviceSn)) {
log.warn("无效的设备编号: {}", deviceSn);
return false;
}
// 验证时间戳
Object timestamp = data.get("timestamp");
if (timestamp == null) {
log.warn("缺少时间戳字段");
return false;
}
// 验证指标数据
@SuppressWarnings("unchecked")
Map<String, Object> metrics = (Map<String, Object>) data.get("metrics");
if (metrics == null || metrics.isEmpty()) {
log.warn("缺少指标数据");
return false;
}
// 验证每个指标
for (Map.Entry<String, Object> entry : metrics.entrySet()) {
String metricKey = entry.getKey();
Object metricValue = entry.getValue();
MetricType metricType = MetricType.fromName(metricKey);
if (metricType != null && !isValidValue(metricType, metricValue)) {
log.warn("指标值超出合理范围: metric={}, value={}, range={}",
metricKey, metricValue, VALID_RANGES.get(metricType));
return false;
}
}
return true;
}
/**
* 验证水质数据
*/
public static boolean validateQualityData(Map<String, Object> data) {
if (data == null || data.isEmpty()) {
log.warn("水质数据为空");
return false;
}
// 必需字段验证
String[] requiredFields = {"testType", "testPoint", "pointType", "area"};
for (String field : requiredFields) {
if (!data.containsKey(field) || data.get(field) == null) {
log.warn("缺少必需字段: {}", field);
return false;
}
}
// 数值字段验证
String[] numericFields = {"turbidity", "ph", "residualChlorine"};
for (String field : numericFields) {
Object value = data.get(field);
if (value != null) {
try {
double numericValue = convertToDouble(value);
// 特殊验证水质指标
if (field.equals("ph") && (numericValue < 0 || numericValue > 14)) {
log.warn("PH值超出合理范围: {}", numericValue);
return false;
}
if (field.equals("residualChlorine") && numericValue < 0) {
log.warn("余氯值不能为负数: {}", numericValue);
return false;
}
} catch (NumberFormatException e) {
log.warn("无法转换水质数据: field={}, value={}", field, value);
return false;
}
}
}
// 合格性验证
Object isQualified = data.get("isQualified");
if (isQualified != null && !(isQualified instanceof Boolean)) {
log.warn("合格性字段类型错误: {}", isQualified);
return false;
}
return true;
}
/**
* 转换为双精度浮点数
*/
private static double convertToDouble(Object value) throws NumberFormatException {
if (value instanceof Number) {
return ((Number) value).doubleValue();
} else if (value instanceof String) {
return Double.parseDouble((String) value);
} else {
throw new NumberFormatException("无法转换类型: " + value.getClass());
}
}
/**
* 生成数据质量评分
*/
public static double calculateDataQualityScore(Map<String, Object> data) {
double score = 100.0;
// 设备编号缺失扣分
if (!data.containsKey("deviceSn") || !isValidDeviceSn((String) data.get("deviceSn"))) {
score -= 20;
}
// 时间戳缺失扣分
if (!data.containsKey("timestamp")) {
score -= 10;
}
// 指标数据缺失扣分
if (!data.containsKey("metrics") || ((Map<?, ?>) data.get("metrics")).isEmpty()) {
score -= 30;
}
// 数值超出范围扣分
@SuppressWarnings("unchecked")
Map<String, Object> metrics = (Map<String, Object>) data.get("metrics");
if (metrics != null) {
int invalidCount = 0;
for (Map.Entry<String, Object> entry : metrics.entrySet()) {
MetricType metricType = MetricType.fromName(entry.getKey());
if (metricType != null && !isValidValue(metricType, entry.getValue())) {
invalidCount++;
}
}
score -= invalidCount * 5;
}
return Math.max(0, score);
}
}
@@ -44,6 +44,27 @@ minio:
secret-key: ${MINIO_SECRET_KEY:minioadmin}
bucket: water-management
# TDengine 配置
tda:
host: ${TDENGINE_HOST:127.0.0.1}
port: ${TDENGINE_PORT:6030}
username: ${TDENGINE_USER:root}
password: ${TDENGINE_PASS:taosdata}
database: ${TDENGINE_DB:water_iot}
# MQTT 配置
mqtt:
broker-url: ${MQTT_BROKER_URL:tcp://127.0.0.1:1883}
client-id: ${MQTT_CLIENT_ID:water-data-engine}
username: ${MQTT_USERNAME:water}
password: ${MQTT_PASSWORD:water123}
timeout: 30
keep-alive: 60
topic:
iot-telemetry: iot/telemetry/+
iot-command: iot/command/+
quality-data: quality/data/+
# 日志配置
logging:
level:
@@ -0,0 +1,84 @@
-- 抄表管理相关表结构
-- Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
-- 水表信息表
CREATE TABLE IF NOT EXISTS meter_info (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
account_no VARCHAR(50) NOT NULL COMMENT '用户编号',
meter_no VARCHAR(50) NOT NULL COMMENT '水表编号',
meter_type VARCHAR(20) COMMENT '水表类型: mechanical/digital/smart',
meter_caliber VARCHAR(20) COMMENT '水表口径: DN15/DN20/DN25/DN32/DN40/DN50/DN80/DN100/DN150/DN200',
location VARCHAR(200) COMMENT '水表位置',
install_date DATETIME COMMENT '安装日期',
initial_reading DECIMAL(12,3) COMMENT '初始读数',
current_reading DECIMAL(12,3) COMMENT '当前读数',
last_reading DECIMAL(12,3) COMMENT '上次抄表读数',
status VARCHAR(20) DEFAULT 'active' COMMENT '水表状态: active/inactive/maintaining/replaced',
brand VARCHAR(50) COMMENT '水表品牌',
model VARCHAR(50) COMMENT '水表型号',
protocol VARCHAR(20) COMMENT '通讯协议: NB-IoT/LoRaWAN/4G/RS485/M-BUS',
device_address VARCHAR(100) COMMENT '设备地址/IMEI',
last_online_time DATETIME COMMENT '最后在线时间',
battery_status VARCHAR(20) COMMENT '电池状态: normal/low/replace',
signal_strength INT COMMENT '信号强度',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_meter_no (meter_no),
KEY idx_account_no (account_no),
KEY idx_status (status)
) ENGINE=InnoDB COMMENT='水表信息表';
-- 抄表记录表
CREATE TABLE IF NOT EXISTS meter_read_record (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
account_no VARCHAR(50) NOT NULL COMMENT '用户编号',
meter_no VARCHAR(50) NOT NULL COMMENT '水表编号',
read_date DATETIME NOT NULL COMMENT '抄表日期',
read_value DECIMAL(12,3) NOT NULL COMMENT '本次读数',
last_read_value DECIMAL(12,3) COMMENT '上次读数',
read_difference DECIMAL(12,3) COMMENT '用水量(本次读数-上次读数)',
read_type VARCHAR(20) DEFAULT 'manual' COMMENT '抄表类型: manual/auto/remote',
read_method VARCHAR(20) COMMENT '抄表方式: field/phone/web/iot',
reader_name VARCHAR(50) COMMENT '抄表员姓名',
reader_id VARCHAR(50) COMMENT '抄表员ID',
remarks TEXT COMMENT '备注',
data_quality VARCHAR(20) DEFAULT 'normal' COMMENT '数据质量: normal/abnormal/verify',
photo_path VARCHAR(255) COMMENT '照片路径',
is_verified BOOLEAN DEFAULT FALSE COMMENT '是否已验证',
verified_by VARCHAR(50) COMMENT '验证人',
verified_at DATETIME COMMENT '验证时间',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
KEY idx_account_no (account_no),
KEY idx_meter_no (meter_no),
KEY idx_read_date (read_date),
KEY idx_read_type (read_type),
KEY idx_data_quality (data_quality),
KEY idx_is_verified (is_verified)
) ENGINE=InnoDB COMMENT='抄表记录表';
-- 抄表任务表
CREATE TABLE IF NOT EXISTS meter_read_task (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
task_name VARCHAR(100) NOT NULL COMMENT '任务名称',
task_type VARCHAR(20) NOT NULL COMMENT '任务类型: regular/remote/batch',
description TEXT COMMENT '任务描述',
execute_date DATE NOT NULL COMMENT '执行日期',
plan_start_time DATETIME COMMENT '计划开始时间',
plan_end_time DATETIME COMMENT '计划结束时间',
status VARCHAR(20) DEFAULT 'pending' COMMENT '任务状态: pending/progress/completed/failed',
assignee VARCHAR(50) COMMENT '分配抄表员',
priority VARCHAR(20) DEFAULT 'medium' COMMENT '任务优先级: high/medium/low',
area VARCHAR(100) COMMENT '抄表区域',
estimated_count INT DEFAULT 0 COMMENT '预计抄表数量',
actual_count INT DEFAULT 0 COMMENT '实际抄表数量',
completion_rate DECIMAL(5,2) COMMENT '完成率',
abnormal_count INT DEFAULT 0 COMMENT '异常数量',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
KEY idx_task_type (task_type),
KEY idx_status (status),
KEY idx_assignee (assignee),
KEY idx_area (area),
KEY idx_execute_date (execute_date)
) ENGINE=InnoDB COMMENT='抄表任务表';
@@ -0,0 +1,133 @@
-- 阶梯水价相关表结构
-- Issue #50: 抄表管理(人工+远传集成)+ 阶梯水价计算
-- 阶梯水价配置表
CREATE TABLE IF NOT EXISTS tariff_ladder_config (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
config_name VARCHAR(100) NOT NULL COMMENT '配置名称',
config_code VARCHAR(50) NOT NULL COMMENT '配置代码',
water_type VARCHAR(20) NOT NULL COMMENT '水类型: residential/commercial/industrial',
area_code VARCHAR(20) COMMENT '区域代码(空表示全区域)',
start_date DATE NOT NULL COMMENT '开始日期',
end_date DATE NOT NULL COMMENT '结束日期',
description TEXT COMMENT '描述',
is_active BOOLEAN DEFAULT TRUE COMMENT '是否激活',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_config_code (config_code),
KEY idx_water_type (water_type),
KEY idx_area_code (area_code),
KEY idx_is_active (is_active)
) ENGINE=InnoDB COMMENT='阶梯水价配置表';
-- 阶梯水价详情表
CREATE TABLE IF NOT EXISTS tariff_ladder_detail (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
config_id BIGINT NOT NULL COMMENT '配置ID',
step INT NOT NULL COMMENT '阶梯序号',
start_volume DECIMAL(10,3) NOT NULL COMMENT '起始水量',
end_volume DECIMAL(10,3) COMMENT '结束水量(null表示无上限)',
unit_price DECIMAL(10,3) NOT NULL COMMENT '单价',
include_start BOOLEAN DEFAULT TRUE COMMENT '是否包含起始量',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
KEY idx_config_id (config_id),
KEY idx_step (step)
) ENGINE=InnoDB COMMENT='阶梯水价详情表';
-- 客户账户表
CREATE TABLE IF NOT EXISTS customer_account (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
account_no VARCHAR(50) NOT NULL COMMENT '户号',
customer_name VARCHAR(100) NOT NULL COMMENT '客户姓名',
phone VARCHAR(20) COMMENT '联系电话',
address VARCHAR(500) COMMENT '地址',
meter_type VARCHAR(20) COMMENT '水表类型',
meter_caliber VARCHAR(20) COMMENT '水表口径',
water_usage_type VARCHAR(20) NOT NULL COMMENT '用水性质: residential/commercial/industrial',
area_code VARCHAR(20) COMMENT '区域代码',
basic_water_amount DECIMAL(10,3) COMMENT '基本水量',
is_active BOOLEAN DEFAULT TRUE COMMENT '是否激活',
open_date DATE NOT NULL COMMENT '开户日期',
last_read_date DATETIME COMMENT '上次抄表日期',
last_reading DECIMAL(12,3) COMMENT '上次抄表读数',
total_consumption DECIMAL(12,3) DEFAULT 0 COMMENT '累计用量',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_account_no (account_no),
KEY idx_water_usage_type (water_usage_type),
KEY idx_area_code (area_code),
KEY idx_is_active (is_active)
) ENGINE=InnoDB COMMENT='客户账户表';
-- 账单周期表
CREATE TABLE IF NOT EXISTS bill_cycle (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
cycle_name VARCHAR(100) NOT NULL COMMENT '周期名称',
cycle_code VARCHAR(50) NOT NULL COMMENT '周期代码',
cycle_type VARCHAR(20) NOT NULL COMMENT '周期类型: monthly/quarterly/yearly/custom',
cycle_length INT DEFAULT 1 COMMENT '周期长度(月)',
start_date DATE NOT NULL COMMENT '开始日期',
end_date DATE NOT NULL COMMENT '结束日期',
read_start_date DATE NOT NULL COMMENT '抄表开始日期',
read_end_date DATE NOT NULL COMMENT '抄表结束日期',
bill_date DATE NOT NULL COMMENT '账单生成日期',
due_date DATE NOT NULL COMMENT '缴费截止日期',
is_active BOOLEAN DEFAULT TRUE COMMENT '是否激活',
description TEXT COMMENT '描述',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_cycle_code (cycle_code)
) ENGINE=InnoDB COMMENT='账单周期表';
-- 账单主表
CREATE TABLE IF NOT EXISTS bill_main (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
bill_no VARCHAR(50) NOT NULL COMMENT '账单号',
account_no VARCHAR(50) NOT NULL COMMENT '用户编号',
customer_name VARCHAR(100) NOT NULL COMMENT '客户姓名',
cycle_id BIGINT NOT NULL COMMENT '周期ID',
billing_period_start DATE NOT NULL COMMENT '计费周期开始',
billing_period_end DATE NOT NULL COMMENT '计费周期结束',
meter_reading_start DECIMAL(12,3) NOT NULL COMMENT '期初读数',
meter_reading_end DECIMAL(12,3) NOT NULL COMMENT '期末读数',
water_consumption DECIMAL(12,3) NOT NULL COMMENT '用水量',
basic_water_fee DECIMAL(12,2) NOT NULL COMMENT '基本水费',
ladder_water_fee DECIMAL(12,2) NOT NULL COMMENT '阶梯水费',
surcharge_fee DECIMAL(12,2) DEFAULT 0 COMMENT '附加费',
total_amount DECIMAL(12,2) NOT NULL COMMENT '总金额',
status VARCHAR(20) DEFAULT 'generated' COMMENT '状态: generated/sent/paid/overdue',
sent_date DATE COMMENT '发送日期',
due_date DATE NOT NULL COMMENT '到期日期',
payment_method VARCHAR(20) COMMENT '支付方式: cash/bank/alipay/wechat',
payment_date DATE COMMENT '支付日期',
payment_amount DECIMAL(12,2) COMMENT '支付金额',
notes TEXT COMMENT '备注',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
KEY idx_account_no (account_no),
KEY idx_cycle_id (cycle_id),
KEY idx_status (status),
KEY idx_due_date (due_date)
) ENGINE=InnoDB COMMENT='账单主表';
-- 账单明细表
CREATE TABLE IF NOT EXISTS bill_detail (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
bill_id BIGINT NOT NULL COMMENT '账单ID',
detail_type VARCHAR(20) NOT NULL COMMENT '明细类型: basic/ladder/surcharge',
item_name VARCHAR(100) NOT NULL COMMENT '项目名称',
unit VARCHAR(20) COMMENT '单位',
quantity DECIMAL(10,3) COMMENT '数量',
unit_price DECIMAL(12,2) COMMENT '单价',
amount DECIMAL(12,2) NOT NULL COMMENT '金额',
start_reading DECIMAL(12,3) COMMENT '开始读数',
end_reading DECIMAL(12,3) COMMENT '结束读数',
water_volume DECIMAL(12,3) COMMENT '用水量',
step_number INT COMMENT '阶梯序号',
remarks TEXT COMMENT '备注',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
KEY idx_bill_id (bill_id),
KEY idx_detail_type (detail_type)
) ENGINE=InnoDB COMMENT='账单明细表';
@@ -1,5 +1,8 @@
package com.water.data_engine.service;
import com.water.data_engine.mapper.CollectRecordMapper;
import com.water.data_engine.mapper.CollectTaskMapper;
import com.water.data_engine.mapper.DataSourceMapper;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
@@ -0,0 +1,214 @@
package com.water.data_engine.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.water.data_engine.mapper.CollectRecordMapper;
import com.water.data_engine.mapper.CollectTaskMapper;
import com.water.data_engine.mapper.DataSourceMapper;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;
/**
* Kafka 消费者测试
*/
@ExtendWith(MockitoExtension.class)
class KafkaConsumerTest {
@Mock
private KafkaTemplate<String, String> kafkaTemplate;
@Mock
private JdbcTemplate jdbcTemplate;
@Mock
private DataSourceMapper dataSourceMapper;
@Mock
private CollectTaskMapper collectTaskMapper;
@Mock
private CollectRecordMapper collectRecordMapper;
@Mock
private SimpMessagingTemplate wsMessagingTemplate;
private DataCollectService collectService;
private ObjectMapper objectMapper;
@BeforeEach
void setUp() {
collectService = new DataCollectService(
kafkaTemplate, jdbcTemplate, dataSourceMapper,
collectTaskMapper, collectRecordMapper, wsMessagingTemplate
);
objectMapper = new ObjectMapper();
}
@Test
@DisplayName("Kafka消费-IoT原始数据")
void testConsumeIotRaw() {
// Given
String deviceSn = "FM001";
String message = buildIotTelemetryMessage(deviceSn);
// When
collectService.consumeIotRaw(message);
// Then
verify(jdbcTemplate, times(3)).update(
eq("INSERT INTO water_iot.iot_telemetry (ts, device_sn, metric_key, metric_value, quality) VALUES (NOW, ?, ?, ?, 1)"),
eq(deviceSn),
anyString(),
any()
);
}
@Test
@DisplayName("Kafka消费-水质数据")
void testConsumeQualityData() {
// Given
String testPoint = "水厂出口";
String message = buildQualityDataMessage(testPoint);
// When
collectService.consumeQualityData(message);
// Then
verify(jdbcTemplate).update(
eq("INSERT INTO water_quality_record (test_type, test_point, point_type, area, " +
"turbidity, ph, residual_chlorine, is_qualified, created_at) " +
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, NOW())"),
any(),
eq(testPoint),
any(),
any(),
any(),
any(),
any()
);
}
@Test
@DisplayName("数据验证-合格数据")
void testDataValidation_ValidData() {
Map<String, Object> validData = new HashMap<>();
validData.put("deviceSn", "FM001");
validData.put("timestamp", System.currentTimeMillis());
validData.put("metrics", List.of(
Map.of("key", "LL", "value", 12.5),
Map.of("key", "YL", "value", 0.35),
Map.of("key", "PH", "value", 7.2)
));
// 使用反射访问私有方法
boolean result = collectService.validateData("iot", validData);
assertTrue(result, "合格数据应该通过验证");
}
@Test
@DisplayName("数据验证-无效设备编号")
void testDataValidation_InvalidDeviceSn() {
Map<String, Object> invalidData = new HashMap<>();
invalidData.put("deviceSn", "INVALID_DEVICE"); // 超过20个字符
invalidData.put("timestamp", System.currentTimeMillis());
invalidData.put("metrics", List.of(
Map.of("key", "LL", "value", 12.5)
));
boolean result = collectService.validateData("iot", invalidData);
assertFalse(result, "无效设备编号应该无法通过验证");
}
@Test
@DisplayName("数据验证-数值超出范围")
void testDataValidation_InvalidValue() {
Map<String, Object> invalidData = new HashMap<>();
invalidData.put("deviceSn", "FM001");
invalidData.put("timestamp", System.currentTimeMillis());
invalidData.put("metrics", List.of(
Map.of("key", "LL", "value", 999999) // 流量超出合理范围
));
boolean result = collectService.validateData("iot", invalidData);
assertFalse(result, "超出范围的数值应该无法通过验证");
}
@Test
@DisplayName("Topic路由测试")
void testRouteTopic() {
assertEquals("iot.raw.generic", collectService.routeTopic("iot"));
assertEquals("iot.raw.generic", collectService.routeTopic("mqtt"));
assertEquals("data.quality", collectService.routeTopic("quality"));
assertEquals("data.manual", collectService.routeTopic("manual"));
assertEquals("data.api", collectService.routeTopic("api"));
assertEquals("data.raw", collectService.routeTopic("unknown"));
}
/**
* 构建IoT遥测数据消息
*/
private String buildIotTelemetryMessage(String deviceSn) {
Map<String, Object> data = new HashMap<>();
data.put("deviceSn", deviceSn);
data.put("timestamp", System.currentTimeMillis());
data.put("metrics", List.of(
Map.of("key", "LL", "value", 12.5),
Map.of("key", "YL", "value", 0.35),
Map.of("key", "PH", "value", 7.2)
));
Map<String, Object> envelope = new HashMap<>();
envelope.put("sourceType", "iot");
envelope.put("sourceId", deviceSn);
envelope.put("timestamp", System.currentTimeMillis());
envelope.put("data", data);
try {
return objectMapper.writeValueAsString(envelope);
} catch (Exception e) {
throw new RuntimeException("构建测试消息失败", e);
}
}
/**
* 构建水质数据消息
*/
private String buildQualityDataMessage(String testPoint) {
Map<String, Object> data = new HashMap<>();
data.put("testType", "常规检测");
data.put("testPoint", testPoint);
data.put("pointType", "出厂水");
data.put("area", "主城区");
data.put("turbidity", 0.5);
data.put("ph", 7.2);
data.put("residualChlorine", 0.3);
data.put("isQualified", true);
Map<String, Object> envelope = new HashMap<>();
envelope.put("sourceType", "quality");
envelope.put("sourceId", "WQ001");
envelope.put("timestamp", System.currentTimeMillis());
envelope.put("data", data);
try {
return objectMapper.writeValueAsString(envelope);
} catch (Exception e) {
throw new RuntimeException("构建测试消息失败", e);
}
}
}