Skip to content

Commit 1e7313c

Browse files
gmlexxgmlexx
committed
feat: add more data integration
1 parent ad91563 commit 1e7313c

7 files changed

Lines changed: 597 additions & 318 deletions

File tree

‎scheduler/config.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -340,6 +340,15 @@ func (c *Config) Validate() error {
340340
return fmt.Errorf("miner_power_super must be non-negative, got: %f", c.MinerPowerSuper)
341341
}
342342

343+
// Validate PV integration settings
344+
if c.PVPollInterval <= 0 {
345+
return fmt.Errorf("pv_poll_interval must be greater than 0, got: %s", c.PVPollInterval)
346+
}
347+
348+
if c.PVIntegrationPeriod <= 0 {
349+
return fmt.Errorf("pv_integration_period must be greater than 0, got: %s", c.PVIntegrationPeriod)
350+
}
351+
343352
return nil
344353
}
345354

‎scheduler/data.go‎

Lines changed: 357 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,357 @@
1+
package scheduler
2+
3+
import (
4+
"database/sql"
5+
"sync"
6+
"time"
7+
8+
"github.com/devskill-org/ems/meteo"
9+
"github.com/devskill-org/ems/sigenergy"
10+
)
11+
12+
type WeatherForecastCache struct {
13+
mu sync.RWMutex
14+
forecast *meteo.METJSONForecast
15+
fetchedAt time.Time
16+
cacheDuration time.Duration
17+
}
18+
19+
func (w *WeatherForecastCache) Get() (*meteo.METJSONForecast, bool) {
20+
w.mu.RLock()
21+
defer w.mu.RUnlock()
22+
23+
if w.forecast == nil {
24+
return nil, false
25+
}
26+
27+
if time.Since(w.fetchedAt) > w.cacheDuration {
28+
return nil, false
29+
}
30+
31+
return w.forecast, true
32+
}
33+
34+
func (w *WeatherForecastCache) Set(forecast *meteo.METJSONForecast) {
35+
w.mu.Lock()
36+
defer w.mu.Unlock()
37+
38+
w.forecast = forecast
39+
w.fetchedAt = time.Now()
40+
}
41+
42+
type DataSample struct {
43+
pvPower float64
44+
gridPower float64 // positive = import, negative = export
45+
batteryPower float64 // positive = charging, negative = discharging
46+
evdcPower float64
47+
batterySoc float64 // %
48+
ts time.Time
49+
}
50+
51+
type DataSamples struct {
52+
mu sync.Mutex
53+
samples []DataSample
54+
}
55+
56+
func (d *DataSamples) AddSample(pvPower, gridPower, batteryPower, evdcPower, batterySoc float64, ts time.Time) {
57+
d.mu.Lock()
58+
defer d.mu.Unlock()
59+
d.samples = append(d.samples, DataSample{
60+
pvPower: pvPower,
61+
gridPower: gridPower,
62+
batteryPower: batteryPower,
63+
evdcPower: evdcPower,
64+
batterySoc: batterySoc,
65+
ts: ts,
66+
})
67+
}
68+
69+
type IntegratedData struct {
70+
pvTotalPower float64
71+
gridExportPower float64
72+
gridImportPower float64
73+
batteryChargePower float64
74+
batteryDischargePower float64
75+
evdcChargePower float64
76+
loadPower float64
77+
batterySoc float64
78+
}
79+
80+
func (d *DataSamples) IntegrateSamples(pollInterval time.Duration) IntegratedData {
81+
d.mu.Lock()
82+
defer d.mu.Unlock()
83+
84+
var result IntegratedData
85+
86+
for _, sample := range d.samples {
87+
energyKWh := pollInterval.Seconds() / 3600.0 // Convert to hours
88+
89+
result.pvTotalPower += sample.pvPower * energyKWh
90+
91+
// Grid power: positive = import, negative = export
92+
if sample.gridPower > 0 {
93+
result.gridImportPower += sample.gridPower * energyKWh
94+
} else if sample.gridPower < 0 {
95+
result.gridExportPower += -sample.gridPower * energyKWh
96+
}
97+
98+
// Battery power: positive = charging, negative = discharging
99+
if sample.batteryPower > 0 {
100+
result.batteryChargePower += sample.batteryPower * energyKWh
101+
} else if sample.batteryPower < 0 {
102+
result.batteryDischargePower += -sample.batteryPower * energyKWh
103+
}
104+
105+
// EV DC charging power
106+
result.evdcChargePower += sample.evdcPower * energyKWh
107+
108+
// Keep the last battery SOC
109+
result.batterySoc = sample.batterySoc
110+
}
111+
112+
// Calculate load: Load = PV + Battery Discharge + Grid Import - Battery Charge - Grid Export - EV Charge
113+
result.loadPower = result.pvTotalPower + result.batteryDischargePower + result.gridImportPower -
114+
result.batteryChargePower - result.gridExportPower - result.evdcChargePower
115+
116+
d.samples = d.samples[:0]
117+
return result
118+
}
119+
120+
func (d *DataSamples) IsEmpty() bool {
121+
d.mu.Lock()
122+
defer d.mu.Unlock()
123+
return len(d.samples) == 0
124+
}
125+
126+
// GetLatestPower returns the most recent PV power sample, or 0 if no samples exist
127+
func (d *DataSamples) GetLatestPower() float64 {
128+
d.mu.Lock()
129+
defer d.mu.Unlock()
130+
if len(d.samples) == 0 {
131+
return 0
132+
}
133+
return d.samples[len(d.samples)-1].pvPower
134+
}
135+
136+
func (s *MinerScheduler) runDataPoll(samples *DataSamples) {
137+
if s.config.PlantModbusAddress == "" {
138+
return
139+
}
140+
client, err := sigenergy.NewTCPClient(s.config.PlantModbusAddress, sigenergy.PlantAddress)
141+
if err != nil {
142+
s.logger.Printf("Data integration: failed to create modbus client: %v", err)
143+
return
144+
}
145+
defer client.Close()
146+
info, err := client.ReadPlantRunningInfo()
147+
if err != nil {
148+
s.logger.Printf("Data integration: failed to read PlantRunningInfo: %v", err)
149+
return
150+
}
151+
samples.AddSample(
152+
info.PhotovoltaicPower,
153+
info.GridSensorActivePower,
154+
info.ESSPower,
155+
info.DCChargerOutputPower,
156+
info.ESSSOC,
157+
time.Now(),
158+
)
159+
}
160+
161+
func (s *MinerScheduler) runDataIntegration(samples *DataSamples, pollInterval time.Duration, dataDB *sql.DB, deviceID int, dryRun bool) {
162+
if samples.IsEmpty() {
163+
s.logger.Printf("Data integration: no samples collected in period")
164+
return
165+
}
166+
167+
data := samples.IntegrateSamples(pollInterval)
168+
timestamp := time.Now()
169+
170+
if dataDB == nil {
171+
return
172+
}
173+
174+
// Fetch weather data from meteo API
175+
cloudCoverage, err := s.fetchCloudCoverage()
176+
if err != nil {
177+
s.logger.Printf("Data integration: failed to fetch cloud coverage: %v", err)
178+
}
179+
180+
weatherSymbol, err := s.fetchWeatherSymbol()
181+
if err != nil {
182+
s.logger.Printf("Data integration: failed to fetch weather symbol: %v", err)
183+
}
184+
185+
// Calculate costs using current energy prices
186+
config := s.GetConfig()
187+
188+
// Get current spot price for cost calculations
189+
var gridImportCost, gridExportCost float64
190+
s.mu.RLock()
191+
marketData := s.pricesMarketData
192+
s.mu.RUnlock()
193+
194+
if marketData != nil {
195+
spotPrice, found := marketData.LookupAveragePriceInHourByTime(timestamp)
196+
if found && spotPrice > 0 {
197+
// Import cost: (spot price + operator fee + delivery fee) * energy in MWh
198+
importPricePerMWh := spotPrice + config.ImportPriceOperatorFee + config.ImportPriceDeliveryFee
199+
gridImportCost = (importPricePerMWh / 1000.0) * data.gridImportPower // Convert to EUR
200+
201+
// Export revenue (negative cost): (spot price - operator fee) * energy in MWh
202+
exportPricePerMWh := spotPrice - config.ExportPriceOperatorFee
203+
gridExportCost = (exportPricePerMWh / 1000.0) * data.gridExportPower // Convert to EUR
204+
}
205+
}
206+
207+
if dryRun {
208+
// DRY-RUN MODE: Log the action without saving to database
209+
s.logger.Printf("Data integration [DRY-RUN]: would save metrics for device_id=%d at %s", deviceID, timestamp.Format(time.RFC3339))
210+
s.logger.Printf(" PV: %.3f kWh, Grid Import: %.3f kWh (€%.3f), Grid Export: %.3f kWh (€%.3f)",
211+
data.pvTotalPower, data.gridImportPower, gridImportCost, data.gridExportPower, gridExportCost)
212+
s.logger.Printf(" Battery Charge: %.3f kWh, Battery Discharge: %.3f kWh, SOC: %.1f%%",
213+
data.batteryChargePower, data.batteryDischargePower, data.batterySoc)
214+
s.logger.Printf(" EV Charge: %.3f kWh, Load: %.3f kWh", data.evdcChargePower, data.loadPower)
215+
if cloudCoverage != nil {
216+
s.logger.Printf(" Cloud Coverage: %.1f%%", *cloudCoverage)
217+
}
218+
if weatherSymbol != nil {
219+
s.logger.Printf(" Weather: %s", *weatherSymbol)
220+
}
221+
} else {
222+
// Insert comprehensive energy flow data
223+
_, err = dataDB.Exec(
224+
`INSERT INTO metrics (
225+
timestamp, device_id, metric_name,
226+
pv_total_power, cloud_coverage, weather_symbol,
227+
grid_export_power, grid_import_power,
228+
battery_charge_power, battery_discharge_power, battery_soc,
229+
evdc_charge_power, load_power,
230+
grid_export_cost, grid_import_cost
231+
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)`,
232+
timestamp, deviceID, "energy_flow",
233+
data.pvTotalPower, cloudCoverage, weatherSymbol,
234+
data.gridExportPower, data.gridImportPower,
235+
data.batteryChargePower, data.batteryDischargePower, data.batterySoc,
236+
data.evdcChargePower, data.loadPower,
237+
gridExportCost, gridImportCost,
238+
)
239+
if err != nil {
240+
s.logger.Printf("Data integration: failed to insert metrics: %v", err)
241+
return
242+
}
243+
244+
s.logger.Printf("Data integration: saved metrics for device_id=%d at %s", deviceID, timestamp.Format(time.RFC3339))
245+
s.logger.Printf(" PV: %.3f kWh, Grid Import: %.3f kWh (€%.3f), Grid Export: %.3f kWh (€%.3f)",
246+
data.pvTotalPower, data.gridImportPower, gridImportCost, data.gridExportPower, gridExportCost)
247+
s.logger.Printf(" Battery Charge: %.3f kWh, Battery Discharge: %.3f kWh, SOC: %.1f%%",
248+
data.batteryChargePower, data.batteryDischargePower, data.batterySoc)
249+
s.logger.Printf(" EV Charge: %.3f kWh, Load: %.3f kWh", data.evdcChargePower, data.loadPower)
250+
}
251+
}
252+
253+
func (s *MinerScheduler) fetchCloudCoverage() (*float64, error) {
254+
// Check cache first
255+
if cachedForecast, ok := s.weatherCache.Get(); ok {
256+
current := cachedForecast.GetCurrentWeather()
257+
if current == nil {
258+
return nil, nil
259+
}
260+
return current.GetCloudCoverage(), nil
261+
}
262+
263+
// Cache miss, fetch from API
264+
s.logger.Printf("Data integration: fetching weather forecast from API")
265+
config := s.GetConfig()
266+
client := meteo.NewClient(config.UserAgent)
267+
268+
location := meteo.Location{
269+
Latitude: config.Latitude,
270+
Longitude: config.Longitude,
271+
}
272+
273+
params := meteo.QueryParams{Location: location}
274+
forecast, err := client.GetCompact(params)
275+
if err != nil {
276+
return nil, err
277+
}
278+
279+
// Store in cache
280+
s.weatherCache.Set(forecast)
281+
282+
current := forecast.GetCurrentWeather()
283+
if current == nil {
284+
return nil, nil
285+
}
286+
287+
return current.GetCloudCoverage(), nil
288+
}
289+
290+
func (s *MinerScheduler) fetchWeatherSymbol() (*string, error) {
291+
// Check cache first
292+
if cachedForecast, ok := s.weatherCache.Get(); ok {
293+
current := cachedForecast.GetCurrentWeather()
294+
if current == nil {
295+
return nil, nil
296+
}
297+
symbol := current.GetSymbolCode()
298+
if symbol == nil {
299+
return nil, nil
300+
}
301+
symbolStr := string(*symbol)
302+
return &symbolStr, nil
303+
}
304+
305+
// Cache miss, fetch from API
306+
config := s.GetConfig()
307+
client := meteo.NewClient(config.UserAgent)
308+
309+
location := meteo.Location{
310+
Latitude: config.Latitude,
311+
Longitude: config.Longitude,
312+
}
313+
314+
params := meteo.QueryParams{Location: location}
315+
forecast, err := client.GetCompact(params)
316+
if err != nil {
317+
return nil, err
318+
}
319+
320+
// Store in cache
321+
s.weatherCache.Set(forecast)
322+
323+
current := forecast.GetCurrentWeather()
324+
if current == nil {
325+
return nil, nil
326+
}
327+
328+
symbol := current.GetSymbolCode()
329+
if symbol == nil {
330+
return nil, nil
331+
}
332+
symbolStr := string(*symbol)
333+
return &symbolStr, nil
334+
}
335+
336+
// GetPlantRunningInfo returns the current plant running information
337+
// If PlantModbusAddress is not configured, returns nil
338+
func (s *MinerScheduler) GetPlantRunningInfo() *sigenergy.PlantRunningInfo {
339+
if s.config.PlantModbusAddress == "" {
340+
return nil
341+
}
342+
343+
client, err := sigenergy.NewTCPClient(s.config.PlantModbusAddress, sigenergy.PlantAddress)
344+
if err != nil {
345+
s.logger.Printf("Failed to create modbus client for plant info: %v", err)
346+
return nil
347+
}
348+
defer client.Close()
349+
350+
info, err := client.ReadPlantRunningInfo()
351+
if err != nil {
352+
s.logger.Printf("Failed to read plant running info: %v", err)
353+
return nil
354+
}
355+
356+
return info
357+
}

‎scheduler/miners.go‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,13 @@ import (
1313
func (s *MinerScheduler) discoverMiners(ctx context.Context) error {
1414
s.logger.Printf("Discovering miners on network: %s", s.config.Network)
1515

16-
newlyDiscoveredMiners := miners.Discover(ctx, s.config.Network)
16+
// Use injected discovery function for testing, otherwise use default
17+
var newlyDiscoveredMiners []*miners.AvalonQHost
18+
if s.minerDiscoveryFunc != nil {
19+
newlyDiscoveredMiners = s.minerDiscoveryFunc(ctx, s.config.Network)
20+
} else {
21+
newlyDiscoveredMiners = miners.Discover(ctx, s.config.Network)
22+
}
1723

1824
s.mu.Lock()
1925
defer s.mu.Unlock()

0 commit comments

Comments
 (0)