Resolve "Comptadors per un camí de dades independent" (et/somenergia-…

…jardiner!127)

Closes #163

Merge branch '163-comptadors-per-un-cami-de-dades-independent' into 'main'

dbt_jardiner/models/jardiner/intermediate/dset/int_dset_meter__readings.sql

@@ -0,0 +1,47 @@
+{{ config(materialized="view") }}
+
+with quarterhourly_spine as (
+  select generate_series('2023-12-01', now(), '15 minutes') as ts
+),
+meter_metadata as (
+  select
+    metadata.plant_uuid::uuid,
+    metadata.plant_name,
+    metadata.signal_name,
+    metadata.metric_name,
+    metadata.device_name,
+    metadata.device_type,
+    metadata.device_uuid,
+    metadata.device_parent,
+    metadata.device_parent_uuid,
+    metadata.signal_uuid,
+    metadata.is_enabled
+  from {{ ref("raw_gestio_actius__signal_denormalized") }} as metadata
+  where
+    metadata.device_type = 'meter'
+    and metadata.is_enabled
+    and metadata.metric_name ilike 'energia_%'
+),
+raw_meter_readings as (
+  select
+    meter_readings.ts,
+    meter_readings.signal_value,
+    meter_readings.signal_uuid,
+    meter_readings.signal_unit,
+    null as queried_at
+  from {{ ref("raw_dset_meter__readings") }} as meter_readings
+  where meter_readings.signal_uuid is not null
+),
+meter_readings_with_metadata as (
+  select
+    quarterhourly_spine.ts,
+    meter_metadata.*,
+    raw_meter_readings.signal_value,
+    raw_meter_readings.signal_unit,
+    raw_meter_readings.queried_at,
+    now() as materialized_at
+  from quarterhourly_spine
+    left join meter_metadata on true
+    left join raw_meter_readings using (signal_uuid, ts)
+)
+select * from meter_readings_with_metadata

dbt_jardiner/models/jardiner/intermediate/dset/int_dset_meter__readings_wide.sql

@@ -0,0 +1,31 @@
+{{ config(materialized="view") }}
+
+with meter_readings_wide as (
+  select
+    mr.ts,
+    mr.plant_uuid,
+    mr.plant_name,
+    {{
+        pivot(
+            column="metric_name",
+            names=dbt_utils.get_column_values(table=ref("int_dset_meter__readings"), column="metric_name"),
+            value_column="signal_value",
+            agg="max",
+        )
+    }}
+  from {{ ref("int_dset_meter__readings") }} as mr
+  group by mr.plant_uuid, mr.plant_name, mr.ts
+  order by mr.ts desc
+)
+select
+  ts,
+  plant_uuid,
+  plant_name,
+  energia_activa_exportada as meter_exported_energy,
+  energia_activa_importada as meter_imported_energy,
+  energia_reactiva_q1 as meter_reactive_energy_q1,
+  energia_reactiva_q2 as meter_reactive_energy_q2,
+  energia_reactiva_q3 as meter_reactive_energy_q3,
+  energia_reactiva_q4 as meter_reactive_energy_q4,
+  energia_activa_exportada_instantania as meter_instant_exported_energy
+from meter_readings_wide

dbt_jardiner/models/jardiner/intermediate/dset/int_dset_meter__readings_wide_hourly.sql

@@ -0,0 +1,19 @@
+{{ config(materialized="view") }}
+
+with meter_readings_hourly as (
+  select
+    date_trunc('hour', ts) as start_hour,
+    plant_uuid,
+    plant_name,
+    sum(meter_exported_energy) as meter_exported_energy,
+    sum(meter_imported_energy) as meter_imported_energy,
+    sum(meter_reactive_energy_q1) as meter_reactive_energy_q1,
+    sum(meter_reactive_energy_q2) as meter_reactive_energy_q2,
+    sum(meter_reactive_energy_q3) as meter_reactive_energy_q3,
+    sum(meter_reactive_energy_q4) as meter_reactive_energy_q4,
+    sum(meter_instant_exported_energy) as meter_instant_exported_energy
+  from {{ ref("int_dset_meter__readings_wide") }}
+  group by plant_uuid, plant_name, ts
+  order by ts desc
+)
+select * from meter_readings_hourly

dbt_jardiner/models/jardiner/intermediate/erp/int_erp_meter_registry__hourly.sql

