Break up parsing and writing to file, typehints, clarifying variable names

Author: mrhheffernan

fit_to_csv.py

@@ -3,11 +3,12 @@
 import glob
 import os
 from datetime import timezone
+from typing import Any
 from zoneinfo import ZoneInfo
 
 import fitparse
 
-allowed_fields = [
+FIELDS_ALLOWED = [
     "timestamp",
     "position_lat",
     "position_long",
@@ -21,43 +22,57 @@
     "cadence",
     "fractional_cadence",
 ]
-required_fields = ["timestamp", "position_lat", "position_long"]
+FIELDS_REQUIRED = ["timestamp", "position_lat", "position_long"]
 
 UTC = timezone.utc
 TZ = ZoneInfo("US/Central")
 
 
-def write_fitfile_to_csv(fitfile, output_path: str, tz: ZoneInfo = TZ):
-    messages = fitfile.messages
+def write_to_csv(data: list[dict[str, Any]], output_path: str) -> None:
+    # write to csv
+    with open(output_path, "w") as f:
+        writer = csv.writer(f)
+        writer.writerow(FIELDS_ALLOWED)
+        for entry in data:
+            writer.writerow([str(entry.get(k, "")) for k in FIELDS_ALLOWED])
+    print("wrote %s" % output_path)
+
+
+def collect_data(filepath: str, tz: ZoneInfo = TZ) -> list[dict[str, Any]]:
+    # Parse the .fit file
+    fitfile = fitparse.FitFile(
+        filepath, data_processor=fitparse.StandardUnitsDataProcessor()
+    )
+
     data = []
+    messages = fitfile.messages
+
     for m in messages:
         skip = False
         if not hasattr(m, "fields"):
             continue
         fields = m.fields
-        # check for important data types
+
+        # check for desired data and collect it
         mdata = {}
         for field in fields:
-            if field.name in allowed_fields:
+            if field.name in FIELDS_ALLOWED:
                 if field.name == "timestamp":
-                    ts_value = field.value
-                    if ts_value.tzinfo is None:
-                        ts_value = ts_value.replace(tzinfo=UTC)
-                    mdata[field.name] = ts_value.astimezone(tz)
+                    timestamp_value = field.value
+                    if timestamp_value.tzinfo is None:
+                        timestamp_value = timestamp_value.replace(tzinfo=UTC)
+                    mdata[field.name] = timestamp_value.astimezone(tz)
                 else:
                     mdata[field.name] = field.value
-        for rf in required_fields:
-            if rf not in mdata:
+
+        for required_field in FIELDS_REQUIRED:
+            if required_field not in mdata:
                 skip = True
+
         if not skip:
             data.append(mdata)
-    # write to csv
-    with open(output_path, "w") as f:
-        writer = csv.writer(f)
-        writer.writerow(allowed_fields)
-        for entry in data:
-            writer.writerow([str(entry.get(k, "")) for k in allowed_fields])
-    print("wrote %s" % output_path)
+
+    return data
 
 
 def parse_args() -> argparse.Namespace:
@@ -86,19 +101,20 @@ def parse_args() -> argparse.Namespace:
 def main():
     args = parse_args()
 
+    # Identify .fit files
     fit_files = glob.glob(args.dir + "/*.fit")
+
     for file in fit_files:
+        # Use the same filename, just change extension to .csv
         base_filename = file.removesuffix(".fit")
         new_filename = base_filename + ".csv"
         if not args.overwrite and os.path.exists(new_filename):
-            # print('%s already exists. skipping.' % new_filename)
             continue
-        fitfile = fitparse.FitFile(
-            file, data_processor=fitparse.StandardUnitsDataProcessor()
-        )
 
         print("converting %s" % file)
-        write_fitfile_to_csv(fitfile, new_filename, tz=ZoneInfo(args.timezone))
+        data = collect_data(file, tz=ZoneInfo(args.timezone))
+        write_to_csv(data, new_filename)
+
     print("finished conversions")