335 lines
12 KiB
Python
335 lines
12 KiB
Python
|
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_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
|
||
|
|
job["dut_job_phases"][-1]["end_time"] = datetime.now().isoformat()
|
||
|
|
|
||
|
|
# Create a new phase
|
||
|
|
phase_data = {
|
||
|
|
"name": phase_name,
|
||
|
|
"start_time": datetime.now().isoformat(),
|
||
|
|
"end_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.")
|
||
|
|
|
||
|
|
# 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"
|
||
|
|
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"] == "":
|
||
|
|
job["dut_job_phases"][-1]["end_time"] = datetime.now().isoformat()
|
||
|
|
|
||
|
|
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()
|