[0924] 超短期

forecast-backend-server/src/main/java/com/ufo/project/data/processer/DataFileProcesserPowerImpl.java

@@ -12,6 +12,7 @@ import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeMap;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -34,7 +35,9 @@ import com.ufo.project.forecast.mediumterm.power.mapper.DataMediumTermPowerForec
 import com.ufo.project.forecast.shortterm.power.domain.DataShortTermPowerForecastRaw;
 import com.ufo.project.forecast.shortterm.power.mapper.DataShortTermPowerForecastRawMapper;
 import com.ufo.project.forecast.supershort.domain.DataSuperShortTermForecastRaw;
+import com.ufo.project.forecast.supershort.domain.DataUltraShortTermPowerForecasting;
 import com.ufo.project.forecast.supershort.mapper.DataSuperShortTermForecastRawMapper;
+import com.ufo.project.forecast.supershort.mapper.DataUltraShortTermPowerForecastingMapper;
 
 /**
  * 功率预测数据文件处理器(短期/中期/超短期)
@@ -100,6 +103,12 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
     private static final Long ANALYZE_HIS_ID = 0L;
     private static final Long ARCHIVE_RECORD_ID = 0L;
 
+    /** data_ultra_short_term_power_forecasting 的 powerXX 列: 分钟间隔步长与最大时效(t+15..t+240) */
+    private static final long FORECAST_STEP_MINUTES = 15L;
+    private static final long FORECAST_MAX_MINUTES = 240L;
+    /** data_ultra_short_term_power_forecasting 的 entity_type(场站/交易主体/省域), 场站取1 */
+    private static final Long ULTRA_ENTITY_TYPE = 1L;
+
     /** 短期/中期/超短期各自的元数据组(id同时用作data_source/data_sourcegrab_id/data_model_id/source_id/subject_id/machine_id) */
     static final Term SHORT_TERM = new Term(Kind.SHORT, 5001L, "EC", "ShortTermModel", "短期", 3L, 72L);
     static final Term MEDIUM_TERM = new Term(Kind.MEDIUM, 5002L, "WRF", "MediumTermModel", "中期", 10L, 240L);
@@ -114,6 +123,9 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
     @Autowired
     private DataSuperShortTermForecastRawMapper dataSuperShortTermForecastRawMapper;
 
+    @Autowired
+    private DataUltraShortTermPowerForecastingMapper dataUltraShortTermPowerForecastingMapper;
+
     @Override
     protected String parseAndSave(DataFileSyncFiles file)
     {
@@ -153,7 +165,7 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
             return "csv无有效数据行";
         }
 
-        int[] result;
+        int[] result = new int[]{0, 0};
         switch (term.kind)
         {
             case SHORT:
@@ -162,8 +174,11 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
             case MEDIUM:
                 result = saveMediumTerm(term, station, startTime, powerByForecastTime);
                 break;
-            default:
+            case ULTRA:
                 result = saveUltraShortTerm(term, station, startTime, powerByForecastTime);
+                saveUltraShortTermNew(term, station, startTime, powerByForecastTime);
+                break;
+            default:
                 break;
         }
         log.info("[DataFileProcesserPower] 处理完成, file={}, 场站={}, {}, start_time={}, 更新{}条, 新增{}条",
@@ -336,6 +351,65 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
                 dataSuperShortTermForecastRawMapper::batchInsertDataSuperShortTermForecastRaw);
     }
 
