| # Copyright lowRISC contributors. | |
| # Licensed under the Apache License, Version 2.0, see LICENSE for details. | |
| # SPDX-License-Identifier: Apache-2.0 | |
| r"""An abstraction for maintaining job runtime and its units. | |
| """ | |
| from copy import copy | |
| from typing import Tuple | |
| import unittest | |
| class JobTime: | |
| # Possible units. | |
| units = ["h", "m", "s", "ms", "us", "ns", "ps", "fs"] | |
| dividers = [60.0, ] * 3 + [1000.0, ] * 5 | |
| def __init__(self, time: float = 0.0, unit: str = "s", normalize: bool = True): | |
| self.set(time, unit, normalize) | |
| def set(self, time: float, unit: str, normalize: bool = True): | |
| """Public API to set the instance variables time, unit.""" | |
| self.__time = time | |
| self.__unit = unit | |
| assert self.__unit in self.units | |
| if normalize: | |
| self._normalize() | |
| def get(self) -> Tuple[float, str]: | |
| """Returns the time and unit as a tuple.""" | |
| return self.__time, self.__unit | |
| def with_unit(self, unit: str): | |
| """Return a copy of this object that has a specific unit and a value | |
| scaled accordingly. | |
| Note that the scaling may not be lossless due to rounding errors and | |
| limited precision. | |
| """ | |
| target_index = self.units.index(unit) | |
| index = self.units.index(self.__unit) | |
| jt = copy(self) | |
| while index < target_index: | |
| index += 1 | |
| jt.__time *= self.dividers[index] | |
| jt.__unit = self.units[index] | |
| while index > target_index: | |
| jt.__time /= self.dividers[index] | |
| index -= 1 | |
| jt.__unit = self.units[index] | |
| return jt | |
| def _normalize(self): | |
| """Brings the time and its units to a more meaningful magnitude. | |
| If the time is very large with a lower magnitude, this method divides | |
| the time to get it to the next higher magnitude recursively, stopping | |
| if the next division causes the time to go < 1. Examples: | |
| 123123232ps -> 123.12us | |
| 23434s -> 6.509h | |
| The supported magnitudes and their associated divider values are | |
| provided by JobTime.units and JobTime.dividers. | |
| """ | |
| if self.__time == 0: | |
| return | |
| index = self.units.index(self.__unit) | |
| normalized_time = self.__time | |
| while index > 0 and normalized_time >= self.dividers[index]: | |
| normalized_time = normalized_time / self.dividers[index] | |
| index = index - 1 | |
| self.__time = normalized_time | |
| self.__unit = self.units[index] | |
| def __str__(self): | |
| """Indicates <time><unit> as string. | |
| The time value is truncated to 3 decimal places. | |
| Returns an empty string if the __time is set to 0. | |
| """ | |
| if self.__time == 0: | |
| return "" | |
| else: | |
| return f"{self.__time:.3f}{self.__unit}" | |
| def __eq__(self, other) -> bool: | |
| assert isinstance(other, JobTime) | |
| other_time, other_unit = other.get() | |
| return self.__unit == other_unit and self.__time == other_time | |
| def __gt__(self, other) -> bool: | |
| if self.__time == 0: | |
| return False | |
| assert isinstance(other, JobTime) | |
| other_time, other_unit = other.get() | |
| if other_time == 0: | |
| return True | |
| sidx = JobTime.units.index(self.__unit) | |
| oidx = JobTime.units.index(other_unit) | |
| if sidx < oidx: | |
| return True | |
| elif sidx > oidx: | |
| return False | |
| else: | |
| return self.__time > other_time | |
| class TestJobTimeMethods(unittest.TestCase): | |
| def test_with_unit(self): | |
| # First data set | |
| h = JobTime(6, 'h', normalize=False) | |
| m = JobTime(360, 'm', normalize=False) | |
| s = JobTime(21600, 's', normalize=False) | |
| ms = JobTime(21600000, 'ms', normalize=False) | |
| for src in [h, m, s, ms]: | |
| for unit, dst in [('h', h), ('m', m), ('s', s), ('ms', ms)]: | |
| self.assertEqual(src.with_unit(unit), dst) | |
| # Second data set | |
| fs = JobTime(123456000000, 'fs', normalize=False) | |
| ps = JobTime(123456000, 'ps', normalize=False) | |
| ns = JobTime(123456, 'ns', normalize=False) | |
| us = JobTime(123.456, 'us', normalize=False) | |
| ms = JobTime(0.123456, 'ms', normalize=False) | |
| for src in [fs, ps, ns, us, ms]: | |
| for unit, dst in [('fs', fs), ('ps', ps), ('ns', ns), ('us', us), | |
| ('ms', ms)]: | |
| self.assertEqual(src.with_unit(unit), dst) |