Refactor meter score tables to match leaks/irrigation pattern (#219)

Replace CREATE OR REPLACE with CREATE TABLE IF NOT EXISTS + DELETE + INSERT
for idempotency, and add a new meters_score_{org_id}_agg table grouped by
class_full_name with array_agg of per-device scores and avg/stddev summaries.

Co-authored-by: Vicente Ferreyra <36196252+vicenteferreyra@users.noreply.github.com>

Author: dev-vicente-ferreyra

amiadapters/storage/snowflake.py

@@ -360,9 +360,49 @@ def exec_postprocessor(self, run_id: str, min_date: datetime, max_date: datetime
 
             conn = self.sink_config.connection()
 
+            # Ensure meters_score table exists (no-op after first run)
+            conn.cursor().execute(f"""
+                create table if not exists meters_score_{self.org_id} (
+                    org_id              text,
+                    device_id           text,
+                    class_full_name     text,
+                    first_read          timestamp_ntz,
+                    last_read           timestamp_ntz,
+                    last_read_hours     float,
+                    observations        number(18,0),
+                    hours_in_period     float,
+                    read_freq           float,
+                    unread_freq         float,
+                    stuck_events        number(18,0),
+                    stuck_hours         float,
+                    duration_hours_thr  float,
+                    total_usage         float,
+                    used_hours          float,
+                    idle_hours          float,
+                    avg_hourly_usage    float,
+                    unaccounted         float,
+                    z_last_read         float,
+                    z_unread_freq       float,
+                    z_stuck_hours       float,
+                    z_stuck_events      float,
+                    z_unaccounted       float,
+                    score_last_read     float,
+                    score_unread_freq   float,
+                    score_stuck_hours   float,
+                    score_stuck_events  float,
+                    score_unaccounted   float,
+                    score               float
+                )
+            """)
+
+            # Delete existing rows for this org, then insert fresh results
+            conn.cursor().execute(f"""
+                delete from meters_score_{self.org_id}
+                where org_id = '{self.org_id}'
+            """)
+
             ami_meters_score_sql = f"""
-            create or replace table meters_score_{self.org_id}
-            as
+            insert into meters_score_{self.org_id}
             with cte as
             (
                 select org_id
@@ -465,6 +505,31 @@ def exec_postprocessor(self, run_id: str, min_date: datetime, max_date: datetime
             """
             conn.cursor().execute(ami_meters_score_sql)
 
+            ami_meters_score_agg_sql = f"""
+                create or replace table meters_score_{self.org_id}_agg
+                as
+                select
+                    org_id
+                    , class_full_name
+                    , array_agg(device_id)          as device_ids
+                    , array_agg(score)              as scores
+                    , array_agg(score_last_read)    as scores_last_read
+                    , array_agg(score_unread_freq)  as scores_unread_freq
+                    , array_agg(score_stuck_hours)  as scores_stuck_hours
+                    , array_agg(score_stuck_events) as scores_stuck_events
+                    , array_agg(score_unaccounted)  as scores_unaccounted
+                    , avg(score)                    as avg_score
+                    , stddev(score)                 as stddev_score
+                    , avg(score_last_read)          as avg_score_last_read
+                    , avg(score_unread_freq)        as avg_score_unread_freq
+                    , avg(score_stuck_hours)        as avg_score_stuck_hours
+                    , avg(score_stuck_events)       as avg_score_stuck_events
+                    , avg(score_unaccounted)        as avg_score_unaccounted
+                from meters_score_{self.org_id}
+                group by org_id, class_full_name
+            """
+            conn.cursor().execute(ami_meters_score_agg_sql)
+
             # Ensure leaks table exists (no-op after first run)
             conn.cursor().execute(f"""
                 create table if not exists leaks_{self.org_id} (

test/amiadapters/storage/test_snowflake.py

@@ -269,11 +269,67 @@ def test_exec_postprocessor__creates_irrigation_tables_per_org(self):
         all_sqls = [call[0][0] for call in self.mock_cursor.execute.call_args_list]
         normalized = [self.normalize_sql(sql) for sql in all_sqls]
 
-        # Expects exactly 9 execute calls:
-        # 1 meters_score, 1 leaks CREATE, 1 leaks DELETE, 1 leaks INSERT,
-        # 1 leaks_agg, 1 irrigation CREATE, 1 irrigation DELETE,
-        # 1 irrigation INSERT, 1 irrigation_agg
-        self.assertEqual(9, self.mock_cursor.execute.call_count)
+        # Expects exactly 12 execute calls:
+        # 1 meters_score CREATE, 1 meters_score DELETE, 1 meters_score INSERT,
+        # 1 meters_score_agg,
+        # 1 leaks CREATE, 1 leaks DELETE, 1 leaks INSERT, 1 leaks_agg,
+        # 1 irrigation CREATE, 1 irrigation DELETE, 1 irrigation INSERT, 1 irrigation_agg
+        self.assertEqual(12, self.mock_cursor.execute.call_count)
+
+        # --- meters_score detail table ---
+        meters_score_create = next(
+            (
+                s
+                for s in normalized
+                if "create table if not exists meters_score_org-id" in s.lower()
+            ),
+            None,
+        )
+        self.assertIsNotNone(
+            meters_score_create, "Should create meters_score_{org_id} table"
+        )
+        self.assertIn("class_full_name", meters_score_create)
+        self.assertIn("score", meters_score_create)
+        self.assertIn("stuck_hours", meters_score_create)
+
+        # --- DELETE by org_id ---
+        meters_score_delete = next(
+            (s for s in normalized if "delete from meters_score_org-id" in s.lower()),
+            None,
+        )
+        self.assertIsNotNone(
+            meters_score_delete, "Should DELETE from meters_score_{org_id}"
+        )
+        self.assertIn("org_id = 'org-id'", meters_score_delete)
+
+        # --- INSERT from readings ---
+        meters_score_insert = next(
+            (s for s in normalized if "insert into meters_score_org-id" in s.lower()),
+            None,
+        )
+        self.assertIsNotNone(
+            meters_score_insert, "Should INSERT into meters_score_{org_id}"
+        )
+        self.assertIn("ami_connect.readings", meters_score_insert)
+        self.assertIn("z_scores", meters_score_insert)
+        self.assertIn("2025-01-01", meters_score_insert)
+        self.assertIn("2025-01-31", meters_score_insert)
+
+        # --- agg table ---
+        meters_score_agg = next(
+            (
+                s
+                for s in normalized
+                if "create or replace table meters_score_org-id_agg" in s.lower()
+            ),
+            None,
+        )
+        self.assertIsNotNone(
+            meters_score_agg, "Should create meters_score_{org_id}_agg table"
+        )
+        self.assertIn("array_agg", meters_score_agg)
+        self.assertIn("avg_score", meters_score_agg)
+        self.assertIn("class_full_name", meters_score_agg)
 
         # --- irrigation detail table ---
         irrigation_create = next(