Get list of missing time buckets between two items. Args: prev_item: Previous time bucket curr_item: Current time bucket granularity: Time granularity level Returns: List of missing time bucket dictionarie
(prev_item: Dict, curr_item: Dict, granularity: str)
| 101 | |
| 102 | @staticmethod |
| 103 | def get_missing_buckets(prev_item: Dict, curr_item: Dict, granularity: str) -> List[Dict]: |
| 104 | """ |
| 105 | Get list of missing time buckets between two items. |
| 106 | |
| 107 | Args: |
| 108 | prev_item: Previous time bucket |
| 109 | curr_item: Current time bucket |
| 110 | granularity: Time granularity level |
| 111 | |
| 112 | Returns: |
| 113 | List of missing time bucket dictionaries |
| 114 | """ |
| 115 | missing = [] |
| 116 | |
| 117 | if granularity == TimeUtils.GRANULARITY_HOUR: |
| 118 | prev_hour = int(prev_item.get('HH', '00')) |
| 119 | curr_hour = int(curr_item.get('HH', '00')) |
| 120 | |
| 121 | for hour in range(prev_hour + 1, curr_hour): |
| 122 | missing.append({'HH': f"{hour:02d}", 'MI': '00'}) |
| 123 | |
| 124 | elif granularity == TimeUtils.GRANULARITY_MINUTE: |
| 125 | prev_hour = int(prev_item.get('HH', '00')) |
| 126 | prev_min = int(prev_item.get('MI', '00')) |
| 127 | curr_hour = int(curr_item.get('HH', '00')) |
| 128 | curr_min = int(curr_item.get('MI', '00')) |
| 129 | |
| 130 | # Convert to total minutes for easier calculation |
| 131 | prev_total = prev_hour * 60 + prev_min |
| 132 | curr_total = curr_hour * 60 + curr_min |
| 133 | |
| 134 | for total_min in range(prev_total + 1, curr_total): |
| 135 | hour = (total_min // 60) % 24 |
| 136 | minute = total_min % 60 |
| 137 | missing.append({'HH': f"{hour:02d}", 'MI': f"{minute:02d}"}) |
| 138 | |
| 139 | else: # HH:MI:S10 |
| 140 | prev_hour = int(prev_item.get('HH', '00')) |
| 141 | prev_min = int(prev_item.get('MI', '00')) |
| 142 | prev_s10 = TimeUtils.parse_s10_value(prev_item.get('S10', '00')) |
| 143 | |
| 144 | curr_hour = int(curr_item.get('HH', '00')) |
| 145 | curr_min = int(curr_item.get('MI', '00')) |
| 146 | curr_s10 = TimeUtils.parse_s10_value(curr_item.get('S10', '00')) |
| 147 | |
| 148 | # Convert to total 10-second buckets |
| 149 | prev_total = prev_hour * 360 + prev_min * 6 + prev_s10 // 10 |
| 150 | curr_total = curr_hour * 360 + curr_min * 6 + curr_s10 // 10 |
| 151 | |
| 152 | for total_s10 in range(prev_total + 1, curr_total): |
| 153 | hour = (total_s10 // 360) % 24 |
| 154 | minute = (total_s10 % 360) // 6 |
| 155 | second = (total_s10 % 6) * 10 |
| 156 | missing.append({ |
| 157 | 'HH': f"{hour:02d}", |
| 158 | 'MI': f"{minute:02d}", |
| 159 | 'S10': f"{second:02d}" |
| 160 | }) |