tinybufr 0.1.2

A decoder for BUFR meteorological data format
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154

# /// script
# requires-python = ">=3.13"
# dependencies = [
#     "pandas",
# ]
# ///

import subprocess

import pandas as pd

TABLE_B_HEAD = """
//! This file is generated from BUFRCREX_TableB_en.txt.

use super::{XY, TableBEntry};
"""

TABLE_C_HEAD = """
//! This file is generated from BUFR_TableC_en.txt.

use super::TableCEntry;
"""

TABLE_D_HEAD = """
//! This file is generated from BUFR_TableD_en.txt.

use super::{Descriptor, XY, TableDEntry};
"""


def escape(s: str | float) -> str:
    if isinstance(s, str):
        return s.replace('"', '\\"')
    elif isinstance(s, float):
        return ""
    else:
        raise ValueError(f"Unknown type: {type(s)}")


def make_table_b() -> None:
    df = pd.read_csv("./BUFR4/txt/BUFRCREX_TableB_en.txt")
    with open("./src/tables/table_b.rs", "w") as f:
        f.write(TABLE_B_HEAD)
        f.write("\n")
        f.write(f"pub static TABLE_B: [TableBEntry; {len(df)}] = [\n")
        for _, row in df.iterrows():
            if row["Status"] == "Deprecated":  # type: ignore
                continue
            fxy = row["FXY"]  # type: ignore
            assert fxy < 1000000  # type: ignore
            x = (fxy % 100000) // 1000
            y = fxy % 1000
            f.write("TableBEntry {\n")
            f.write(f"xy: XY {{ x: {x}, y: {y} }},\n")
            f.write(f'class_name: "{row["ClassName_en"]}",\n')
            f.write(f'element_name: "{escape(row["ElementName_en"])}",\n')  # type: ignore
            f.write(f"scale: {row['BUFR_Scale']},\n")
            f.write(f"reference_value: {row['BUFR_ReferenceValue']},\n")
            unit = row["BUFR_Unit"]  # type: ignore
            bits = row["BUFR_DataWidth_Bits"]  # type: ignore
            if bits >= 33:  # type: ignore
                assert unit == "CCITT IA5"  # type: ignore
            f.write(f'unit: "{unit}",\n')
            f.write(f"bits: {bits},\n")
            f.write("},\n")
        f.write("];")


def make_table_c() -> None:
    df = pd.read_csv("./BUFR4/txt/BUFR_TableC_en.txt")
    with open("./src/tables/table_c.rs", "w") as f:
        f.write(TABLE_C_HEAD)
        f.write("\n")
        f.write(f"pub static TABLE_C: [TableCEntry; {len(df)}] = [\n")
        for _, row in df.iterrows():
            if row["Status"] == "Deprecated":  # type: ignore
                continue
            fxy = row["FXY"]  # type: ignore
            assert fxy[0] == "2"  # type: ignore
            x = int(str(fxy)[:3]) % 100  # type: ignore
            if isinstance(fxy, str) and fxy.endswith("YYY"):
                y = None
            else:
                y = int(fxy) % 1000  # type: ignore
            f.write("TableCEntry {\n")
            if y is None:
                f.write(f"xy: ({x}, {None}),\n")
            else:
                f.write(f"xy: ({x}, Some({y})),\n")
            f.write(f'operator_name: "{row["OperatorName_en"]}",\n')
            f.write(f'operation_definition: "{row["OperationDefinition_en"]}",\n')
            f.write("},\n")
        f.write("];")


def make_table_d() -> None:
    df = pd.read_csv("./BUFR4/txt/BUFR_TableD_en.txt")
    count = 0
    previous_fxy = None
    closed = set()
    with open("./src/tables/table_d.rs", "w") as f:
        f.write(TABLE_D_HEAD)
        f.write("\n")

        size = df[df["Status"] != "Deprecated"].FXY1.nunique()  # type: ignore

        f.write(f"pub static TABLE_D: [TableDEntry; {size}] = [\n")
        for _, row in df.iterrows():
            if row["Status"] == "Deprecated":  # type: ignore
                continue
            fxy = row["FXY1"]  # type: ignore
            if fxy == previous_fxy:  # type: ignore
                count += 1
            elif previous_fxy is not None:
                assert previous_fxy not in closed
                closed.add(previous_fxy)
                count = 0
                f.write("],\n")
                f.write("},\n")
            previous_fxy = fxy
            x1 = (fxy % 100000) // 1000
            y1 = fxy % 1000
            fxy2 = row["FXY2"]  # type: ignore
            f2 = fxy2 // 100000
            x2 = (fxy2 % 100000) // 1000
            y2 = fxy2 % 1000
            if count == 0:
                f.write("TableDEntry {\n")
                f.write(f"xy: XY {{ x: {x1}, y: {y1} }},\n")
                f.write(f'category: "{escape(row["CategoryOfSequences_en"])}",\n')  # type: ignore
                f.write(
                    f'title: "{escape(row["Title_en"]).removeprefix("(").removesuffix(")")}",\n'  # type: ignore
                )
                f.write(f'sub_title: "{escape(row["SubTitle_en"])}",\n')  # type: ignore
                f.write("elements: &[\n")
            f.write(f"Descriptor {{ f: {f2}, x: {x2}, y: {y2} }},\n")

        f.write("],\n")
        if count > 0:
            f.write("},\n")
            f.write("];\n")


def main():
    print("generating...")
    make_table_b()
    make_table_c()
    make_table_d()
    print("cargo fmt...")
    subprocess.run(["cargo", "fmt"], check=True)


if __name__ == "__main__":
    main()