@@ -3,25 +3,25 @@
 
 with meter_registry_hourly_raw as (
   select
-    date_trunc('hour', mr.start_hour) AS start_hour,
+    date_trunc('hour', mr.start_hour) as start_hour,
     mr.meter_id,
     mr.meter_name,
     mr.plant_id,
     p.plant_uuid,
     p.plant_name,
     p.plant_codename,
-    round(avg(export_energy_wh),2) as export_energy_wh,
-    round(avg(import_energy_wh),2) as import_energy_wh
+    round(avg(mr.export_energy_wh), 2) as export_energy_wh,
+    round(avg(mr.import_energy_wh), 2) as import_energy_wh
   from {{ ref('raw_meterregistry') }} as mr
-  left join {{ ref("raw_plantmonitor_plants") }} p using(plant_id)
-  group by date_trunc('hour', start_hour), plant_uuid, plant_id, plant_name, plant_codename, meter_id, meter_name
+    left join {{ ref("raw_plantmonitor_plants") }} as p using (plant_id)
+  group by date_trunc('hour', mr.start_hour), p.plant_uuid, mr.plant_id, p.plant_name, p.plant_codename, mr.meter_id, mr.meter_name
 )
 
 select
   *,
-  CASE
-    WHEN export_energy_wh > 0 THEN 1
-    WHEN export_energy_wh = 0 THEN 0
-    ELSE NULL
-  END as has_energy
-FROM meter_registry_hourly_raw
+  case
+    when export_energy_wh > 0 then 1
+    when export_energy_wh = 0 then 0
+    else null
+  end as has_energy
+from meter_registry_hourly_raw

dbt_jardiner/models/jardiner/intermediate/obt/obt_hourly_incremental.sql

@@ -1,7 +1,11 @@
 {{ config(materialized='table',
         docs={'node_color': '#d05808'}) }}
 
