若依实现

<module>ruoyi-energy</module>

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <groupId>com.ruoyi</groupId>
        <artifactId>ruoyi</artifactId>
        <!-- 与根 pom 的 version 保持一致 -->
        <version>3.8.9</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>ruoyi-energy</artifactId>

    <dependencies>
        <dependency>
            <groupId>com.ruoyi</groupId>
            <artifactId>ruoyi-common</artifactId>
        </dependency>
        <dependency>
            <groupId>com.ruoyi</groupId>
            <artifactId>ruoyi-framework</artifactId>
        </dependency>
        <dependency>
            <groupId>com.ruoyi</groupId>
            <artifactId>ruoyi-system</artifactId>
        </dependency>
        <dependency>
            <groupId>com.taosdata.jdbc</groupId>
            <artifactId>taos-jdbcdriver</artifactId>
            <version>3.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.5</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>easyexcel</artifactId>
            <version>3.3.4</version>
        </dependency>
    </dependencies>
</project>

<dependency>
    <groupId>com.ruoyi</groupId>
    <artifactId>ruoyi-energy</artifactId>
</dependency>

com.ruoyi.energy
├── config
│   ├── TdengineConfig.java
│   ├── MqttConfig.java
│   └── ThreadPoolConfig.java
├── controller
│   ├── EnergyMeterController.java
│   ├── EnergyPointController.java
│   ├── EnergyStatController.java
│   └── CarbonAccountingController.java
├── service
│   ├── impl
│   ├── collect
│   ├── stat
│   ├── carbon
│   └── alarm
├── mapper
├── domain
├── enums
└── util

@RestController
@RequestMapping("/energy/meter")
public class EnergyMeterController extends BaseController {

    @Autowired
    private IEnergyMeterService meterService;

    @PreAuthorize("@ss.hasPermi('energy:meter:list')")
    @GetMapping("/list")
    public TableDataInfo list(EnergyMeter meter) {
        startPage();
        List<EnergyMeter> list = meterService.selectMeterList(meter);
        return getDataTable(list);
    }

    @PreAuthorize("@ss.hasPermi('energy:meter:query')")
    @GetMapping("/{meterId}")
    public AjaxResult getInfo(@PathVariable Long meterId) {
        return success(meterService.selectMeterById(meterId));
    }

    @PreAuthorize("@ss.hasPermi('energy:meter:add')")
    @Log(title = "计量点", businessType = BusinessType.INSERT)
    @PostMapping
    public AjaxResult add(@Validated @RequestBody EnergyMeter meter) {
        return toAjax(meterService.insertMeter(meter));
    }

    @PreAuthorize("@ss.hasPermi('energy:meter:edit')")
    @Log(title = "计量点", businessType = BusinessType.UPDATE)
    @PutMapping
    public AjaxResult edit(@Validated @RequestBody EnergyMeter meter) {
        return toAjax(meterService.updateMeter(meter));
    }

    @PreAuthorize("@ss.hasPermi('energy:meter:remove')")
    @Log(title = "计量点", businessType = BusinessType.DELETE)
    @DeleteMapping("/{meterIds}")
    public AjaxResult remove(@PathVariable Long[] meterIds) {
        return toAjax(meterService.deleteMeterByIds(meterIds));
    }
}

collector/
├── pom.xml
└── src/main
    ├── java/com/example/collector
    │   ├── CollectorApplication.java
    │   ├── mqtt/MqttIngestService.java
    │   ├── modbus/ModbusPollingService.java
    │   ├── store/TimeSeriesWriter.java
    │   ├── store/LatestValueCache.java
    │   ├── alarm/AlarmEvaluator.java
    │   └── model/TelemetryMessage.java
    └── resources/application.yml

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.5</version>
    </dependency>
    <dependency>
        <groupId>com.taosdata.jdbc</groupId>
        <artifactId>taos-jdbcdriver</artifactId>
        <version>3.3.0</version>
    </dependency>
    <dependency>
        <groupId>com.intelligt.modbus</groupId>
        <artifactId>jlibmodbus</artifactId>
        <version>1.2.9.7</version>
    </dependency>
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
</dependencies>

@Service
public class MqttIngestService implements MqttCallbackExtended {

    private static final Logger log = LoggerFactory.getLogger(MqttIngestService.class);

    @Value("${mqtt.broker}")
    private String broker;
    @Value("${mqtt.client-id}")
    private String clientId;
    @Value("${mqtt.username}")
    private String username;
    @Value("${mqtt.password}")
    private String password;
    @Value("${mqtt.qos:1}")
    private int qos;