+    /**
+     * 合并保存一条 data_ultra_short_term_power_forecasting: 按(entity_id, start_time)查重,
+     * 已存在则按主键覆盖powerXX预测列(不动实发power), 否则新增(create_time取库端当前epoch秒)
+     */
+    private void saveUltraShortTermNew(Term term, ConfigStation station, long startTime, Map<Long, String> powerByForecastTime)
+    {
+        DataUltraShortTermPowerForecasting row = buildUltraForecasting(term, station, startTime, powerByForecastTime);
+        DataUltraShortTermPowerForecasting query = new DataUltraShortTermPowerForecasting();
+        query.setEntityId(row.getEntityId());
+        query.setStartTime(startTime);
+        List<DataUltraShortTermPowerForecasting> dbList =
+                dataUltraShortTermPowerForecastingMapper.selectDataUltraShortTermPowerForecastingList(query);
+        if (dbList != null && !dbList.isEmpty() && dbList.get(0).getId() != null)
+        {
+            row.setId(dbList.get(0).getId());
+            dataUltraShortTermPowerForecastingMapper.updateDataUltraShortTermPowerForecastingFromFile(row);
+        }
+        else
+        {
+            dataUltraShortTermPowerForecastingMapper.insertDataUltraShortTermPowerForecastingFromFile(row);
+        }
+    }
+
+    /**
+     * 组装超短期透视行: powerByForecastTime按key升序, 以与startTime的间隔分钟数取对应powerXX列
+     * (仅t+15..t+240的15分钟整倍数, 其余跳过并告警); 元数据与raw链路同源(term=5003/EC-WRF),
+     * entity_id为场站id字符串、entity_type场站取1、entity_name场站名;
+     * forecast_time取最后一个有效预测时间(列可空), 实发power不写
+     */
+    DataUltraShortTermPowerForecasting buildUltraForecasting(Term term, ConfigStation station, long startTime, Map<Long, String> powerByForecastTime)
+    {
+        DataUltraShortTermPowerForecasting row = new DataUltraShortTermPowerForecasting();
+        row.setDataSourcegrabId(term.id);
+        row.setDataSourcegrabPatternName(term.patternName);
+        row.setEntityId(String.valueOf(station.getId()));
+        row.setEntityType(ULTRA_ENTITY_TYPE);
+        row.setEntityName(station.getStationName());
+        row.setStartTime(startTime);
+        Long lastForecastTime = null;
+        for (Map.Entry<Long, String> entry : new TreeMap<>(powerByForecastTime).entrySet())
+        {
+            long offsetSeconds = entry.getKey() - startTime;
+            long minutes = offsetSeconds / 60L;
+            if (offsetSeconds > 0 && offsetSeconds % (FORECAST_STEP_MINUTES * 60L) == 0
+                    && minutes >= FORECAST_STEP_MINUTES && minutes <= FORECAST_MAX_MINUTES)
+            {
+                setEntityField(row, "power" + minutes, entry.getValue());
+                lastForecastTime = entry.getKey();
+            }
+            else
+            {
+                log.warn("[DataFileProcesserPower] 超短期预测时间与起报间隔不符合powerXX列, 跳过, start_time={}, forecast_time={}",
+                        startTime, entry.getKey());
+            }
+        }
+        row.setForecastTime(lastForecastTime);
+        return row;
+    }
+
     /**
      * 三条链路共用: 按(station_id, start_time)查已有记录, 以forecast_time匹配,
      * 已有置Id进更新组、其余进插入组, 各按BATCH_SIZE一批写库;

forecast-backend-server/src/main/java/com/ufo/project/forecast/supershort/mapper/DataUltraShortTermPowerForecastingMapper.java

@@ -81,4 +81,20 @@ public interface DataUltraShortTermPowerForecastingMapper
      */
     public List<DataUltraShortTermPowerForecasting> selectByStartTimeRange(@Param("entityId") String entityId,
             @Param("startTime") Long startTime, @Param("endTime") Long endTime);
+
+    /**
+     * 新增单条超短期功率预测数据(功率数据文件处理器, create_time取库端当前epoch秒)
+     *
+     * @param dataUltraShortTermPowerForecasting 超短期功率预测数据
+     * @return 结果
+     */
+    public int insertDataUltraShortTermPowerForecastingFromFile(DataUltraShortTermPowerForecasting dataUltraShortTermPowerForecasting);
+
+    /**
+     * 按主键更新powerXX预测功率列(功率数据文件处理器, 全量覆盖, 不动实发power)
+     *
+     * @param dataUltraShortTermPowerForecasting 超短期功率预测数据(需带主键Id)
+     * @return 结果
+     */
+    public int updateDataUltraShortTermPowerForecastingFromFile(DataUltraShortTermPowerForecasting dataUltraShortTermPowerForecasting);
 }

