init

bin/ci/custom_logger.py

@@ -0,0 +1,353 @@
+import argparse
+import logging
+from datetime import datetime
+from pathlib import Path
+
+from structured_logger import StructuredLogger
+
+
+class CustomLogger:
+    def __init__(self, log_file):
+        self.log_file = log_file
+        self.logger = StructuredLogger(file_name=self.log_file)
+
+    def get_last_dut_job(self):
+        """
+        Gets the details of the most recent DUT job.
+
+        Returns:
+            dict: Details of the most recent DUT job.
+
+        Raises:
+            ValueError: If no DUT jobs are found in the logger's data.
+        """
+        try:
+            job = self.logger.data["dut_jobs"][-1]
+        except KeyError:
+            raise ValueError(
+                "No DUT jobs found. Please create a job via create_dut_job call."
+            )
+
+        return job
+
+    def update(self, **kwargs):
+        """
+        Updates the log file with provided key-value pairs.
+
+        Args:
+            **kwargs: Key-value pairs to be updated.
+
+        """
+        with self.logger.edit_context():
+            for key, value in kwargs.items():
+                self.logger.data[key] = value
+
+    def create_dut_job(self, **kwargs):
+        """
+        Creates a new DUT job with provided key-value pairs.
+
+        Args:
+            **kwargs: Key-value pairs for the new DUT job.
+
+        """
+        with self.logger.edit_context():
+            if "dut_jobs" not in self.logger.data:
+                self.logger.data["dut_jobs"] = []
+            new_job = {
+                "status": "",
+                "submitter_start_time": datetime.now().isoformat(),
+                "dut_submit_time": "",
+                "dut_start_time": "",
+                "dut_end_time": "",
+                "dut_duration_time": "",
+                "dut_name": "",
+                "dut_state": "pending",
+                "dut_job_phases": [],
+                **kwargs,
+            }
+            self.logger.data["dut_jobs"].append(new_job)
+
+    def update_dut_job(self, key, value):
+        """
+        Updates the last DUT job with a key-value pair.
+
+        Args:
+            key : The key to be updated.
+            value: The value to be assigned.
+
+        """
+        with self.logger.edit_context():
+            job = self.get_last_dut_job()
+            job[key] = value
+
+    def update_status_fail(self, reason=""):
+        """
+        Sets the status of the last DUT job to 'fail' and logs the failure reason.
+
+        Args:
+            reason (str, optional): The reason for the failure. Defaults to "".
+
+        """
+        with self.logger.edit_context():
+            job = self.get_last_dut_job()
+            job["status"] = "fail"
+            job["dut_job_fail_reason"] = reason
+
+    def create_job_phase(self, phase_name):
+        """
+        Creates a new job phase for the last DUT job.
+
+        Args:
+            phase_name : The name of the new job phase.
+
+        """
+        with self.logger.edit_context():
+            job = self.get_last_dut_job()
+            if job["dut_job_phases"] and job["dut_job_phases"][-1]["end_time"] == "":
+                # If the last phase exists and its end time is empty, set the end time
+                timestamp = datetime.now().isoformat()
+                job["dut_job_phases"][-1]["end_time"] = timestamp
+                job["dut_job_phases"][-1]["duration_time"] = self.get_duration_time(job["dut_job_phases"][-1]["start_time"], timestamp)
+
+            # Create a new phase
+            phase_data = {
+                "name": phase_name,
+                "start_time": datetime.now().isoformat(),
+                "end_time": "",
+                "duration_time": "",
+            }
+            job["dut_job_phases"].append(phase_data)
+
+    def check_dut_timings(self, job):
+        """
+        Check the timing sequence of a job to ensure logical consistency.
+
+        The function verifies that the job's submission time is not earlier than its start time and that
+        the job's end time is not earlier than its start time. If either of these conditions is found to be true,
+        an error is logged for each instance of inconsistency.
+
+        Args:
+        job (dict): A dictionary containing timing information of a job. Expected keys are 'dut_start_time',
+                    'dut_submit_time', and 'dut_end_time'.
+
+        Returns:
+        None: This function does not return a value; it logs errors if timing inconsistencies are detected.
+
+        The function checks the following:
+        - If 'dut_start_time' and 'dut_submit_time' are both present and correctly sequenced.
+        - If 'dut_start_time' and 'dut_end_time' are both present and correctly sequenced.
+        """
+
+        # Check if the start time and submit time exist
+        if job.get("dut_start_time") and job.get("dut_submit_time"):
+            # If they exist, check if the submission time is before the start time
+            if job["dut_start_time"] < job["dut_submit_time"]:
+                logging.error("Job submission is happening before job start.")
+
+        # Check if the start time and end time exist
+        if job.get("dut_start_time") and job.get("dut_end_time"):
+            # If they exist, check if the end time is after the start time
+            if job["dut_end_time"] < job["dut_start_time"]:
+                logging.error("Job ended before it started.")
+
+    def get_duration_time(self, start_time, end_time):
+        """
+        Computes duration time, in minutes and seconds.
+        """
+        try:
+            start = datetime.fromisoformat(start_time)
+            end = datetime.fromisoformat(end_time)
+            duration = end - start
+            return str(duration)
+        except ValueError:
+            return ""
+
+    # Method to update DUT start, submit and end time
+    def update_dut_time(self, value, custom_time):
+        """
+        Updates DUT start, submit, and end times.
+
+        Args:
+            value : Specifies which DUT time to update. Options: 'start', 'submit', 'end'.
+            custom_time : Custom time to set. If None, use current time.
+
+        Raises:
+            ValueError: If an invalid argument is provided for value.
+
+        """
+        with self.logger.edit_context():
+            job = self.get_last_dut_job()
+            timestamp = custom_time if custom_time else datetime.now().isoformat()
+            if value == "start":
+                job["dut_start_time"] = timestamp
+                job["dut_state"] = "running"
+            elif value == "submit":
+                job["dut_submit_time"] = timestamp
+                job["dut_state"] = "submitted"
+            elif value == "end":
+                job["dut_end_time"] = timestamp
+                job["dut_state"] = "finished"
+                job["dut_duration_time"] = self.get_duration_time(job["dut_start_time"], job["dut_end_time"])
+            else:
+                raise ValueError(
+                    "Error: Invalid argument provided for --update-dut-time. Use 'start', 'submit', 'end'."
+                )
+            # check the sanity of the partial structured log
+            self.check_dut_timings(job)
+
+    def close_dut_job(self):
+        """
+        Closes the most recent DUT (Device Under Test) job in the logger's data.
+
+        The method performs the following operations:
+        1. Validates if there are any DUT jobs in the logger's data.
+        2. If the last phase of the most recent DUT job has an empty end time, it sets the end time to the current time.
+
+        Raises:
+            ValueError: If no DUT jobs are found in the logger's data.
+        """
+        with self.logger.edit_context():
+            job = self.get_last_dut_job()
+            # Check if the last phase exists and its end time is empty, then set the end time
+            if job["dut_job_phases"] and job["dut_job_phases"][-1]["end_time"] == "":
+                timestamp = datetime.now().isoformat()
+                job["dut_job_phases"][-1]["end_time"] = timestamp
+                job["dut_job_phases"][-1]["duration_time"] = self.get_duration_time(job["dut_job_phases"][-1]["start_time"], timestamp)
+
+    def close(self):
+        """
+        Closes the most recent DUT (Device Under Test) job in the logger's data.
+
+        The method performs the following operations:
+        1. Determines the combined status of all DUT jobs.
+        2. Sets the submitter's end time to the current time.
+        3. Updates the DUT attempt counter to reflect the total number of DUT jobs.
+
+        """
+        with self.logger.edit_context():
+            job_status = []
+            for job in self.logger.data["dut_jobs"]:
+                if "status" in job:
+                    job_status.append(job["status"])
+
+            if not job_status:
+                job_combined_status = "null"
+            else:
+                # Get job_combined_status
+                if "pass" in job_status:
+                    job_combined_status = "pass"
+                else:
+                    job_combined_status = "fail"
+
+            self.logger.data["job_combined_status"] = job_combined_status
+            self.logger.data["dut_attempt_counter"] = len(self.logger.data["dut_jobs"])
+            job["submitter_end_time"] = datetime.now().isoformat()
+
+
+def process_args(args):
+    # Function to process key-value pairs and call corresponding logger methods
+    def process_key_value_pairs(args_list, action_func):
+        if not args_list:
+            raise ValueError(
+                f"No key-value pairs provided for {action_func.__name__.replace('_', '-')}"
+            )
+        if len(args_list) % 2 != 0:
+            raise ValueError(
+                f"Incomplete key-value pairs for {action_func.__name__.replace('_', '-')}"
+            )
+        kwargs = dict(zip(args_list[::2], args_list[1::2]))
+        action_func(**kwargs)
+
+    # Create a CustomLogger object with the specified log file path
+    custom_logger = CustomLogger(Path(args.log_file))
+
+    if args.update:
+        process_key_value_pairs(args.update, custom_logger.update)
+
+    if args.create_dut_job:
+        process_key_value_pairs(args.create_dut_job, custom_logger.create_dut_job)
+
+    if args.update_dut_job:
+        key, value = args.update_dut_job
+        custom_logger.update_dut_job(key, value)
+
+    if args.create_job_phase:
+        custom_logger.create_job_phase(args.create_job_phase)
+
+    if args.update_status_fail:
+        custom_logger.update_status_fail(args.update_status_fail)
+
+    if args.update_dut_time:
+        if len(args.update_dut_time) == 2:
+            action, custom_time = args.update_dut_time
+        elif len(args.update_dut_time) == 1:
+            action, custom_time = args.update_dut_time[0], None
+        else:
+            raise ValueError("Invalid number of values for --update-dut-time")
+
+        if action in ["start", "end", "submit"]:
+            custom_logger.update_dut_time(action, custom_time)
+        else:
+            raise ValueError(
+                "Error: Invalid argument provided for --update-dut-time. Use 'start', 'submit', 'end'."
+            )
+
+    if args.close_dut_job:
+        custom_logger.close_dut_job()
+
+    if args.close:
+        custom_logger.close()
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Custom Logger Command Line Tool")
+    parser.add_argument("log_file", help="Path to the log file")
+    parser.add_argument(
+        "--update",
+        nargs=argparse.ZERO_OR_MORE,
+        metavar=("key", "value"),
+        help="Update a key-value pair e.g., --update key1 value1 key2 value2)",
+    )
+    parser.add_argument(
+        "--create-dut-job",
+        nargs=argparse.ZERO_OR_MORE,
+        metavar=("key", "value"),
+        help="Create a new DUT job with key-value pairs (e.g., --create-dut-job key1 value1 key2 value2)",
+    )
+    parser.add_argument(
+        "--update-dut-job",
+        nargs=argparse.ZERO_OR_MORE,
+        metavar=("key", "value"),
+        help="Update a key-value pair in DUT job",
+    )
+    parser.add_argument(
+        "--create-job-phase",
+        help="Create a new job phase (e.g., --create-job-phase name)",
+    )
+    parser.add_argument(
+        "--update-status-fail",
+        help="Update fail as the status and log the failure reason (e.g., --update-status-fail reason)",
+    )
+    parser.add_argument(
+        "--update-dut-time",
+        nargs=argparse.ZERO_OR_MORE,
+        metavar=("action", "custom_time"),
+        help="Update DUT start and end time. Provide action ('start', 'submit', 'end') and custom_time (e.g., '2023-01-01T12:00:00')",
+    )
+    parser.add_argument(
+        "--close-dut-job",
+        action="store_true",
+        help="Close the dut job by updating end time of last dut job)",
+    )
+    parser.add_argument(
+        "--close",
+        action="store_true",
+        help="Updates combined status, submitter's end time and DUT attempt counter",
+    )
+    args = parser.parse_args()
+
+    process_args(args)
+
+
+if __name__ == "__main__":
+    main()