    private final ObjectMapper objectMapper = new ObjectMapper();
    private final TimeSeriesWriter writer;
    private final LatestValueCache cache;
    private final AlarmEvaluator alarmEvaluator;

    private MqttClient client;

    public MqttIngestService(TimeSeriesWriter writer,
                             LatestValueCache cache,
                             AlarmEvaluator alarmEvaluator) {
        this.writer = writer;
        this.cache = cache;
        this.alarmEvaluator = alarmEvaluator;
    }

    @PostConstruct
    public void start() throws MqttException {
        client = new MqttClient(broker, clientId, new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setAutomaticReconnect(true);
        options.setCleanSession(false);
        options.setKeepAliveInterval(30);
        client.setCallback(this);
        client.connect(options);
        client.subscribe("energy/+/+/telemetry", qos);
        client.subscribe("energy/+/+/status", qos);
        log.info("MQTT connected: {}", broker);
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        try {
            client.subscribe("energy/+/+/telemetry", qos);
            client.subscribe("energy/+/+/status", qos);
            log.info("MQTT reconnected, resubscribed: {}", serverURI);
        } catch (MqttException e) {
            log.error("resubscribe failed", e);
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        try {
            String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
            if (topic.endsWith("/status")) {
                handleStatus(topic, payload);
            } else {
                handleTelemetry(topic, payload);
            }
        } catch (Exception e) {
            log.error("handle message failed, topic={}", topic, e);
        }
    }

    private void handleTelemetry(String topic, String payload) throws Exception {
        JsonNode root = objectMapper.readTree(payload);
        String gatewayId = root.path("gatewayId").asText();
        String meterCode = root.path("meterCode").asText();
        String energyType = root.path("energyType").asText("");
        long ts = root.path("ts").asLong(System.currentTimeMillis());
        int quality = root.path("quality").asInt(0);
        JsonNode points = root.path("points");

        Iterator<Map.Entry<String, JsonNode>> it = points.fields();
        while (it.hasNext()) {
            Map.Entry<String, JsonNode> entry = it.next();
            String pointCode = meterCode + "." + entry.getKey();
            double value = entry.getValue().asDouble();
            writer.append(pointCode, meterCode, energyType, ts, value, quality);
            cache.update(pointCode, value, ts, quality);
            alarmEvaluator.evaluate(pointCode, value, ts);
        }
    }

    private void handleStatus(String topic, String payload) {
        // 解析网关心跳，更新在线状态和最后心跳时间
        log.debug("gateway status: {} {}", topic, payload);
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.error("MQTT connection lost", cause);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 采集服务只订阅，不发布，忽略
    }
}

@Configuration
public class TdengineConfig {

    @Bean(name = "tdengineDataSource")
    @ConfigurationProperties(prefix = "tdengine.datasource")
    public DataSource tdengineDataSource() {
        return DataSourceBuilder.create().build();
    }
}

@Component
public class TimeSeriesWriter {

    private static final Logger log = LoggerFactory.getLogger(TimeSeriesWriter.class);

    private final BlockingQueue<PointSample> queue = new LinkedBlockingQueue<>(100000);
    private final DataSource dataSource;
    private final int batchSize;
    private final long flushIntervalMs;

    public TimeSeriesWriter(@Qualifier("tdengineDataSource") DataSource dataSource,
                            @Value("${tdengine.batch-size:500}") int batchSize,
                            @Value("${tdengine.flush-interval-ms:1000}") long flushIntervalMs) {
        this.dataSource = dataSource;
        this.batchSize = batchSize;
        this.flushIntervalMs = flushIntervalMs;
    }

    public void append(String pointCode, String meterCode, String energyType,
                       long ts, double value, int quality) {
        boolean offered = queue.offer(
                new PointSample(pointCode, meterCode, energyType, ts, value, quality));
        if (!offered) {
            log.warn("collector queue full, dropping sample: {}", pointCode);
        }
    }

    @Scheduled(fixedDelayString = "${tdengine.flush-interval-ms:1000}")
    public void flush() {
        List<PointSample> batch = new ArrayList<>(batchSize);
        queue.drainTo(batch, batchSize);
        if (batch.isEmpty()) {
            return;
        }
        try (Connection conn = dataSource.getConnection()) {
            conn.setAutoCommit(false);
            try (Statement stmt = conn.createStatement()) {
                for (PointSample s : batch) {
                    stmt.addBatch(buildInsertSql(s));
                }
                stmt.executeBatch();
                conn.commit();
            }
        } catch (Exception e) {
            log.error("flush to TDengine failed, size={}", batch.size(), e);
            // 失败时放回队列，等待下一次重试（生产环境要做幂等和限流）
            queue.addAll(batch);
        }
    }

    private String buildInsertSql(PointSample s) {
        String table = "p_" + Math.abs(s.getPointCode().hashCode());
        String sql = "INSERT INTO energy." + table
                + " USING energy.point_data TAGS ('" + esc(s.getPointCode()) + "','"
                + esc(s.getMeterCode()) + "','" + esc(s.getEnergyType()) + "',0)"
                + " VALUES (" + s.getTs() + "," + s.getValue() + "," + s.getQuality() + ")";
        return sql;
    }

    private String esc(String value) {
        return value == null ? "" : value.replace("'", "''");
    }
}

@Component
public class LatestValueCache {

    private final StringRedisTemplate redisTemplate;

    public LatestValueCache(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void update(String pointCode, double value, long ts, int quality) {
        String key = "energy:latest:" + pointCode;
        Map<String, String> map = new HashMap<>();
        map.put("value", String.valueOf(value));
        map.put("ts", String.valueOf(ts));
        map.put("quality", String.valueOf(quality));
        redisTemplate.opsForHash().putAll(key, map);
        redisTemplate.expire(key, 10, TimeUnit.MINUTES);
    }
}

@Service
public class ModbusPollingService {

    private static final Logger log = LoggerFactory.getLogger(ModbusPollingService.class);

    @Scheduled(fixedDelay = 60000)
    public void poll() {
        // 实际项目从数据库加载启用的网关和点位，这里用示例参数
        ModbusMaster master = null;
        try {
            master = ModbusMasterFactory.createModbusMasterTCP("192.168.10.11", 502);
            master.setResponseTimeout(3000);
            master.connect();

            // 从站地址 3，起始寄存器 0，读 10 个寄存器
            ReadHoldingRegistersRequest request =
                    new ReadHoldingRegistersRequest(3, 0, 10);
            ReadHoldingRegistersResponse response =
                    (ReadHoldingRegistersResponse) master.processRequest(request);

            int[] regs = response.getRegisters();
            double voltageA = regs[0] * 0.1;
            long energyRaw = ((long) regs[8] << 16) | (regs[9] & 0xFFFFL);
            double energy = energyRaw * 0.01;

            log.info("UA={}V, EP={}kWh", voltageA, energy);
        } catch (Exception e) {
            log.error("modbus poll failed", e);
        } finally {
            if (master != null) {
                try {
                    master.disconnect();
                } catch (Exception ignored) {
                }
            }
        }
    }
}

public final class ModbusValueUtil {

    private ModbusValueUtil() {
    }

    public static long toUInt32(int high, int low) {
        return ((long) high << 16) | (low & 0xFFFFL);
    }

    public static int toInt16(int raw) {
        return (short) raw;
    }

    public static float toFloatABCD(int r0, int r1) {
        int bits = (r0 << 16) | (r1 & 0xFFFF);
        return Float.intBitsToFloat(bits);
    }

    public static float toFloatCDAB(int r0, int r1) {
        return toFloatABCD(r1, r0);
    }

    public static BigDecimal scale(long raw, BigDecimal scale, BigDecimal offset) {
        return BigDecimal.valueOf(raw)
                .multiply(scale)
                .add(offset);
    }
}

@Component("energyStatTask")
public class EnergyStatTask {

    @Autowired
    private IEnergyStatService statService;

    /**
     * 聚合指定小时，参数格式 yyyy-MM-dd HH
     * 例如：energyStatTask.hour('2026-10-01 10')
     */
    public void hour(String hour) {
        statService.aggregateHour(hour);
    }

    public void day(String day) {
        statService.aggregateDay(day);
    }

    public void month(String month) {
        statService.aggregateMonth(month);
    }
}

@Service
public class EnergyStatServiceImpl implements IEnergyStatService {

    @Autowired
    private EnergyPointMapper pointMapper;
    @Autowired
    private TimeSeriesQueryDao timeSeriesQueryDao;
    @Autowired
    private EnergyStatMapper statMapper;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void aggregateHour(String hour) {
        LocalDateTime start = LocalDateTime.parse(hour + ":00:00",
                DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        LocalDateTime end = start.plusHours(1);

        // selectEnabledPoints 联表 energy_meter，带出 mediaId、deptId
        List<EnergyPoint> points = pointMapper.selectEnabledPoints();
        for (EnergyPoint point : points) {
            TimeSeriesSummary summary = timeSeriesQueryDao.summarize(
                    point.getPointCode(), start, end);

            BigDecimal value;
            if ("CUMULATIVE".equals(point.getPointType())) {
                value = summary.getLastValue().subtract(summary.getFirstValue());
            } else {
                value = summary.getAvgValue();
            }

            EnergyStat stat = new EnergyStat();
            stat.setStatType("HOUR");
            stat.setStatTime(Timestamp.valueOf(start));
            stat.setPointId(point.getPointId());
            stat.setMeterId(point.getMeterId());
            stat.setMediaId(point.getMediaId());
            stat.setDeptId(point.getDeptId());
            stat.setValue(value);
            stat.setUnit(point.getUnit());
            stat.setDataCount(summary.getCount());
            stat.setExpectedCount(60 * 60 / Math.max(point.getPeriodSec(), 1));
            stat.setQuality(summary.getCount() >= stat.getExpectedCount() * 0.98 ? 0 : 1);
            statMapper.upsert(stat);
        }
    }
}

INSERT INTO energy_stat
  (stat_type, stat_time, point_id, meter_id, media_id, dept_id,
   value, unit, data_count, expected_count, quality, create_time)
VALUES
  (#{statType}, #{statTime}, #{pointId}, #{meterId}, #{mediaId}, #{deptId},
   #{value}, #{unit}, #{dataCount}, #{expectedCount}, #{quality}, NOW())
ON DUPLICATE KEY UPDATE
  value = VALUES(value),
  data_count = VALUES(data_count),
  expected_count = VALUES(expected_count),
  quality = VALUES(quality);

public interface EmissionCalculator {
    String sourceType();
    BigDecimal calculate(EmissionContext context);
}

@Component
public class ElectricityCalculator implements EmissionCalculator {

    @Override
    public String sourceType() {
        return "ELECTRICITY";
    }

    @Override
    public BigDecimal calculate(EmissionContext context) {
        return context.getActivity()
                .multiply(context.getFactorValue())
                .setScale(6, RoundingMode.HALF_UP);
    }
}

@Component
public class FuelCalculator implements EmissionCalculator {

    @Override
    public String sourceType() {
        return "FUEL";
    }

    @Override
    public BigDecimal calculate(EmissionContext context) {
        // 活动数据 = 消耗量 × 低位热值 × 含碳量 × 氧化率 × 44/12
        // 实际项目把这些参数做成因子配置，不要写死
        return context.getActivity()
                .multiply(context.getNcv())
                .multiply(context.getCarbonContent())
                .multiply(context.getOxidationRate())
                .multiply(new BigDecimal("3.666666667"))
                .divide(new BigDecimal("1000"), 6, RoundingMode.HALF_UP);
    }
}

@Service
public class CarbonAccountingServiceImpl implements ICarbonAccountingService {

    @Autowired
    private CarbonAccountingMapper accountingMapper;
    @Autowired
    private CarbonEmissionMapper emissionMapper;
    @Autowired
    private CarbonFactorMapper factorMapper;
    @Autowired
    private EnergyStatMapper statMapper;

    private final Map<String, EmissionCalculator> calculatorMap;

    public CarbonAccountingServiceImpl(List<EmissionCalculator> calculators) {
        this.calculatorMap = calculators.stream()
                .collect(Collectors.toMap(EmissionCalculator::sourceType, Function.identity()));
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void calculate(Long accountingId) {
        CarbonAccounting accounting = accountingMapper.selectById(accountingId);
        if (accounting == null) {
            throw new ServiceException("核算任务不存在");
        }
        if ("LOCKED".equals(accounting.getStatus())) {
            throw new ServiceException("已锁定的核算任务不能重算");
        }

        List<CarbonEmission> items = emissionMapper.selectByAccountingId(accountingId);
        BigDecimal scope1 = BigDecimal.ZERO;
        BigDecimal scope2 = BigDecimal.ZERO;
        BigDecimal scope3 = BigDecimal.ZERO;

        for (CarbonEmission item : items) {
            CarbonFactor factor = factorMapper.selectEffective(
                    item.getSourceType(),
                    "NATIONAL",
                    accounting.getPeriodEnd());
            if (factor == null) {
                throw new ServiceException("未找到有效排放因子：" + item.getSourceType());
            }

            EmissionContext context = new EmissionContext();
            context.setActivity(item.getActivityValue());
            context.setFactorValue(factor.getFactorValue());
            context.setSourceType(item.getSourceType());

            EmissionCalculator calculator = calculatorMap.get(item.getSourceType());
            if (calculator == null) {
                throw new ServiceException("不支持的排放源类型：" + item.getSourceType());
            }
            BigDecimal emission = calculator.calculate(context);

            item.setFactorId(factor.getFactorId());
            item.setFactorValue(factor.getFactorValue());
            item.setFactorUnit(factor.getUnit());
            item.setFactorVersion(factor.getVersion());
            item.setEmission(emission);
            emissionMapper.updateById(item);

            if ("SCOPE_1".equals(item.getScope())) {
                scope1 = scope1.add(emission);
            } else if ("SCOPE_2".equals(item.getScope())) {
                scope2 = scope2.add(emission);
            } else {
                scope3 = scope3.add(emission);
            }
        }

        accounting.setScope1(scope1);
        accounting.setScope2(scope2);
        accounting.setScope3(scope3);
        accounting.setTotalEmission(scope1.add(scope2).add(scope3));
        accounting.setStatus("REVIEWED");
        accounting.setCalcTime(new Date());
        accountingMapper.updateById(accounting);
    }
}

CREATE TABLE energy_alarm_rule (
  rule_id        BIGINT NOT NULL AUTO_INCREMENT,
  rule_name      VARCHAR(128) NOT NULL,
  point_id       BIGINT DEFAULT NULL,
  rule_type      VARCHAR(16) NOT NULL COMMENT 'THRESHOLD/RATE/OFFLINE/BUSINESS',
  operator       VARCHAR(8) DEFAULT '>' COMMENT '> >= < <= =',
  threshold      DECIMAL(18,6) DEFAULT NULL,
  duration_sec   INT DEFAULT 0 COMMENT '持续多久才报警',
  alarm_level    VARCHAR(8) DEFAULT 'WARN' COMMENT 'INFO/WARN/CRITICAL',
  notify_channel VARCHAR(64) DEFAULT NULL COMMENT 'SMS/EMAIL/WECHAT',
  status         CHAR(1) DEFAULT '0',
  create_time    DATETIME DEFAULT NULL,
  PRIMARY KEY (rule_id),
  KEY idx_point (point_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='报警规则';

public void raise(String pointCode, BigDecimal value, String level) {
    String key = "energy:alarm:active:" + pointCode;
    Boolean first = redisTemplate.opsForValue()
            .setIfAbsent(key, value.toPlainString(), Duration.ofMinutes(30));
    if (Boolean.TRUE.equals(first)) {
        // 第一次触发，写报警表、发通知
        alarmMapper.insert(...);
    } else {
        // 已有活动报警，只更新最后触发时间，不重复通知
        alarmMapper.updateLastTime(pointCode, new Date());
    }
}

-- 一级菜单：能源管理
INSERT INTO sys_menu
(menu_name, parent_id, order_num, path, component, is_frame, is_cache,
 menu_type, visible, status, perms, icon, create_by, create_time, remark)
VALUES
('能源管理', 0, 5, 'energy', NULL, 1, 0, 'M', '0', '0', '', 'energy', 'admin', SYSDATE(), '能源管理目录');
SET @energyId = LAST_INSERT_ID();

-- 二级菜单：计量点
INSERT INTO sys_menu
(menu_name, parent_id, order_num, path, component, is_frame, is_cache,
 menu_type, visible, status, perms, icon, create_by, create_time, remark)
VALUES
('计量点', @energyId, 1, 'meter', 'energy/meter/index', 1, 0, 'C', '0', '0', 'energy:meter:list', 'meter', 'admin', SYSDATE(), '计量点菜单');
SET @meterId = LAST_INSERT_ID();

-- 按钮权限
INSERT INTO sys_menu
(menu_name, parent_id, order_num, path, component, is_frame, is_cache,
 menu_type, visible, status, perms, icon, create_by, create_time, remark)
VALUES
('计量点查询', @meterId, 1, '', NULL, 1, 0, 'F', '0', '0', 'energy:meter:query', '#', 'admin', SYSDATE(), ''),
('计量点新增', @meterId, 2, '', NULL, 1, 0, 'F', '0', '0', 'energy:meter:add', '#', 'admin', SYSDATE(), ''),
('计量点修改', @meterId, 3, '', NULL, 1, 0, 'F', '0', '0', 'energy:meter:edit', '#', 'admin', SYSDATE(), ''),
('计量点删除', @meterId, 4, '', NULL, 1, 0, 'F', '0', '0', 'energy:meter:remove', '#', 'admin', SYSDATE(), ''),
('计量点导出', @meterId, 5, '', NULL, 1, 0, 'F', '0', '0', 'energy:meter:export', '#', 'admin', SYSDATE(), '');

{
  "gatewayId": "GW-TEST-01",
  "meterCode": "SH01-C01-ELE-001",
  "energyType": "electricity",
  "ts": 1760000000000,
  "quality": 0,
  "points": {
    "UA": 220.1,
    "P": 123.4,
    "EP": 45678.9
  }
}

energy/SH01/GW-TEST-01/telemetry