forecast-backend-server/src/main/resources/mybatis/supershort/DataUltraShortTermPowerForecastingMapper.xml

@@ -184,4 +184,48 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
             #{Id}
         </foreach>
     </delete>
+
+    <!-- 新增单条(功率数据文件处理器): create_time 取当前epoch秒(列为INTEGER) -->
+    <insert id="insertDataUltraShortTermPowerForecastingFromFile" parameterType="DataUltraShortTermPowerForecasting" useGeneratedKeys="true" keyProperty="Id">
+        insert into data_ultra_short_term_power_forecasting (
+            data_sourcegrab_id, data_sourcegrab_pattern_name, entity_id, entity_type, entity_name,
+            create_time, start_time, forecast_time,
+            power, power15, power30, power45, power60, power75, power90, power105, power120,
+            power135, power150, power165, power180, power195, power210, power225, power240
+        ) values (
+            #{dataSourcegrabId}, #{dataSourcegrabPatternName}, #{entityId}, #{entityType}, #{entityName,jdbcType=VARCHAR},
+            EXTRACT(EPOCH FROM now())::INTEGER, #{startTime}, #{forecastTime,jdbcType=INTEGER},
+            #{power,jdbcType=VARCHAR},
+            #{power15,jdbcType=VARCHAR}, #{power30,jdbcType=VARCHAR}, #{power45,jdbcType=VARCHAR}, #{power60,jdbcType=VARCHAR},
+            #{power75,jdbcType=VARCHAR}, #{power90,jdbcType=VARCHAR}, #{power105,jdbcType=VARCHAR}, #{power120,jdbcType=VARCHAR},
+            #{power135,jdbcType=VARCHAR}, #{power150,jdbcType=VARCHAR}, #{power165,jdbcType=VARCHAR}, #{power180,jdbcType=VARCHAR},
+            #{power195,jdbcType=VARCHAR}, #{power210,jdbcType=VARCHAR}, #{power225,jdbcType=VARCHAR}, #{power240,jdbcType=VARCHAR}
+        )
+    </insert>
+
+    <!-- 按主键更新(功率数据文件处理器): 全量覆盖powerXX预测列与元数据, 不动实发power列 -->
+    <update id="updateDataUltraShortTermPowerForecastingFromFile" parameterType="DataUltraShortTermPowerForecasting">
+        update data_ultra_short_term_power_forecasting
+        set data_sourcegrab_id = #{dataSourcegrabId},
+            data_sourcegrab_pattern_name = #{dataSourcegrabPatternName},
+            entity_name = #{entityName,jdbcType=VARCHAR},
+            forecast_time = #{forecastTime,jdbcType=INTEGER},
+            power15 = #{power15,jdbcType=VARCHAR},
+            power30 = #{power30,jdbcType=VARCHAR},
+            power45 = #{power45,jdbcType=VARCHAR},
+            power60 = #{power60,jdbcType=VARCHAR},
+            power75 = #{power75,jdbcType=VARCHAR},
+            power90 = #{power90,jdbcType=VARCHAR},
+            power105 = #{power105,jdbcType=VARCHAR},
+            power120 = #{power120,jdbcType=VARCHAR},
+            power135 = #{power135,jdbcType=VARCHAR},
+            power150 = #{power150,jdbcType=VARCHAR},
+            power165 = #{power165,jdbcType=VARCHAR},
+            power180 = #{power180,jdbcType=VARCHAR},
+            power195 = #{power195,jdbcType=VARCHAR},
+            power210 = #{power210,jdbcType=VARCHAR},
+            power225 = #{power225,jdbcType=VARCHAR},
+            power240 = #{power240,jdbcType=VARCHAR}
+        where _id = #{Id}
+    </update>
 </mapper>