feat(notifications): add hardware_summary action

Add new hardware_summary action in the notifications management
command. This action generates hardware reports based on the subscription
file.

Signed-off-by: Yushan Li <[email protected]>

Author: ximiali

backend/data/notifications/example-hardware-subscription.yaml

@@ -0,0 +1,9 @@
+# Add a subscription file for your hardware platform by following this format.
+# Place the file under: dashboard/backend/data/notifications/subscriptions
+# The file name must end with _hardware.yml or _hardware.yaml
+
+qcs9100-ride: # The hardware platform name
+  origin: maestro
+  default_recipients:
+    - Recipient One <[email protected]>
+    - Recipient Two <[email protected]>

backend/data/notifications/subscriptions/qcs615_hardware.yaml

@@ -0,0 +1,8 @@
+qcs615-ride:
+  origin: maestro
+  default_recipients:
+    - Trilok Soni <[email protected]>
+    - Shiraz Hashim <[email protected]>
+    - Yogesh Lal <[email protected]>
+    - Yijie Yang <[email protected]>
+    - Yushan Li <[email protected]>

backend/data/notifications/subscriptions/qcs6490_hardware.yaml

@@ -0,0 +1,8 @@
+qcs6490-rb3gen2:
+  origin: maestro
+  default_recipients:
+    - Trilok Soni <[email protected]>
+    - Shiraz Hashim <[email protected]>
+    - Yogesh Lal <[email protected]>
+    - Yijie Yang <[email protected]>
+    - Yushan Li <[email protected]>

backend/data/notifications/subscriptions/qcs8300_hardware.yaml

@@ -0,0 +1,8 @@
+qcs8300-ride:
+  origin: maestro
+  default_recipients:
+    - Trilok Soni <[email protected]>
+    - Shiraz Hashim <[email protected]>
+    - Yogesh Lal <[email protected]>
+    - Yijie Yang <[email protected]>
+    - Yushan Li <[email protected]>

backend/data/notifications/subscriptions/qcs9100_hardware.yaml

@@ -0,0 +1,8 @@
+qcs9100-ride:
+  origin: maestro
+  default_recipients:
+    - Trilok Soni <[email protected]>
+    - Shiraz Hashim <[email protected]>
+    - Yogesh Lal <[email protected]>
+    - Yijie Yang <[email protected]>
+    - Yushan Li <[email protected]>

backend/data/notifications/subscriptions/x1e80100_hardware.yaml

@@ -0,0 +1,8 @@
+x1e80100:
+  origin: maestro
+  default_recipients:
+    - Trilok Soni <[email protected]>
+    - Shiraz Hashim <[email protected]>
+    - Yogesh Lal <[email protected]>
+    - Yijie Yang <[email protected]>
+    - Yushan Li <[email protected]>

backend/kernelCI/settings.py

@@ -169,6 +169,16 @@ def get_json_env_var(name, default):
                 "--summary-origins=maestro",
             ],
         ),
+        (
+            "33 3 * * 1",
+            "django.core.management.call_command",
+            [
+                "notifications",
+                "--action=hardware_summary",
+                "--send",
+                "--yes",
+            ],
+        ),
     ]
 
 # Email settings for SMTP backend

backend/kernelCI_app/helpers/hardwares.py

@@ -0,0 +1,50 @@
+from kernelCI_app.typeModels.common import StatusCount
+from kernelCI_app.typeModels.hardwareListing import HardwareItem
+
+
+def sanitize_hardware(
+    hardware: dict,
+) -> HardwareItem:
+    """Sanitizes a HardwareItem that was returned by a 'hardwarelisting-like' query
+
+    Returns a HardwareItem object"""
+    hardware_name = hardware["hardware"]
+    platform = hardware["platform"]
+
+    build_status_summary = StatusCount(
+        PASS=hardware["pass_builds"],
+        FAIL=hardware["fail_builds"],
+        NULL=hardware["null_builds"],
+        ERROR=hardware["error_builds"],
+        MISS=hardware["miss_builds"],
+        DONE=hardware["done_builds"],
+        SKIP=hardware["skip_builds"],
+    )
+
+    test_status_summary = StatusCount(
+        PASS=hardware["pass_tests"],
+        FAIL=hardware["fail_tests"],
+        NULL=hardware["null_tests"],
+        ERROR=hardware["error_tests"],
+        MISS=hardware["miss_tests"],
+        DONE=hardware["done_tests"],
+        SKIP=hardware["skip_tests"],
+    )
+
+    boot_status_summary = StatusCount(
+        PASS=hardware["pass_boots"],
+        FAIL=hardware["fail_boots"],
+        NULL=hardware["null_boots"],
+        ERROR=hardware["error_boots"],
+        MISS=hardware["miss_boots"],
+        DONE=hardware["done_boots"],
+        SKIP=hardware["skip_boots"],
+    )
+
+    return HardwareItem(
+        hardware=hardware_name,
+        platform=platform,
+        test_status_summary=test_status_summary,
+        boot_status_summary=boot_status_summary,
+        build_status_summary=build_status_summary,
+    )

