[文档]classJobInfo(Generic[JobType,JobKey]):""" Stores information of each job. """
[文档]def__init__(self,key:JobKey,job:JobType):self.key=keyself.jobs=[(job,"")]self.narrowing_count=0# not used
def__hash__(self):returnhash((self.key,*tuple(self.jobs)))def__eq__(self,o):returntype(self)istype(o)andself.key==o.keyandself.job==o.jobdef__repr__(self):returnf"<JobInfo {self.key!s}>"@propertydefjob(self)->JobType:""" Get the latest available job. :return: The latest available job. """job,_=self.jobs[-1]returnjob@propertydefmerged_jobs(self):forjob,job_typeinself.jobs:ifjob_type=="merged":yieldjob@propertydefwidened_jobs(self):forjob,job_typeinself.jobs:ifjob_type=="widened":yieldjob
[文档]defadd_job(self,job,merged=False,widened=False):""" Appended a new job to this JobInfo node. :param job: The new job to append. :param bool merged: Whether it is a merged job or not. :param bool widened: Whether it is a widened job or not. """job_type=""ifmerged:job_type="merged"elifwidened:job_type="widened"self.jobs.append((job,job_type))