Strømpris fra både Tibber, Nordpool og Entso-e

[stigvi]

Jeg har etterhvert en god del automasjoner som er avhengig av strømpris og da var det alltid irriterende at pris ikke var tilgjengelig. Ofte var den tilgjengelig på Entso-e integrasjonen, men ikke på Nordpool integrasjonen. Eller motsatt.
Å bytte mellom de to er ikke gjort i en håndvending fordi de leverer data på forskjellig format. Så jeg har laget et pyscript som leser strømpriser fra både Nordpool og Entso-e og tilbyr data på et nytt format som da blir felles for disse to.

Noen vil kalle det en bakdel, men for meg er det en fordel at scriptet samler alle prisene i en enkel liste istedenfor å splitte den i to.

Det er ikke meningen at sciptet skal være gjenbrukbart hos alle og enhver, men mer som tips og inspirasjon til å lage et eget script.

PS1. Scriptet har to funksjoner i toppen for å akkumulere dagens strømkostnad. 
PS2: Nordpool og Entso-e forutsettes satt opp til å levere spotpris inkl mva og i kroner. Skriptet legger til nettleie og trekker fra strømstøtte.
PS3: Sats for nettleie hentes fra to input_number. Tidspunkt en veksler mellom natt og dag er hardkodet. Det samme er strømstøtte.

PS4: Nettleie natt blir brukt for lørdag, søndag og andre røde dager.

 

from datetime import datetime,timedelta
import holidays

YEAR = datetime.today().year
NOR_HOLIDAYS = holidays.NO(years=[YEAR, YEAR+1])

state.persist('pyscript.total_pris_for_strom', default_value=0, default_attributes={"unit_of_measurement":"NOK"})
state.persist('pyscript.snittpris_for_strom', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.spart_paa_strom', default_value=0, default_attributes={"unit_of_measurement":"NOK"})
state.persist('pyscript.strompris', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})

@time_trigger("cron(59 * * * *)")
def akkumulere_stromkostnad():
    p = round(float(pyscript.strompris) * float(sensor.estimated_hourly_consumption) + float(pyscript.total_pris_for_strom), 2)
    pyscript.total_pris_for_strom = p

    if float(sensor.consumption_thisday) > 0:
        pyscript.snittpris_for_strom = round(float(pyscript.total_pris_for_strom) / float(sensor.consumption_thisday), 4)
    
    pyscript.spart_paa_strom = round(float(sensor.consumption_thisday) * (float(pyscript.strompris.averageprice) - float(pyscript.snittpris_for_strom)), 2)

@time_trigger("cron(0 0 * * *)")
def nullstille_stromkostnad():
    pyscript.total_pris_for_strom = 0