backend/kernelCI_app/management/commands/helpers/summary.py

@@ -19,6 +19,9 @@
 type TreeKey = tuple[str, str, str]
 """A tuple (branch, giturl, origin)"""
 
+type HardwareKey = tuple[str, str]
+"""A tuple (hardware, origin)"""
+
 type ReportConfigs = list[dict[str, Any]]
 """A list of dictionaries containing the definition/configuration of a report"""
 
@@ -163,3 +166,61 @@ def get_build_issues_from_checkout(
             result_checkout_issues[checkout_id].append(checkout_issue)
 
     return result_checkout_issues, checkout_builds_without_issues
+
+
+def process_hardware_submissions_files(
+    *,
+    base_dir: Optional[str] = None,
+    signup_folder: Optional[str] = None,
+    hardware_origins: Optional[list[str]] = None,
+) -> tuple[set[HardwareKey], dict[HardwareKey, dict[str, Any]]]:
+    """
+    Processes all hardware submission files and returns the set of HardwareKey
+    and the dict linking each hardware to its default recipients
+    """
+    (base_dir, signup_folder) = _assign_default_folders(
+        base_dir=base_dir, signup_folder=signup_folder
+    )
+
+    hardware_key_set: set[HardwareKey] = set()
+    hardware_prop_map: dict[HardwareKey, dict[str, Any]] = {}
+    """Example:
+    hardware_prop_map[(imx6q-sabrelite, maestro)] = {
+        "default_recipients": [
+            "Recipient One <[email protected]>",
+            "Recipient Two <[email protected]>"
+        ]
+    }
+    """
+
+    full_path = os.path.join(base_dir, signup_folder)
+    for filename in os.listdir(full_path):
+        """Example:
+        Filename: imx6q_hardware.yaml
+
+        imx6q-sabrelite:
+        origin: maestro
+        default_recipients:
+            - Recipient One <[email protected]>
+            - Recipient Two <[email protected]>
+        """
+        if filename.endswith("_hardware.yaml") or filename.endswith("_hardware.yml"):
+            file_path = os.path.join(signup_folder, filename)
+            file_data = read_yaml_file(base_dir=base_dir, file=file_path)
+            for hardware_name, hardware_values in file_data.items():
+                origin = hardware_values.get("origin", DEFAULT_ORIGIN)
+                if hardware_origins is not None and origin not in hardware_origins:
+                    continue
+                default_recipients = hardware_values.get("default_recipients", [])
+
+                hardware_key = (hardware_name, origin)
+                hardware_key_set.add(hardware_key)
+                hardware_prop_map[hardware_key] = {
+                    "default_recipients": default_recipients
+                }
+        else:
+            log_message(
+                f"Skipping file {filename} on loading summary files. Not a hardware yaml file."
+            )
+
+    return hardware_key_set, hardware_prop_map

backend/kernelCI_app/management/commands/notifications.py

@@ -5,7 +5,7 @@
 import sys
 
 from collections import defaultdict
-from datetime import datetime, timezone
+from datetime import datetime, timezone, timedelta
 from types import SimpleNamespace
 from urllib.parse import quote_plus
 
@@ -29,6 +29,7 @@
     TreeKey,
     get_build_issues_from_checkout,
     process_submissions_files,
+    process_hardware_submissions_files,
 )
 from kernelCI_app.queries.notifications import (
     get_checkout_summary_data,
@@ -55,6 +56,12 @@
 from kernelCI_cache.typeModels.databases import PossibleIssueType
 from kernelCI_app.utils import group_status
 
+from kernelCI_app.queries.hardware import (
+    get_hardware_summary_data,
+    get_hardware_listing_data_bulk,
+)
+from kernelCI_app.helpers.hardwares import sanitize_hardware
+
 KERNELCI_RESULTS = "[email protected]"
 KERNELCI_REPLYTO = "[email protected]"
 REGRESSIONS_LIST = "[email protected]"
@@ -723,6 +730,101 @@ def generate_test_report(*, service, test_id, email_args, signup_folder):
     )
 
 
