readabs.recalibrate

  1"""Recalibrate a Series or DataFrame so the data is in the range -1000 to 1000."""
  2
  3import sys
  4from collections.abc import Callable
  5from operator import mul, truediv
  6from typing import Any
  7
  8import numpy as np
  9from pandas import DataFrame, Series
 10
 11from readabs.datatype import Datatype as DataT
 12
 13# Constants
 14NDIM_SERIES = 1
 15NDIM_DATAFRAME = 2
 16MAX_VALUE_THRESHOLD = 1000
 17MIN_VALUE_THRESHOLD = 1
 18STEP_SIZE = 3
 19DIVISOR = 1000
 20
 21
 22# --- public
 23def recalibrate(
 24    data: DataT,
 25    units: str,
 26) -> tuple[DataT, str]:
 27    """Recalibrate a Series or DataFrame so the data is in the range -1000 to 1000.
 28
 29    Change the name of the units to reflect the recalibration.
 30
 31    Note, DataT = TypeVar("DataT", Series, DataFrame). DataT is a constrained typevar.
 32    If you provide a Series, you will get a Series back. If you provide a DataFrame,
 33    you will get a DataFrame back.
 34
 35    Parameters
 36    ----------
 37    data : Series or DataFrame
 38        The data to recalibrate.
 39    units : str
 40        The units of the data. This string should be in the form of
 41        "Number", "Thousands", "Millions", "Billions", etc. The units
 42        should be in title case.
 43
 44    Returns
 45    -------
 46    Series or DataFrame
 47        The recalibrated data will be a Series if a Series was provided,
 48        or a DataFrame if a DataFrame was provided.
 49
 50    Examples
 51    --------
 52    ```python
 53    from pandas import Series
 54    from readabs import recalibrate
 55    s = Series([1_000, 10_000, 100_000, 1_000_000])
 56    recalibrated, units = recalibrate(s, "$")
 57    print(f"{recalibrated=}, {units=}")
 58    ```
 59
 60    """
 61    if not isinstance(data, (Series, DataFrame)):
 62        raise TypeError("data must be a Series or DataFrame")
 63    units, restore_name = _prepare_units(units)
 64    flat_data = data.to_numpy().flatten()
 65    flat_data, units = _recalibrate(flat_data, units)
 66
 67    if restore_name:
 68        units = f"{restore_name} {units}"
 69        for n in "numbers", "number":
 70            if n in units:
 71                units = units.replace(n, "").strip()
 72                break
 73    units = units.title()
 74
 75    result = data.__class__(flat_data.reshape(data.shape))
 76    result.index = data.index
 77    # restore the column labels (DataFrame) or series name (Series); the
 78    # isinstance checks narrow the constrained TypeVar so the attributes type-check
 79    if isinstance(data, DataFrame) and isinstance(result, DataFrame):
 80        result.columns = data.columns
 81    elif isinstance(data, Series) and isinstance(result, Series):
 82        result.name = data.name
 83    return result, units
 84
 85
 86def recalibrate_value(value: float, units: str) -> tuple[float, str]:
 87    """Recalibrate a floating point value.
 88
 89    The value will be recalibrated so it is in the range -1000 to 1000.
 90    The units will be changed to reflect the recalibration.
 91
 92    Parameters
 93    ----------
 94    value : float
 95        The value to recalibrate.
 96    units : str
 97        The units of the value. This string should be in the form of
 98        "Number", "Thousands", "Millions", "Billions", etc. The units
 99        should be in title case.
100
101    Returns
102    -------
103    tuple[float, str]
104        A tuple containing the recalibrated value and the recalibrated units.
105
106    Examples
107    --------
108    ```python
109    from readabs import recalibrate_value
110    recalibrated, units = recalibrate_value(10_000_000, "Thousand")
111    print(recalibrated, units)
112    ```
113
114    """
115    series = Series([value])
116    output, units = recalibrate(series, units)
117    return output.to_numpy()[0], units
118
119
120# --- private
121_MIN_RECALIBRATE = "number"  # all lower case
122_MAX_RECALIBRATE = "decillion"  # all lower case
123_keywords = {
124    _MIN_RECALIBRATE.title(): 0,
125    "Thousand": 3,
126    "Million": 6,
127    "Billion": 9,
128    "Trillion": 12,
129    "Quadrillion": 15,
130    "Quintillion": 18,
131    "Sextillion": 21,
132    "Septillion": 24,
133    "Octillion": 27,
134    "Nonillion": 30,
135    _MAX_RECALIBRATE.title(): 33,
136}
137_r_keywords = {v: k for k, v in _keywords.items()}
138
139
140def _prepare_units(units: str) -> tuple[str, str]:
141    """Prepare the units for recalibration."""
142    substitutions = [
143        ("000 Hours", "Thousand Hours"),
144        ("$'000,000", "$ Million"),
145        ("$'000", " $ Thousand"),
146        ("'000,000", "Millions"),
147        ("'000", "Thousands"),
148        ("000,000", "Millions"),
149        ("000", "Thousands"),
150    ]
151    units = units.strip()
152    for pattern, replacement in substitutions:
153        units = units.replace(pattern, replacement)
154
155    # manage the names for some gnarly units
156    possible_units = ("$", "Tonnes")  # there may be more possible units
157    found_unit = ""
158    for pu in possible_units:
159        if pu.lower() in units.lower():
160            units = units.lower().replace(pu.lower(), "").strip()
161            if units == "":
162                units = "number"
163            found_unit = pu
164            break
165
166    return units, found_unit
167
168
169def _find_calibration(units: str) -> str | None:
170    found = None
171    for keyword in _keywords:
172        if keyword in units or keyword.lower() in units:
173            found = keyword
174            break
175    return found
176
177
178# private
179def _perfect_already(data: np.ndarray) -> bool:
180    """No need to recalibrate if the data is already perfect."""
181    check_max = np.nanmax(np.abs(data))
182    return bool(MIN_VALUE_THRESHOLD <= check_max < MAX_VALUE_THRESHOLD)
183
184
185def _all_zero(data: np.ndarray) -> bool:
186    """Cannot recalibrate if all the data is zero."""
187    if np.nanmax(np.abs(data)) == 0:
188        print("recalibrate(): All zero data")
189        return True
190    return False
191
192
193def _not_numbers(data: np.ndarray) -> bool:
194    """Cannot recalibrate if the data is not numeric."""
195    if (not np.issubdtype(data.dtype, np.number)) or np.isinf(data).any() or np.isnan(data).all():
196        print("recalibrate(): Data is partly or completely non-numeric.")
197        return True
198    return False
199
200
201def _can_recalibrate(flat_data: np.ndarray, units: str) -> bool:
202    """Check if the data can be recalibrated."""
203    if _find_calibration(units) is None:
204        print(f"recalibrate(): Units not appropriately calibrated: {units}")
205        return False
206
207    return all(not f(flat_data) for f in (_not_numbers, _all_zero, _perfect_already))
208
209
210def _recalibrate(flat_data: np.ndarray, units: str) -> tuple[np.ndarray, str]:
211    """Recalibrate the data.
212
213    Loop over the data until its maximum value is between -1000 and 1000.
214    """
215    if _can_recalibrate(flat_data, units):
216        while True:
217            maximum = np.nanmax(np.abs(flat_data))
218            if maximum >= MAX_VALUE_THRESHOLD:
219                if _MAX_RECALIBRATE in units.lower():
220                    print("recalibrate() is not designed for very big units")
221                    break
222                flat_data, units = _do_recal(flat_data, units, STEP_SIZE, truediv)
223                continue
224            if maximum < 1:
225                if _MIN_RECALIBRATE in units.lower():
226                    print("recalibrate() is not designed for very small units")
227                    break
228                flat_data, units = _do_recal(flat_data, units, -STEP_SIZE, mul)
229                continue
230            break
231    return flat_data, units
232
233
234def _do_recal(
235    flat_data: np.ndarray, units: str, step: int, operator: Callable[[np.ndarray, int], np.ndarray]
236) -> tuple[np.ndarray, str]:
237    calibration = _find_calibration(units)
238    if calibration is None:
239        raise ValueError(f"No calibration found for units: {units}")
240    factor = _keywords[calibration]
241    if factor + step not in _r_keywords:
242        print(f"Unexpected factor: {factor + step}")
243        sys.exit(-1)
244    replacement = _r_keywords[factor + step]
245    units = units.replace(calibration, replacement)
246    units = units.replace(calibration.lower(), replacement)
247    flat_data = operator(flat_data, DIVISOR)
248    return flat_data, units
249
250
251# --- test
252if __name__ == "__main__":
253
254    def test_example() -> None:
255        """Test the example in the docstring."""
256        s = Series([1_000, 10_000, 100_000, 1_000_000])
257        recalibrated, units = recalibrate(s, "$")
258        print(f"{recalibrated=}, {units=}")
259
260        recalibrated_val, units_val = recalibrate_value(10_000_000, "Thousand")
261        print(f"{recalibrated_val=}, {units_val=}")
262        print("=" * 40)
263
264    test_example()
265
266    def test_recalibrate() -> None:
267        """Test the recalibrate() function."""
268
269        def run_test(dataset: tuple[tuple[list[Any], str], ...]) -> None:
270            for d, u in dataset:
271                data: Series[Any] = Series(d)
272                recalibrated, units = recalibrate(data, u)
273                print(f"{data.to_numpy()}, {u} ==> {recalibrated.to_numpy()}, {units}")
274                print("=" * 40)
275
276        # good examples
277        good = (
278            ([1, 2, 3, 4, 5], "Number"),  # no change
279            ([1_000, 10_000, 100_000, 1_000_000], "$"),
280            ([1_000, 10_000, 100_000, 1_000_000], "Number Spiders"),
281            ([1_000, 10_000, 100_000, 1_000_000], "Thousand"),
282            ([0.2, 0.3], "Thousands"),
283            ([0.000_000_2, 0.000_000_3], "Trillion"),
284        )
285        run_test(good)
286
287        # bad sets of data - should produce error messages and do nothing
288        bad = (
289            ([1, 2, 3, 4, 5], "Hundreds"),
290            ([0, 0, 0], "Thousands"),
291            ([np.nan, 0, 0], "Thousands"),
292            ([np.inf, 1, 2], "Thousands"),
293            ([0, 0, "a"], "Thousands"),
294        )
295        run_test(bad)
296
297    test_recalibrate()
298
299    def test_recalibrate_value() -> None:
300        """Test the recalibrate_value() function."""
301        # good example
302        recalibrated, units = recalibrate_value(10_000_000, "Thousand")
303        print(recalibrated, units)
304        print("=" * 40)
305
306        # bad example
307        recalibrated, units = recalibrate_value(3_900, "Spiders")
308        print(recalibrated, units)
309        print("=" * 40)
310
311    test_recalibrate_value()