@time_trigger("cron(0 * * * *)")
@time_trigger("once(now)")
@state_trigger("sensor.nordpool", "sensor.nordpool.raw_tomorrow")
@state_trigger("sensor.current_electricity_market_price", "sensor.average_electricity_price_today.prices_tomorrow")
def strompris():
    #state.set("pyscript.strompris", new_attributes={})
    
    grid_night = float(input_number.nettleie_natt) / 100.0
    grid_day = float(input_number.nettleie_dag) / 100.0
    additional_costs = float(input_number.paslag_strom) / 100.0
    
    if (sensor.nordpool.raw_tomorrow is not None
        and len(sensor.nordpool.raw_tomorrow) > 0
        and sensor.nordpool.raw_tomorrow[0]['value'] is not None):
        pyscript.strompris.updatetime = datetime.now().isoformat()
        pyscript.strompris.source = "Nordpool"
        l = []

        for sourceprice in sensor.nordpool.raw_today + sensor.nordpool.raw_tomorrow:
            pr = {}
            d = sourceprice["start"]
            pr["start"] = d.isoformat()
            pr["end"] = (d + timedelta(hours=1)).isoformat()
            pr["spotprice"] = float(sourceprice["value"])
            pr["gridprice"] = grid_day if is_peak(d) else grid_night
            pr["payback"] = max((float(pr["spotprice"]) - 0.875) * 0.9, 0.0)
            pr["totalprice"] = float(pr["spotprice"]) + float(pr["gridprice"]) + float(additional_costs) - float(pr["payback"])
            l.append(pr)

        pyscript.strompris.prices = l

    elif (sensor.average_electricity_price_today.prices_tomorrow is not None
        and len(sensor.average_electricity_price_today.prices_tomorrow) > 0):
        pyscript.strompris.updatetime = datetime.now().isoformat()
        pyscript.strompris.source = "Entso-e"
        l = []
        
        for sourceprice in sensor.average_electricity_price_today.prices_today + sensor.average_electricity_price_today.prices_tomorrow:
            pr = {}
            d = datetime.strptime(sourceprice["time"], "%Y-%m-%d %H:%M:%S%z")
            pr["start"] = d.isoformat()
            pr["end"] = (d + timedelta(hours=1)).isoformat()
            pr["spotprice"] = float(sourceprice["price"])
            pr["gridprice"] = grid_day if is_peak(d) else grid_night
            pr["payback"] = max((float(pr["spotprice"]) - 0.875) * 0.9, 0.0)
            pr["totalprice"] = float(pr["spotprice"]) + float(pr["gridprice"]) + float(additional_costs) - float(pr["payback"])
            l.append(pr)
        
        pyscript.strompris.prices = l
    
    price_sum = 0.0
    peak_sum = 0.0
    offpeak1_sum = 0.0
    offpeak2_sum = 0.0
    high = -1000
    low = 1000

    for sourceprice in pyscript.strompris.prices:
        if datetime.now() >= datetime.fromisoformat(sourceprice["start"]).replace(tzinfo=None) and datetime.now() < datetime.fromisoformat(sourceprice["end"]).replace(tzinfo=None):
            pyscript.strompris = float(sourceprice["totalprice"])
            pyscript.strompris.current_gridprice = float(sourceprice["gridprice"])
            pyscript.strompris.current_payback = float(sourceprice["payback"])
            pyscript.strompris.current_spotprice = float(sourceprice["spotprice"])
        
        if datetime.now().day == datetime.fromisoformat(sourceprice["start"]).day:
            price_sum += sourceprice["totalprice"]

            if datetime.fromisoformat(sourceprice["start"]).hour < 8:
                offpeak1_sum += sourceprice["totalprice"]
            elif datetime.fromisoformat(sourceprice["start"]).hour < 20:
                peak_sum += sourceprice["totalprice"]
            else:
                offpeak2_sum += sourceprice["totalprice"]
            
            if high < sourceprice["totalprice"]:
                high = sourceprice["totalprice"]
            
            if low > sourceprice["totalprice"]:
                low = sourceprice["totalprice"]
    
    pyscript.strompris.averageprice = float(price_sum) / 24.0
    pyscript.strompris.peak = float(peak_sum) / 12.0
    pyscript.strompris.offpeak1 = float(offpeak1_sum) / 8.0
    pyscript.strompris.offpeak2 = float(offpeak2_sum) / 4.0
    pyscript.strompris.high = float(high)
    pyscript.strompris.low = float(low)

def is_peak(t : datetime):
    if t.isoweekday() >= 6 or t.hour <= 5 or t.hour >= 22 or t.date() in NOR_HOLIDAYS:
        return False
    else:
        return True

 

[stigvi]

Jeg endrer stadig på dette og har tatt med mulighet for å hente strømpris fra Tibber. Skriptet velger selv hvor den skal hente prisen i fra, men hvis dette gjøres feil, kan jeg slå av kildene med brytere i UI.



 

from datetime import datetime,timedelta
import holidays

YEAR = datetime.today().year
NOR_HOLIDAYS = holidays.NO(years=[YEAR, YEAR+1])

