RFCT move crc computation to records.Crc, add test

Author: xmedeko

fitparse/base.py

@@ -12,18 +12,19 @@
 from fitparse.processors import FitFileDataProcessor
 from fitparse.profile import FIELD_TYPE_TIMESTAMP, MESSAGE_TYPES
 from fitparse.records import (
-    DataMessage, FieldData, FieldDefinition, DevFieldDefinition, DefinitionMessage, MessageHeader,
-    BASE_TYPES, BASE_TYPE_BYTE, DevField,
+    Crc, DataMessage, FieldData, FieldDefinition, DevFieldDefinition, DefinitionMessage, MessageHeader,
+    BASE_TYPES, BASE_TYPE_BYTE,
     add_dev_data_id, add_dev_field_description, get_dev_type
 )
-from fitparse.utils import calc_crc, fileish_open, FitParseError, FitEOFError, FitCRCError, FitHeaderError
+from fitparse.utils import fileish_open, FitParseError, FitEOFError, FitCRCError, FitHeaderError
 
 
 class FitFile(object):
     def __init__(self, fileish, check_crc=True, data_processor=None):
         self._file = fileish_open(fileish, 'rb')
 
         self.check_crc = check_crc
+        self._crc = None
         self._processor = data_processor or FitFileDataProcessor()
 
         # Get total filesize
@@ -55,34 +56,40 @@ def _read(self, size):
         if size <= 0:
             return None
         data = self._file.read(size)
-        self._crc = calc_crc(data, self._crc)
+        if size != len(data):
+            raise FitEOFError("Tried to read %d bytes from .FIT file but got %d" % (size, len(data)))
+
+        if self.check_crc:
+            self._crc.update(data)
         self._bytes_left -= len(data)
         return data
 
     def _read_struct(self, fmt, endian='<', data=None, always_tuple=False):
-        fmt_with_endian = "%s%s" % (endian, fmt)
+        if fmt.startswith('<') or fmt.startswith('>'):
+            # fmt contains endian
+            fmt_with_endian = fmt
+        else:
+            fmt_with_endian = "%s%s" % (endian, fmt)
         size = struct.calcsize(fmt_with_endian)
         if size <= 0:
             raise FitParseError("Invalid struct format: %s" % fmt_with_endian)
 
         if data is None:
             data = self._read(size)
 
-        if size != len(data):
-            raise FitEOFError("Tried to read %d bytes from .FIT file but got %d" % (size, len(data)))
-
         unpacked = struct.unpack(fmt_with_endian, data)
         # Flatten tuple if it's got only one value
         return unpacked if (len(unpacked) > 1) or always_tuple else unpacked[0]
 
     def _read_and_assert_crc(self, allow_zero=False):
         # CRC Calculation is little endian from SDK
-        crc_expected, crc_actual = self._crc, self._read_struct('H')
-
-        if (crc_actual != crc_expected) and not (allow_zero and (crc_actual == 0)):
-            if self.check_crc:
-                raise FitCRCError('CRC Mismatch [expected = 0x%04X, actual = 0x%04X]' % (
-                    crc_expected, crc_actual))
+        crc_computed, crc_read = self._crc.value, self._read_struct(Crc.FMT)
+        if not self.check_crc:
+            return
+        if crc_computed == crc_read or (allow_zero and crc_read == 0):
+            return
+        raise FitCRCError('CRC Mismatch [computed: %s, read: %s]' % (
+            Crc.format(crc_computed), Crc.format(crc_read)))
 
     ##########
     # Private Data Parsing Methods
@@ -94,7 +101,7 @@ def _parse_file_header(self):
         self._bytes_left = -1
         self._complete = False
         self._compressed_ts_accumulator = 0
-        self._crc = 0
+        self._crc = Crc()
         self._local_mesgs = {}
         self._messages = []
 

fitparse/records.py

@@ -4,8 +4,10 @@
 # Python 2 compat
 try:
     int_types = (int, long,)
+    byte_iter = bytearray
 except NameError:
     int_types = (int,)
+    byte_iter = lambda x: x
 
 try:
     from itertools import zip_longest
@@ -331,6 +333,52 @@ def render(self, raw_value):
         return raw_value
 
 
