Dexcom .csv merge
A simple tool to merge several Dexcom Studio .csv/.txt export files into a single file.
NB: Now being further developed in iPancreas-dexcom.
A simple tool to merge several Dexcom Studio .csv/.txt export files into a single file.
NB: Now being further developed in iPancreas-dexcom.
| # usage: dexcom_merge.py [-h] [-o OUTPUT_FILE] [-p DIR_PATH] | |
| # | |
| # Merge a set of Dexcom .csv exports (from Dexcom Studio) into one .csv file. | |
| # | |
| # optional arguments: | |
| # -h, --help show this help message and exit | |
| # -o OUTPUT_FILE, --output_file OUTPUT_FILE | |
| # path and/or name of output file | |
| # -p DIR_PATH, --path DIR_PATH | |
| # path to the directory where all your Dexcom Studio | |
| # .csv exports are stored | |
| # | |
| # Copyright (c) 2014, Jana E. Beck | |
| # Contact: jana.eliz.beck@gmail.com | |
| # License: GPLv3 (http://www.gnu.org/licenses/gpl-3.0.txt) | |
| from __future__ import print_function | |
| import argparse | |
| import csv | |
| import os | |
| class DexcomSet: | |
| def __init__(self, files): | |
| self.set = set([]) | |
| self.files = files | |
| self.total_so_far = 0 | |
| for f in self.files: | |
| self._add_rows_from_file(f) | |
| def _add_rows_from_file(self, this_file): | |
| """Add the rows from a file to the DexcomSet.""" | |
| count = 0 | |
| missing = 0 | |
| with open(this_file, 'rU') as f: | |
| rdr = csv.reader(f, delimiter='\t') | |
| # exclude header | |
| rdr.next() | |
| for row in rdr: | |
| # set uses a hash internally and hash keys must be immutable types | |
| # i.e., tuple, not list | |
| # also replacing first two cols with '' because want to avoid duplicates with rows that have ID info | |
| row[0] = '' | |
| row[1] = '' | |
| self.set.add(tuple(row)) | |
| count += 1 | |
| print("%i readings in %s." %(count, this_file)) | |
| print("%i items in DexcomSet." %(len(self.set))) | |
| if len(self.set) - self.total_so_far != count: | |
| duplicates = ((self.total_so_far + count) - len(self.set)) | |
| print("%i duplicate records in this file." %(duplicates)) | |
| print() | |
| self.total_so_far = len(self.set) | |
| def _sort(self): | |
| """Sort the DexcomSet by GlucoseInternalTime.""" | |
| return sorted(self.set, key=lambda t: t[2]) | |
| def print_set(self, header, output_file = 'merged-dexcom.csv'): | |
| """Print the DexcomSet row-by-row to file.""" | |
| count = 0 | |
| with open(output_file, 'w') as f: | |
| wrtr = csv.writer(f, delimiter='\t') | |
| print("### Writing to new file %s..." %(output_file)) | |
| print() | |
| wrtr.writerow(header) | |
| for item in self._sort(): | |
| wrtr.writerow(list(item)) | |
| count += 1 | |
| print("%i non-duplicate records printed to %s." %(count, output_file)) | |
| print() | |
| def get_header(this_file): | |
| """Get the header of a Dexcom file.""" | |
| with open(this_file, 'rU') as f: | |
| rdr = csv.reader(f, delimiter='\t') | |
| return rdr.next() | |
| def get_file_list(path = ""): | |
| """Get the list of files with .csv or .txt extensions in the target directory.""" | |
| all_files = [] | |
| for root, dirs, files in os.walk(path): | |
| all_files += [os.path.join(root, f) for f in files if (f.endswith('.txt') or f.endswith('.csv') and not (f.startswith('$') or f.startswith('._')))] | |
| return all_files | |
| def main(): | |
| parser = argparse.ArgumentParser(description='Merge a set of Dexcom .csv exports (from Dexcom Studio) into one .csv file.') | |
| parser.add_argument('-c', '--csv', action= 'store_true', help='comma- (instead of tab-)delimited output') | |
| parser.add_argument('-o', '--output_file', action = 'store', dest = "output_file", help='path and/or name of output file') | |
| parser.add_argument('-p', '--path', action = 'store', dest = "dir_path", help='path to the directory where all your Dexcom Studio .csv exports are stored') | |
| args = parser.parse_args() | |
| try: | |
| files = get_file_list(args.dir_path) | |
| except TypeError: | |
| files = get_file_list() | |
| expected_header = ['PatientInfoField', 'PatientInfoValue', 'GlucoseInternalTime', 'GlucoseDisplayTime', 'GlucoseValue', 'MeterInternalTime', 'MeterDisplayTime', 'MeterValue', 'EventLoggedInternalTime', 'EventLoggedDisplayTime', 'EventTime', 'EventType', 'EventDescription'] | |
| to_remove = [] | |
| for f in files: | |
| if get_header(f) != expected_header: | |
| print() | |
| print("!!! This file doesn't look like a Dexcom file: \n%s \nI'm skipping it. :(" %(f)) | |
| print() | |
| to_remove.append(f) | |
| [files.remove(f) for f in to_remove] | |
| try: | |
| header = get_header(files[0]) | |
| except IndexError: | |
| print() | |
| print("Sorry, I couldn't find any Dexcom files in this directory. Try giving me a path to the directory where you've stored them.") | |
| print() | |
| exit(0) | |
| print('### Merging the following files:') | |
| [print(f) for f in files] | |
| print() | |
| dex = DexcomSet(files) | |
| try: | |
| dex.print_set(header, args.output_file) | |
| except TypeError: | |
| dex.print_set(header) | |
| if __name__ == '__main__': | |
| main() |