[content/test/gpu] Run pylint check of gpu tests in unittest instead of PRESUBMIT

third_party/logilab/logilab/common/dbf.py

+# -*- coding: utf-8 -*-
+# copyright 2003-2011 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
+#
+# This file is part of logilab-common.
+#
+# logilab-common is free software: you can redistribute it and/or modify it under
+# the terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation, either version 2.1 of the License, or (at your option) any
+# later version.
+#
+# logilab-common is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+# FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with logilab-common.  If not, see <http://www.gnu.org/licenses/>.
+"""This is a DBF reader which reads Visual Fox Pro DBF format with Memo field
+
+Usage:
+
+>>> rec = readDbf('test.dbf')
+>>> for line in rec:
+>>>     print line['name']
+
+
+:date: 13/07/2007
+
+http://www.physics.ox.ac.uk/users/santoso/Software.Repository.html
+page says code is "available as is without any warranty or support".
+"""
+from __future__ import print_function
+
+import struct
+import os, os.path
+import sys
+import csv
+import tempfile
+
+from six.moves import range
+
+class Dbase:
+    def __init__(self):
+        self.fdb = None
+        self.fmemo = None
+        self.db_data = None
+        self.memo_data = None
+        self.fields = None
+        self.num_records = 0
+        self.header = None
+        self.memo_file = ''
+        self.memo_header = None
+        self.memo_block_size = 0
+        self.memo_header_len = 0
+
+    def _drop_after_NULL(self, txt):
+        for i in range(0, len(txt)):
+            if ord(struct.unpack('c', txt[i])[0])==0:
+                return txt[:i]
+        return txt
+
+    def _reverse_endian(self, num):
+        if not len(num):
+            return 0
+        val = struct.unpack('<L', num)
+        val = struct.pack('>L', val[0])
+        val = struct.unpack('>L', val)
+        return val[0]
+
+    def _assign_ids(self, lst, ids):
+        result = {}
+        idx = 0
+        for item in lst:
+            id = ids[idx]
+            result[id] = item
+            idx += 1
+        return result
+
+    def open(self, db_name):
+        filesize = os.path.getsize(db_name)
+        if filesize <= 68:
+            raise IOError('The file is not large enough to be a dbf file')
+
+        self.fdb = open(db_name, 'rb')
+
+        self.memo_file = ''
+        if os.path.isfile(db_name[0:-1] + 't'):
+            self.memo_file = db_name[0:-1] + 't'
+        elif os.path.isfile(db_name[0:-3] + 'fpt'):
+            self.memo_file = db_name[0:-3] + 'fpt'
+
+        if self.memo_file:
+            #Read memo file
+            self.fmemo = open(self.memo_file, 'rb')
+            self.memo_data = self.fmemo.read()
+            self.memo_header = self._assign_ids(struct.unpack('>6x1H', self.memo_data[:8]), ['Block size'])
+            block_size = self.memo_header['Block size']
+            if not block_size:
+                block_size = 512
+            self.memo_block_size = block_size
+            self.memo_header_len = block_size
+            memo_size = os.path.getsize(self.memo_file)
+
+        #Start reading data file
+        data = self.fdb.read(32)
+        self.header = self._assign_ids(struct.unpack('<B 3B L 2H 20x', data), ['id', 'Year', 'Month', 'Day', '# of Records', 'Header Size', 'Record Size'])
+        self.header['id'] = hex(self.header['id'])
+
+        self.num_records = self.header['# of Records']
+        data = self.fdb.read(self.header['Header Size']-34)
+        self.fields = {}
+        x = 0
+        header_pattern = '<11s c 4x B B 14x'
+        ids = ['Field Name', 'Field Type', 'Field Length', 'Field Precision']
+        pattern_len = 32
+        for offset in range(0, len(data), 32):
+            if ord(data[offset])==0x0d:
+                break
+            x += 1
+            data_subset = data[offset: offset+pattern_len]
+            if len(data_subset) < pattern_len:
+                data_subset += ' '*(pattern_len-len(data_subset))
+            self.fields[x] = self._assign_ids(struct.unpack(header_pattern, data_subset), ids)
+            self.fields[x]['Field Name'] = self._drop_after_NULL(self.fields[x]['Field Name'])
+
+        self.fdb.read(3)
+        if self.header['# of Records']:
+            data_size = (self.header['# of Records'] * self.header['Record Size']) - 1
+            self.db_data = self.fdb.read(data_size)
+        else:
+            self.db_data = ''
+        self.row_format = '<'
+        self.row_ids = []
+        self.row_len = 0
+        for key in self.fields:
+            field = self.fields[key]
+            self.row_format += '%ds ' % (field['Field Length'])
+            self.row_ids.append(field['Field Name'])
+            self.row_len += field['Field Length']
+
+    def close(self):
+        if self.fdb:
+            self.fdb.close()
+        if self.fmemo:
+            self.fmemo.close()
+
+    def get_numrecords(self):
+        return self.num_records
+
+    def get_record_with_names(self, rec_no):
+        """
+        This function accept record number from 0 to N-1
+        """
+        if rec_no < 0 or rec_no > self.num_records:
+            raise Exception('Unable to extract data outside the range')
+
+        offset = self.header['Record Size'] * rec_no
+        data = self.db_data[offset:offset+self.row_len]
+        record = self._assign_ids(struct.unpack(self.row_format, data), self.row_ids)
+
+        if self.memo_file:
+            for key in self.fields:
+                field = self.fields[key]
+                f_type = field['Field Type']
+                f_name = field['Field Name']
+                c_data = record[f_name]
+
+                if f_type=='M' or f_type=='G' or f_type=='B' or f_type=='P':
+                    c_data = self._reverse_endian(c_data)
+                    if c_data:
+                        record[f_name] = self.read_memo(c_data-1).strip()
+                else:
+                    record[f_name] = c_data.strip()
+        return record
+
+    def read_memo_record(self, num, in_length):
+        """
+        Read the record of given number. The second parameter is the length of
+        the record to read. It can be undefined, meaning read the whole record,
+        and it can be negative, meaning at most the length
+        """
+        if in_length < 0:
+            in_length = -self.memo_block_size
+
+        offset = self.memo_header_len + num * self.memo_block_size
+        self.fmemo.seek(offset)
+        if in_length<0:
+            in_length = -in_length
+        if in_length==0:
+            return ''
+        return self.fmemo.read(in_length)
+
+    def read_memo(self, num):
+        result = ''
+        buffer = self.read_memo_record(num, -1)
+        if len(buffer)<=0:
+            return ''
+        length = struct.unpack('>L', buffer[4:4+4])[0] + 8
+
+        block_size = self.memo_block_size
+        if length < block_size:
+            return buffer[8:length]
+        rest_length = length - block_size
+        rest_data = self.read_memo_record(num+1, rest_length)
+        if len(rest_data)<=0:
+            return ''
+        return buffer[8:] + rest_data
+
+def readDbf(filename):
+    """
+    Read the DBF file specified by the filename and
+    return the records as a list of dictionary.
+
+    :param: filename File name of the DBF
+    :return: List of rows
+    """
+    db = Dbase()
+    db.open(filename)
+    num = db.get_numrecords()
+    rec = []
+    for i in range(0, num):
+        record = db.get_record_with_names(i)
+        rec.append(record)
+    db.close()
+    return  rec
+
+if __name__=='__main__':
+    rec = readDbf('dbf/sptable.dbf')
+    for line in rec:
+        print('%s %s' % (line['GENUS'].strip(), line['SPECIES'].strip()))