+class Crc(object):
+    """FIT file CRC computation."""
+
+    CRC_TABLE = (
+        0x0000, 0xCC01, 0xD801, 0x1400, 0xF001, 0x3C00, 0x2800, 0xE401,
+        0xA001, 0x6C00, 0x7800, 0xB401, 0x5000, 0x9C01, 0x8801, 0x4400,
+    )
+
+    FMT = '<H'
+
+    def __init__(self, value=0, byte_arr=None):
+        self.value = value
+        if byte_arr:
+            self.update(byte_arr)
+
+    def __repr__(self):
+        return '<%s %s>' % (self.__class__.__name__, self.value or "-")
+
+    def __str__(self):
+        return self.format(self.value)
+
+    def update(self, byte_arr):
+        """Read bytes and update the CRC computed."""
+        if byte_arr:
+            self.value = self.calculate(byte_arr, self.value)
+
+    @staticmethod
+    def format(value):
+        """Format CRC value to string."""
+        return '0x%04X' % value
+
+    @classmethod
+    def calculate(cls, byte_arr, crc=0):
+        """Compute CRC for input bytes."""
+        for byte in byte_iter(byte_arr):
+            # Taken verbatim from FIT SDK docs
+            tmp = cls.CRC_TABLE[crc & 0xF]
+            crc = (crc >> 4) & 0x0FFF
+            crc = crc ^ tmp ^ cls.CRC_TABLE[byte & 0xF]
+
+            tmp = cls.CRC_TABLE[crc & 0xF]
+            crc = (crc >> 4) & 0x0FFF
+            crc = crc ^ tmp ^ cls.CRC_TABLE[(byte >> 4) & 0xF]
+        return crc
+
+
 def parse_string(string):
     try:
         end = string.index(0x00)

fitparse/utils.py

@@ -16,25 +16,6 @@ class FitHeaderError(FitParseError):
     pass
 
 
-CRC_TABLE = (
-    0x0000, 0xCC01, 0xD801, 0x1400, 0xF001, 0x3C00, 0x2800, 0xE401,
-    0xA001, 0x6C00, 0x7800, 0xB401, 0x5000, 0x9C01, 0x8801, 0x4400,
-)
-
-
-def calc_crc(byte_arr, crc=0):
-    for byte in bytearray(byte_arr):
-        # Taken verbatim from FIT SDK docs
-        tmp = CRC_TABLE[crc & 0xF]
-        crc = (crc >> 4) & 0x0FFF
-        crc = crc ^ tmp ^ CRC_TABLE[byte & 0xF]
-
-        tmp = CRC_TABLE[crc & 0xF]
-        crc = (crc >> 4) & 0x0FFF
-        crc = crc ^ tmp ^ CRC_TABLE[(byte >> 4) & 0xF]
-    return crc
-
-
 METHOD_NAME_SCRUBBER = re.compile(r'\W|^(?=\d)')
 UNIT_NAME_TO_FUNC_REPLACEMENTS = (
     ('/', ' per '),

tests/test.py

@@ -2,15 +2,14 @@
 
 import csv
 import datetime
-import io
 import os
 from struct import pack
 import sys
 
 from fitparse import FitFile
 from fitparse.processors import UTC_REFERENCE, StandardUnitsDataProcessor
-from fitparse.records import BASE_TYPES
-from fitparse.utils import calc_crc, FitEOFError, FitCRCError, FitHeaderError
+from fitparse.records import BASE_TYPES, Crc
+from fitparse.utils import FitEOFError, FitCRCError, FitHeaderError
 
 if sys.version_info >= (2, 7):
     import unittest
@@ -65,8 +64,8 @@ def generate_fitfile(data=None, endian='<'):
 
     # Prototcol version 1.0, profile version 1.52
     header = pack('<2BHI4s', 14, 16, 152, len(fit_data), b'.FIT')
-    file_data = header + pack('<H', calc_crc(header)) + fit_data
-    return file_data + pack('<H', calc_crc(file_data))
+    file_data = header + pack(Crc.FMT, Crc.calculate(header)) + fit_data
+    return file_data + pack(Crc.FMT, Crc.calculate(file_data))
 
 
 def secs_to_dt(secs):

tests/test_records.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python
+
+import sys
+
+from fitparse.records import Crc
+
+if sys.version_info >= (2, 7):
+    import unittest
+else:
+    import unittest2 as unittest
+
+
+class RecordsTestCase(unittest.TestCase):
+    def test_crc(self):
+        crc = Crc()
+        self.assertEqual(0, crc.value)
+        crc.update(b'\x0e\x10\x98\x00(\x00\x00\x00.FIT')
+        self.assertEqual(0xace7, crc.value)
+        # 0 must not change the crc
+        crc.update(0)
+        self.assertEqual(0xace7, crc.value)
+
+    def test_crc_format(self):
+        self.assertEqual('0x0000', Crc.format(0))
+        self.assertEqual('0x12AB', Crc.format(0x12AB))
+
+
+if __name__ == '__main__':
+    unittest.main()