state.persist('pyscript.total_pris_for_strom', default_value=0, default_attributes={"unit_of_measurement":"NOK"})
state.persist('pyscript.snittpris_for_strom', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.spart_paa_strom', default_value=0, default_attributes={"unit_of_measurement":"NOK"})
state.persist('pyscript.strompris', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.totalstrompris', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.nettleie', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.stromstotte', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.spotpris', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.gjennomsnittlig_strompris', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.peak_strompris', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.offpeak_1_strompris', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.offpeak_2_strompris', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.hoyeste_strompris', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})
state.persist('pyscript.laveste_strompris', default_value=0, default_attributes={"unit_of_measurement":"NOK/kWh"})

@time_trigger("cron(59 * * * *)")
def akkumulere_stromkostnad():
    p = round(float(pyscript.totalstrompris) * float(sensor.estimated_hourly_consumption) + float(pyscript.total_pris_for_strom), 2)
    pyscript.total_pris_for_strom = p

    if float(sensor.consumption_thisday) > 0:
        pyscript.snittpris_for_strom = round(float(pyscript.total_pris_for_strom) / float(sensor.consumption_thisday), 4)
    
    pyscript.spart_paa_strom = round(float(sensor.consumption_thisday) * (float(pyscript.gjennomsnittlig_strompris) - float(pyscript.snittpris_for_strom)), 2)

@time_trigger("cron(0 0 * * *)")
def nullstille_stromkostnad():
    pyscript.total_pris_for_strom = 0

@time_trigger("cron(0 * * * *)")
@state_trigger("input_button.oppdater_strompris")
def strompris():
    #state.set("pyscript.strompris", new_attributes={})
    
    grid_night = float(input_number.nettleie_natt) / 100.0
    grid_day = float(input_number.nettleie_dag) / 100.0
    additional_costs = float(input_number.paslag_strom) / 100.0
    
    entso_valid = False
    if sensor.average_electricity_price_today.prices is not None and len(sensor.average_electricity_price_today.prices) >= 48:
        ld = datetime.strptime(sensor.average_electricity_price_today.prices[47]["time"], "%Y-%m-%d %H:%M:%S%z")
        if datetime.now() < ld.replace(tzinfo=None):
            entso_valid = True
    
    if input_boolean.strompriskilde_tibber == 'on' and sensor.energy_price_gabriel_edlands_veg_16.tomorrow_valid == True:
        pyscript.strompris.updatetime = datetime.now().isoformat()
        pyscript.strompris.source = "Tibber"
        l = []

        for sourceprice in sensor.energy_price_gabriel_edlands_veg_16.raw_today + sensor.energy_price_gabriel_edlands_veg_16.raw_tomorrow:
            pr = {}
            d = datetime.strptime(sourceprice["time"], "%Y-%m-%dT%H:%M:%S.000%z")
            pr["start"] = d.isoformat()
            pr["end"] = (d + timedelta(hours=1)).isoformat()
            pr["spotprice"] = round(float(sourceprice["total"]) - float(additional_costs), 4)
            pr["gridprice"] = round(grid_day if is_peak(d) else grid_night, 4)
            pr["payback"] = round(payback(float(pr["spotprice"])), 4)
            pr["totalprice"] = round(float(pr["spotprice"]) + float(pr["gridprice"]) + float(additional_costs) - float(pr["payback"]), 4)
            l.append(pr)

        pyscript.strompris.prices = l
    elif input_boolean.strompriskilde_nordpool == 'on' and sensor.nordpool.tomorrow_valid == True:
        pyscript.strompris.updatetime = datetime.now().isoformat()
        pyscript.strompris.source = "Nordpool"
        l = []

        for sourceprice in sensor.nordpool.raw_today + sensor.nordpool.raw_tomorrow:
            pr = {}
            d = sourceprice["start"]
            pr["start"] = d.isoformat()
            pr["end"] = (d + timedelta(hours=1)).isoformat()
            pr["spotprice"] = round(float(sourceprice["value"]), 4)
            pr["gridprice"] = round(grid_day if is_peak(d) else grid_night, 4)
            pr["payback"] = round(payback(float(pr["spotprice"])), 4)
            pr["totalprice"] = round(float(pr["spotprice"]) + float(pr["gridprice"]) + float(additional_costs) - float(pr["payback"]), 4)
            l.append(pr)

        pyscript.strompris.prices = l
    elif input_boolean.strompriskilde_entsoe == 'on' and entso_valid == True:
        pyscript.strompris.updatetime = datetime.now().isoformat()
        pyscript.strompris.source = "Entso-e"
        l = []
        
        for sourceprice in sensor.average_electricity_price_today.prices:
            pr = {}
            d = datetime.strptime(sourceprice["time"], "%Y-%m-%d %H:%M:%S%z")
            pr["start"] = d.isoformat()
            pr["end"] = (d + timedelta(hours=1)).isoformat()
            pr["spotprice"] = round(float(sourceprice["price"]), 4)
            pr["gridprice"] = round(grid_day if is_peak(d) else grid_night, 4)
            pr["payback"] = round(payback(float(pr["spotprice"])), 4)
            pr["totalprice"] = round(float(pr["spotprice"]) + float(pr["gridprice"]) + float(additional_costs) - float(pr["payback"]), 4)
            l.append(pr)
        
        pyscript.strompris.prices = l
    
    price_sum = 0.0
    peak_sum = 0.0
    offpeak1_sum = 0.0
    offpeak2_sum = 0.0
    high = -1000
    low = 1000

    for sourceprice in pyscript.strompris.prices:
        if datetime.now() >= datetime.fromisoformat(sourceprice["start"]).replace(tzinfo=None) and datetime.now() < datetime.fromisoformat(sourceprice["end"]).replace(tzinfo=None):
            pyscript.strompris = round(float(sourceprice["totalprice"]), 4)
            pyscript.totalstrompris = round(float(sourceprice["totalprice"]), 4)
            pyscript.nettleie = round(float(sourceprice["gridprice"]), 4)
            pyscript.stromstotte = round(float(sourceprice["payback"]), 4)
            pyscript.spotpris = round(float(sourceprice["spotprice"]), 4)
        
        if datetime.now().day == datetime.fromisoformat(sourceprice["start"]).day:
            price_sum += sourceprice["totalprice"]

            if datetime.fromisoformat(sourceprice["start"]).hour < 8:
                offpeak1_sum += sourceprice["totalprice"]
            elif datetime.fromisoformat(sourceprice["start"]).hour < 20:
                peak_sum += sourceprice["totalprice"]
            else:
                offpeak2_sum += sourceprice["totalprice"]
            
            if high < sourceprice["totalprice"]:
                high = sourceprice["totalprice"]
            
            if low > sourceprice["totalprice"]:
                low = sourceprice["totalprice"]
    
    pyscript.gjennomsnittlig_strompris = round(float(price_sum) / 24.0, 4)
    pyscript.peak_strompris = round(float(peak_sum) / 12.0, 4)
    pyscript.offpeak_1_strompris = round(float(offpeak1_sum) / 8.0, 4)
    pyscript.offpeak_2_strompris = round(float(offpeak2_sum) / 4.0, 4)
    pyscript.hoyeste_strompris = round(float(high), 4)
    pyscript.laveste_strompris = round(float(low), 4)

def is_peak(t : datetime):
    if t.isoweekday() >= 6 or t.hour <= 5 or t.hour >= 22 or t.date() in NOR_HOLIDAYS:
        return False
    else:
        return True

def payback(spot : float):
    return max((spot - 0.9125) * 0.9, 0.0)

 

Endret 23. januar av stigvi

[stigvi]

Og i visning av priser så setter jeg en "opacity" mindre enn 1 slik at jeg må ha kurvene for Tibber, Nordpool og Entso oppå hverandre for å gi en mørk blåfarge. Da er det lett å se på kurven om pris fra en av de mangler.