+def generate_hardware_summary_report(
+    *,
+    service,
+    hardware_origins: Optional[list[str]],
+    email_args,
+    signup_folder: Optional[str] = None,
+):
+    """
+    Generate weekly hardware reports for hardware submission file.
+    Emails are sent only for hardware with failed tests.
+    """
+
+    now = datetime.now(timezone.utc)
+    start_date = now - timedelta(days=7)
+    end_date = now
+
+    # process the hardware submission files
+    hardware_key_set, hardware_prop_map = process_hardware_submissions_files(
+        signup_folder=signup_folder,
+        hardware_origins=hardware_origins,
+    )
+
+    # get detailed data for all hardware
+    hardwares_data_raw = get_hardware_summary_data(
+        keys=list(hardware_key_set),
+        start_date=start_date,
+        end_date=end_date,
+    )
+
+    if not hardwares_data_raw:
+        print("No data for hardware summary")
+        return
+
+    hardwares_data_dict = defaultdict(list)
+    for raw in hardwares_data_raw:
+        try:
+            environment_misc = json.loads(raw.get("environment_misc", "{}"))
+            misc = json.loads(raw.get("misc", "{}"))
+        except json.JSONDecodeError:
+            print(f'Error decoding JSON for key: {raw.get("environment_misc")}')
+            continue
+        hardware_id = environment_misc.get("platform")
+        raw["job_id"] = environment_misc.get("job_id")
+        raw["runtime"] = misc.get("runtime")
+        origin = raw.get("test_origin")
+        key = (hardware_id, origin)
+        hardwares_data_dict[key].append(raw)
+
+    # get the total build/boot/test counts for each hardware
+    hardwares_list_raw = get_hardware_listing_data_bulk(
+        keys=list(hardware_key_set),
+        start_date=start_date,
+        end_date=end_date,
+    )
+
+    # Iterate through each hardware record to render report, extract recipient, send email
+    for (hardware_id, origin), hardware_data in hardwares_data_dict.items():
+        hardware_raw = next(
+            (row for row in hardwares_list_raw if row.get("platform") == hardware_id),
+            None,
+        )
+
+        hardware_item = sanitize_hardware(hardware_raw)
+        build_status_group = group_status(hardware_item.build_status_summary)
+        boot_status_group = group_status(hardware_item.boot_status_summary)
+        test_status_group = group_status(hardware_item.test_status_summary)
+
+        # render the template
+        template = setup_jinja_template("hardware_report.txt.j2")
+        report = {}
+        report["content"] = template.render(
+            hardware_id=hardware_id,
+            hardware_data=hardware_data,
+            build_status_group=build_status_group,
+            boot_status_group=boot_status_group,
+            test_status_group=test_status_group,
+        )
+        report["title"] = (
+            f"hardware {hardware_id} summary - {now.strftime("%Y-%m-%d %H:%M %Z")}"
+        )
+
+        # extract recipient
+        hardware_report = hardware_prop_map.get((hardware_id, origin), {})
+        recipients = hardware_report.get("default_recipients", [])
+
+        # send email
+        send_email_report(
+            service=service,
+            report=report,
+            email_args=email_args,
+            signup_folder=signup_folder,
+            recipients=recipients,
+        )
+
+
 def run_fake_report(*, service, email_args):
     report = {}
     report["content"] = "Testing the email sending path..."
@@ -787,6 +889,7 @@ def add_arguments(self, parser):
                 "summary",
                 "fake_report",
                 "test_report",
+                "hardware_summary",
             ],
             help="Action to perform: new_issues, issue_report, summary, fake_report, or test_report",
         )
@@ -841,6 +944,13 @@ def add_arguments(self, parser):
             help="Add recipients for the given tree name (fake_report only)",
         )
 
+        # hardware summary report specific arguments
+        parser.add_argument(
+            "--hardware-origins",
+            type=lambda s: [origin.strip() for origin in s.split(",")],
+            help="Limit hardware summary to specific origins (hardware summary only, comma-separated list)",
+        )
+
     def handle(self, *args, **options):
         # Setup connections
         service = smtp_setup_connection()
@@ -929,3 +1039,13 @@ def handle(self, *args, **options):
             self.stdout.write(
                 self.style.SUCCESS(f"Test report generated for test {test_id}")
             )
+
+        elif action == "hardware_summary":
+            email_args.update = options.get("update_storage", False)
+            hardware_origins = options.get("hardware_origins")
+            generate_hardware_summary_report(
+                service=service,
+                signup_folder=signup_folder,
+                email_args=email_args,
+                hardware_origins=hardware_origins,
+            )