-with obt_base as (
+with erp_meter_registries_before_2024 as (
+  select * from {{ ref('int_erp_meter_registry__hourly') }}
+  where start_hour < '2024-01-01'
+),
+obt_base as (
   select
     spine.start_hour,
     plant_metadata.plant_uuid,
@@ -13,8 +17,8 @@ with obt_base as (
     dset.irradiation as dset_irradiation_wh,
     dset.inverter_exported_energy as dset_inverter_energy_kwh,
     dset.meter_instant_exported_energy as dset_meter_instant_exported_energy_kwh,
-    dset.meter_exported_energy as dset_meter_exported_energy_kwh,
-    dset.meter_imported_energy as dset_meter_imported_energy_kwh,
+    dset_meter_readings.meter_exported_energy as dset_meter_exported_energy_kwh,
+    dset_meter_readings.meter_imported_energy as dset_meter_imported_energy_kwh,
     forecast.forecastdate as forecast_date,
     forecast.energy_kwh as forecast_energy_kwh,
     sr.tilted_irradiation_wh_m2 as satellite_irradiation_wh_m2,
@@ -31,17 +35,18 @@ with obt_base as (
     {#- plant_parameters is the single source of truth about which plants we have, unexpected plants won't join#}
     left join {{ ref('raw_gestio_actius_plant_parameters') }} as plant_metadata on true
     left join {{ ref('int_dset_metrics_wide_hourly') }} as dset using (start_hour, plant_uuid)
+    left join {{ ref('int_dset_meter__readings_wide_hourly') }} as dset_meter_readings using (start_hour, plant_uuid)
     left join {{ ref('int_energy_forecasts__best_from_plantmonitordb') }} as forecast using (start_hour, plant_uuid)
     left join {{ ref('int_satellite_readings__hourly') }} as sr using (start_hour, plant_uuid)
     left join {{ ref('raw_plantlake_omie_historical_price__with_row_number_per_date') }} as omie using (start_hour)
-    left join {{ ref('int_erp_meter_registry__hourly') }} as meter_registry using (start_hour, plant_uuid)
+    left join erp_meter_registries_before_2024 as meter_registry using (start_hour, plant_uuid)
     left join {{ ref('int_plantmonitordb_solarevent__generous') }} as solar_events
       on plant_metadata.plant_uuid = solar_events.plant_uuid and solar_events.day = spine.start_hour::date
 ), obt_derived as (
   select
     *,
-    dset_meter_exported_energy_kwh as meter_exported_energy_kwh,
-    dset_meter_imported_energy_kwh as meter_imported_energy_kwh,
+    coalesce(dset_meter_exported_energy_kwh, erp_meter_exported_energy_kwh) as meter_exported_energy_kwh,
+    coalesce(dset_meter_imported_energy_kwh, erp_meter_imported_energy_kwh) as meter_imported_energy_kwh,
     {# The /1000 is GSTC[W/m2] #}
     (dset_meter_exported_energy_kwh / peak_power_kw) / (nullif(satellite_irradiation_wh_m2, 0.0) / 1000.0) as pr_hourly
   from obt_base

dbt_jardiner/models/jardiner/marts/dm_dashboard_overview_instant.sql

@@ -7,14 +7,14 @@ select
   overview.peak_power_kw as potencia_pic_kw,
   overview.nominal_power_kw as potencia_nominal_kw,
   overview.irradiance_w_m2_last_registered_at as irradiancia_ts,
-  round(overview.irradiance_w_m2, 2) as irradiancia_w_m2,
+  round(overview.irradiance_w_m2::numeric, 2) as irradiancia_w_m2,
   overview.instant_power_plant_kw_registered_at as potencia_inst_ts,
-  round(overview.instant_power_plant_kw, 2) as potencia_inst_kw,
-  round(overview.instant_power_plant_kw / overview.nominal_power_kw, 2) as potencia_inst_vs_nominal,
+  round(overview.instant_power_plant_kw::numeric, 2) as potencia_inst_kw,
+  round((overview.instant_power_plant_kw / overview.nominal_power_kw)::numeric, 2) as potencia_inst_vs_nominal,
   overview.day as dia,
   case
     when overview.day > now() - interval '2 days' then
-      round((overview.meter_exported_energy_kwh - overview.solargis_meter_expected_energy_kwh) / 1000, 2)
+      round(((overview.meter_exported_energy_kwh - overview.solargis_meter_expected_energy_kwh) / 1000)::numeric, 2)
   end as energia_perduda_ahir_mwh,
-  round((overview.meter_exported_energy_kwh - overview.solargis_meter_expected_energy_kwh) / 1000, 2) as energia_perduda_mwh
+  round(((overview.meter_exported_energy_kwh - overview.solargis_meter_expected_energy_kwh) / 1000)::numeric, 2) as energia_perduda_mwh
 from {{ ref("int_plants_overview_instant") }} as overview

dbt_jardiner/models/jardiner/marts/dm_dset_meter__readings.sql

@@ -0,0 +1,14 @@
+{{ config(materialized='table') }}
+
+select
+  ts,
+  plant_uuid as uuid_planta,
+  plant_name as nom_planta,
+  meter_exported_energy as energia_activa_exportada,
+  meter_imported_energy as energia_activa_importada,
+  meter_reactive_energy_q1 as energia_reactiva_q1,
+  meter_reactive_energy_q2 as energia_reactiva_q2,
+  meter_reactive_energy_q3 as energia_reactiva_q3,
+  meter_reactive_energy_q4 as energia_reactiva_q4,
+  meter_instant_exported_energy as energia_activa_exportada_instantania
+from {{ ref("int_dset_meter__readings_wide") }}

dbt_jardiner/models/jardiner/marts/dm_plant_production_monthly.sql

@@ -9,13 +9,13 @@ select
   potencia_pic_kw as potencia_pic_kw,
   sum(energia_instantania_inversor_kwh) / 1000.0 as energia_instantania_inversor_mwh,
   count(energia_instantania_inversor_kwh) as energia_inversor_count,
-  round(sum(energia_exportada_instantania_comptador_kwh) / 1000.0, 1) as energia_exportada_instantania_comptador_mwh,
+  round((sum(energia_exportada_instantania_comptador_kwh) / 1000.0)::numeric, 1) as energia_exportada_instantania_comptador_mwh,
   count(energia_exportada_instantania_comptador_kwh) as energia_exportada_instantania_comptador_count,
-  round(sum(energia_exportada_comptador_kwh) / 1000.0, 1) as energia_exportada_comptador_mwh,
-  round(sum(energia_importada_comptador_kwh) / 1000.0, 1) as energia_importada_comptador_mwh,
+  round((sum(energia_exportada_comptador_kwh) / 1000.0)::numeric, 1) as energia_exportada_comptador_mwh,
+  round((sum(energia_importada_comptador_kwh) / 1000.0)::numeric, 1) as energia_importada_comptador_mwh,
   --min(data_prediccio) as data_prediccio,
-  round(sum(energia_predita_meteologica_kwh) / 1000.0, 1) as energia_predita_meteologica_mwh,
-  round(sum(energia_esperada_solargis_kwh) / 1000.0, 1) as energia_esperada_solargis_mwh,
+  round((sum(energia_predita_meteologica_kwh) / 1000.0)::numeric, 1) as energia_predita_meteologica_mwh,
+  round((sum(energia_esperada_solargis_kwh) / 1000.0)::numeric, 1) as energia_esperada_solargis_mwh,
   avg(preu_omie_eur_mwh) as preu_omie_eur_mwh,
   sum(irradiation_wh_m2) as irradiation_wh_m2,
   sum(irradiacio_satellit_wh_m2) as irradiacio_satellit_wh_m2,
@@ -25,9 +25,9 @@ select
   sum(hora_disponible) as hora_disponible,
   sum(hora_total) as hora_total,
   sum(hora_disponible) / nullif(sum(hora_total), 0) as disponibilitat,
-  round(sum(energia_desviada_omie_kwh) / 1000.0, 1) as energia_desviada_omie_mwh,
+  round((sum(energia_desviada_omie_kwh) / 1000.0)::numeric, 1) as energia_desviada_omie_mwh,
   round((sum(abs(energia_desviada_omie_kwh_absolute)) / 1000.0)::numeric, 1) as energia_desviada_omie_mwh_absolute, {# should it be abs or we let compensate itself? #}
-  round(sum(energia_perduda_kwh) / 1000.0, 1) as energia_perduda_mwh,
+  round((sum(energia_perduda_kwh) / 1000.0)::numeric, 1) as energia_perduda_mwh,
   round((1 - sum(energia_predita_meteologica_kwh) / nullif(sum(energia_exportada_comptador_kwh), 0))::numeric, 2) as energia_desviada_percent
 from {{ ref("dm_plant_production_daily") }}
 group by date_trunc('month', dia), plant_uuid, nom_planta, tecnologia, potencia_pic_kw

dbt_jardiner/models/jardiner/raw/dset/_raw_dset__models.yaml

@@ -44,3 +44,6 @@ models:
     description: >
       Conté l'últim batch de dades, descartant ts i value. Es fa servir per obtenir signal_last_ts
       i signal_last_value.
+  - name: raw_dset_meter__readings
+    description: >
+      Lectures de tots els comptadors llegits individualment via dset

dbt_jardiner/models/jardiner/raw/dset/_raw_dset__sources.yaml

@@ -11,3 +11,9 @@ sources:
           S'escriuen en base de dades en format JSONB recollint el JSON cru
           de la API de DSET.
           [documentació](https://documenter.getpostman.com/view/6601984/2s946cga7J)
+      - name: dset_meters_readings
+        description: >
+          Lectures dels comptadors llegides amb el DAG
+          [dset_reader_meteres_dag_v2](https://airflow.somenergia.coop/dags/dset_reader_meters_dag_v2)
+          amb el codi a [somenergia-plant-reader](https://gitlab.somenergia.coop/et/somenergia-plant-reader/-/blob/main/dags/dset_get_recent_meters_dag.py).
+          Dset refresca els comptadors cada 24h i el dag pregunta horariament si hi ha canvis.

dbt_jardiner/models/jardiner/raw/dset/raw_dset_meter__readings.sql

@@ -0,0 +1,28 @@
+{{ config(materialized="view") }}
+
+with meter_readings as (
+  select
+    group_name as dset_plant_name,
+    signal_device_external_description as signal_device_type,
+    signal_description,
+    ts::timestamp at time zone signal_tz as ts,
+    signal_value,
+    group_id as dset_plant_id,
+    group_code as dset_plant_code,
+    signal_id as dset_signal_id,
+    signal_code as dset_signal_code,
+    signal_type,
+    signal_frequency,
+    signal_is_virtual::boolean as signal_is_virtual,
+    signal_tz,
+    signal_last_ts::timestamp at time zone signal_tz as signal_last_ts,
+    signal_last_value,
+    signal_unit,
+    case
+      when signal_external_id ~ e'^[[:xdigit:]]{8}-([[:xdigit:]]{4}-){3}[[:xdigit:]]{12}$' then signal_external_id::uuid -- noqa: LT01
+    end as signal_uuid,
+    signal_external_id as signal_uuid_raw,
+    signal_device_external_id as signal_device_uuid
+  from {{ source("lake", "dset_meters_readings") }}
+)
+select * from meter_readings