The TableNormalizer Class

This example describes the functionality of the TableNormalizer class, and is aimed at those wishing to create a custom normalizer or contribute to the project.

import bdf
import polars as pl
from bdf.table_normalizers import TableNormalizer, Syn, DateTimeSyn, ResolvedColumn

A TableNormalizer class defines the mapping between bdf-defined columns and names that may appear as column headings for your data. You can initialise a TableNormalizer by providing synonyms or mappings. We will load a simple example with time, voltage and current:

df = pl.DataFrame(
    {
        "Test-Time": ["00:00:00.00", "00:00:01.00", "00:00:02.00"],
        "Voltage-V": [3.2, 3.3, 3.4],
        "Current-mA": [10.0, 10.0, 10.0],
    }
)

To define synonyms, use the Syn and DateTimeSyn classes. Syn provides a simple mapping for numeric columns. Its sole argument is a string that will be used to match column names in the data. the {unit} placeholder identifies the position of the unit in the column name, and will match any valid unit that the pint library can parse.

DateTimeSyn allows you to define a set of candidate formats that can be used to parse datetime string columns.

normalizer = TableNormalizer(
    test_time_second=[DateTimeSyn(syn=Syn(hdr="Test-Time"), fmts=("%H:%M:%S.%f",))],
    voltage_volt=[Syn(hdr="Voltage-{unit}")],
    current_ampere=[Syn(hdr="Current-{unit}")],
)

Scoring against the normalizer

The TableNormalizer.score_columns() is used to evaluate how well a defined TableNormalizer matches the data. This is used for auto-detection of sources of data. Every column parsable by the synonyms will add to the score_columns.

score_columns = normalizer.score_columns(df.columns)
print("Headers:", df.columns)
print("Score:", score_columns)

Headers: ['Test-Time', 'Voltage-V', 'Current-mA']
Score: 3

The score_columns will drop when we provide columns that do not match the synonyms in the TableNormalizer. For example, making the unit unparseable:

import copy

cols = copy.deepcopy(df.columns)
cols.remove("Current-mA")
cols.append("Current-notamps")
print("Headers:", cols)
print("Score:", normalizer.score_columns(cols))

Headers: ['Test-Time', 'Voltage-V', 'Current-notamps']
Score: 2

Or using a unit that is unreachable from the bdf base unit (e.g. amperes from volts):

cols = copy.deepcopy(df.columns)
cols.remove("Current-mA")
cols.append("Current-V")
print("Headers:", cols)
print("Score:", normalizer.score_columns(cols))

Headers: ['Test-Time', 'Voltage-V', 'Current-V']
Score: 2

Or having a different variant of the quantity in the data:

cols = copy.deepcopy(df.columns)
cols.remove("Current-mA")
cols.append("Amps-mA")
print("Headers:", cols)
print("Score:", normalizer.score_columns(cols))

Headers: ['Test-Time', 'Voltage-V', 'Amps-mA']
Score: 2

However, multiple synonyms can be provided to a single field in the TableNormalizer, allowing the same normalizer to match a range of data even if their headings differ slightly:

normalizer = TableNormalizer(
    test_time_second=[DateTimeSyn(syn=Syn(hdr="Test-Time"), fmts=("%H:%M:%S.%f",))],
    voltage_volt=[Syn(hdr="Voltage-{unit}")],
    current_ampere=[Syn(hdr="Current-{unit}"), Syn(hdr="Amps-{unit}")],
)
print("Headers:", cols)
print("Score:", normalizer.score_columns(cols))

Headers: ['Test-Time', 'Voltage-V', 'Amps-mA']
Score: 3

Resolving column names and unit conversions

When you define a synonym, the TableNormalizer.resolve() method performs the name matching and any unit conversion. It returns a dictionary of the matched bdf columns and defines the operations needed to perform the normalization in a bdf.ResolvedColumn instance. You can see for the current column that the scale attribute reflects the mA unit.

import pprint

pprint.pprint(normalizer.resolve(df.columns))

{'current_ampere': ResolvedColumn(source_header='Current-mA', scale=0.001, offset=0.0, datetime_fmts=(), legacy=False),
 'test_time_second': ResolvedColumn(source_header='Test-Time', scale=1.0, offset=0.0, datetime_fmts=('%H:%M:%S.%f',), legacy=False),
 'voltage_volt': ResolvedColumn(source_header='Voltage-V', scale=1.0, offset=0.0, datetime_fmts=(), legacy=False)}

You can avoid auto-detection entirely by using the ResolvedColumn class in place of synonyms in your TableNormalizer.

resolved_normalizer = TableNormalizer(
    test_time_second=[DateTimeSyn(syn=Syn(hdr="Test-Time"), fmts=("%H:%M:%S.%f",))],
    voltage_volt=[Syn(hdr="Voltage-{unit}")],
    current_ampere=ResolvedColumn(source_header="Current-mA", scale=0.001),
)
print("Headers:", df.columns)
print("Score:", resolved_normalizer.score_columns(df.columns))

Headers: ['Test-Time', 'Voltage-V', 'Current-mA']
Score: 3

Returning the normalized data

The TableNormalizer.normalize() method will normalize the columns into the correct unit, returning a bdf-compliant dataframe:

normalizer.normalize(df)

shape: (3, 3)

Test Time / s	Voltage / V	Current / A
f64	f64	f64
0.0	3.2	0.01
1.0	3.3	0.01
2.0	3.4	0.01

Initialising a TableNormalizer from a schema file

Since TableNormalizer is a Pydantic model, it can be constructed from a plain dict (e.g. loaded from a JSON config file). Each field accepts either a list of synonym strings / DateTimeSyn dicts, or a ResolvedColumn dict.

json_normalizer = TableNormalizer.model_validate(
    {
        "test_time_second": [{"syn": "Test-Time", "fmts": ["%H:%M:%S.%f"]}],
        "voltage_volt": ["Voltage-{unit}"],
        "current_ampere": ["Current-{unit}"],
    }
)
print("Score:", json_normalizer.score_columns(df.columns))

Score: 3