sequencing_job.py (6894B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\cosmetics\client\ships\skins\live_data\sequencing_job.py 2 import datetime 3 from cosmetics.client.ships.skins.errors import convert_sequencing_error_to_proto, convert_sequencing_error_from_proto 4 from datetimeutils import datetime_to_filetime 5 from eve.common.lib import appConst 6 from itertoolsext.Enum import Enum 7 from google.protobuf.timestamp_pb2 import Timestamp 8 from eveProto.generated.eve_public.cosmetic.ship.skin.thirdparty.sequencing.job.job_pb2 import Attributes 9 from eveProto.generated.eve_public.cosmetic.ship.skin.thirdparty.thirdparty_pb2 import Identifier as SkinIdentifier 10 from eveProto.generated.eve_public.character.character_pb2 import Identifier as CharacterIdentifier 11 from localization import GetByLabel, formatters 12 from localization.formatters import dateTimeFormatters, timeIntervalFormatters 13 14 @Enum 15 16 class SequencingJobState(object): 17 PENDING = 'PENDING' 18 STARTED = 'STARTED' 19 FAILED = 'FAILED' 20 COMPLETED = 'COMPLETED' 21 22 23 class SequencingJob(object): 24 25 def __init__(self, job_id, hex_id, char_id, nb_runs): 26 self._job_id = job_id 27 self._skin_hex = hex_id 28 self._character_id = char_id 29 self._nb_runs = nb_runs 30 self._state = None 31 self._time_started = None 32 self._time_completed = None 33 self._planned_completion_time = None 34 self._failure_error = None 35 36 @property 37 def job_id(self): 38 return self._job_id 39 40 @property 41 def character_id(self): 42 return self._character_id 43 44 @property 45 def skin_hex(self): 46 return self._skin_hex 47 48 @property 49 def nb_runs(self): 50 return self._nb_runs 51 52 @property 53 def state(self): 54 return self._state 55 56 @property 57 def time_started(self): 58 if self._state != SequencingJobState.PENDING: 59 return self._time_started 60 61 @property 62 def time_completed(self): 63 if self._state in [SequencingJobState.COMPLETED, SequencingJobState.FAILED]: 64 return self._time_completed 65 66 @property 67 def time_remaining(self): 68 if self.state != SequencingJobState.STARTED: 69 return None 70 time_remaining = self.planned_completion_time - datetime.datetime.now() 71 if time_remaining < datetime.timedelta(0): 72 return None 73 return time_remaining 74 75 def get_completed_ratio(self): 76 if self.state == SequencingJobState.COMPLETED: 77 return 1.0 78 elif self.state == SequencingJobState.STARTED: 79 seconds_total = (self.planned_completion_time - self.time_started).total_seconds() 80 if self.time_remaining: 81 ratio = 1.0 - self.time_remaining.total_seconds() / float(seconds_total) 82 else: 83 ratio = 1.0 84 return max(0.0, min(ratio, 1.0)) 85 else: 86 return 0.0 87 88 def get_time_remaining_text(self): 89 if not self.time_remaining: 90 return None 91 seconds = int(self.time_remaining.total_seconds() * appConst.SEC) 92 if seconds > appConst.HOUR: 93 time_left = timeIntervalFormatters.FormatTimeIntervalShortWritten(seconds, showTo='minute') 94 else: 95 time_left = timeIntervalFormatters.FormatTimeIntervalShortWritten(seconds) 96 return GetByLabel('UI/Personalization/ShipSkins/SKINR/Studio/TimeLeft', time_left=time_left) 97 98 @property 99 def planned_completion_time(self): 100 return self._planned_completion_time 101 102 @property 103 def failure_error(self): 104 return self._failure_error 105 106 def is_active(self): 107 return self.state in [SequencingJobState.STARTED] 108 109 def copy_data_from_other(self, job): 110 self._skin_hex = job.skin_hex 111 self._character_id = job.character_id 112 self._nb_runs = job.nb_runs 113 self._state = job.state 114 self._time_started = job.time_started 115 self._time_completed = job.time_completed 116 self._planned_completion_time = job.planned_completion_time 117 self._failure_error = job.failure_error 118 119 def admin_set_state(self, state): 120 self._state = state 121 122 def admin_set_started_time(self, time): 123 self._time_started = time 124 125 def admin_set_completed_time(self, time): 126 self._time_completed = time 127 128 def admin_set_planned_completion_time(self, time): 129 self._planned_completion_time = time 130 131 def admin_set_failure_error(self, value): 132 self._failure_error = value 133 134 def __eq__(self, other): 135 return self.job_id == other.job_id and self.character_id == other.character_id and self.skin_hex == other.skin_hex and self.nb_runs == other.nb_runs and self.state == other.state 136 137 def __ne__(self, other): 138 return not self.__eq__(other) 139 140 @staticmethod 141 def build_from_proto(job_id, payload): 142 job = SequencingJob(job_id=job_id, hex_id=payload.skin.hex, char_id=payload.sequencer.sequential, nb_runs=payload.quantity) 143 if payload.HasField('pending'): 144 job._state = SequencingJobState.PENDING 145 elif payload.HasField('started'): 146 job._state = SequencingJobState.STARTED 147 job._time_started = datetime.datetime.fromtimestamp(payload.started.seconds) 148 elif payload.HasField('failed'): 149 job._state = SequencingJobState.FAILED 150 job._failure_error = convert_sequencing_error_from_proto(payload.failed) 151 elif payload.HasField('completed'): 152 job._state = SequencingJobState.COMPLETED 153 job._time_completed = datetime.datetime.fromtimestamp(payload.completed.seconds) 154 job._planned_completion_time = datetime.datetime.fromtimestamp(payload.planned_completion.seconds) 155 return job 156 157 @staticmethod 158 def build_proto_from_job(sequencing_job, hex_id): 159 planned_completion = Timestamp() 160 planned_completion.FromDatetime(sequencing_job.planned_completion_time) 161 proto_job = Attributes(skin=SkinIdentifier(hex=hex_id), sequencer=CharacterIdentifier(sequential=sequencing_job.character_id), quantity=sequencing_job.nb_runs, planned_completion=planned_completion) 162 if sequencing_job.state == SequencingJobState.PENDING: 163 proto_job.pending = True 164 elif sequencing_job.state == SequencingJobState.STARTED: 165 started_time = Timestamp() 166 started_time.FromDatetime(sequencing_job.time_started) 167 proto_job.started.CopyFrom(started_time) 168 elif sequencing_job.state == SequencingJobState.FAILED: 169 proto_job.failed = convert_sequencing_error_to_proto(sequencing_job.failure_error) 170 elif sequencing_job.state == SequencingJobState.COMPLETED: 171 completed_time = Timestamp() 172 completed_time.FromDatetime(sequencing_job.time_completed) 173 proto_job.completed.CopyFrom(completed_time) 174 return proto_job