A simple UCR data parser

readme.md

Uniform Crime Report Parser

The FBI's Uniform Crime Report (UCR) MASTER data is stored in a fixed-width format which is not a standard format. This parser allows for conversion to CSV (tabular) data to allow for easier analysis.

Prerequisites

• Python3 (tested with 3.6-3.8) with Pandas in your environment

Usage

After creating BOTH a header.csv and record.csv to identify the field type, the parser can run.

The following example is for the Arrest (ASR) MASTER data-set with the sample headers and records CSVs in this GIST:

python parser.py --headercsv sample_header_columns.csv --detailcsv sample_detail_columns.csv --input 2018_ASR1MON_NATIONAL_MASTER_FILE.txt

Header and Record creation

Each UCR data set contains a .doc file or .pdf file that lists the structure of the included fixed-width txt source file. This parser expects you to create two CSVs that have UNIQUE names and the start and stop column location of the specific record being converted. Since 0-based indexes are being used, be sure to subtract 1 from the start location. The resulting 2 CSVs should be separated based on the record type, which is indicated in the documentation by 'HEADER' and 'DETAIL'. Review the attached PDF for an example and compare the PDF 'position' and 'description' with the values in the UCR zip.

parser.py

import csv
import os
import sys
import logging
import argparse
import pandas as pd
import fileinput

LOGLEVEL = os.environ.get('LOGLEVEL', 'WARNING').upper()
logging.basicConfig(level=LOGLEVEL)

class RecordParser(object):
    def __init__(self, header_csv, record_csv, **kwargs):
        self.primary_list = []
        self.header_data = {}
        self.header_data_records = self._set_colspecs(header_csv)
        self.detail_data = {}
        self.detail_data_records = self._set_colspecs(record_csv)
        self.last_header_row = 0
        self.current_file = None
        
    def _get_fixed_data(self, line, header):
        item_data = {}
        for col_name, (start_col, end_col) in header:
            item_data[col_name] = line[start_col:end_col]
        
        return item_data
    
    def _set_colspecs(self, file_name):
        name = []
        colspecs = []
        
        df = pd.read_csv(file_name)
        
        for index, row in df.iterrows():
            name.append(row[0])
            colspecs.append((row[1], row[2]))
        
        return list(zip(name, colspecs))
    
    def _process_row(self, line):
        if line[22:25] == "000":
            if (self.last_header_row == (self.current_file.filelineno() - 1) and not self.current_file.isfirstline()):
                logging.info(f"Found empty header at {self.current_file.filelineno()}")
                self.primary_list.append(self.header_data.copy())
            self.last_header_row = self.current_file.filelineno()
            self.header_data.clear()
            self.header_data = self._get_fixed_data(line, self.header_data_records)
            logging.info(f"Header data set to: {self.header_data}")
            return self.header_data
        else:
            line_object = {**self.header_data, **self._get_fixed_data(line, self.detail_data_records)}
            self.primary_list.append(line_object)
            logging.info(f"Line data set to: {line_object}")
            return self.primary_list
    
    def parse_file(self, file_name):
        with fileinput.input(file_name) as file:
            self.current_file = file
            for line in file:
                self._process_row(line)
                logging.debug(f"Parsed line #{file.filelineno()} : {line}")
    
    def write_csv(self, outfile="output.csv", newline=""):
        with open(outfile, "w", newline=newline) as csv_file:
            cols = sorted(set(key for key, value in (self.detail_data_records + self.header_data_records)))
            writer = csv.DictWriter(csv_file,
                            fieldnames=cols,
                            restval='',
                            extrasaction='ignore')
            
            logging.info("Writing output file.")
            writer.writeheader()
            writer.writerows(self.primary_list)
    
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("-v", "--verbose", help="Print detailed parsing information.",
                        action="store_true")
    parser.add_argument("-l", "--headercsv", help="The path to the CSV file containing the header line record format.")
    parser.add_argument("-d", "--detailcsv", help="The path to the CSV file containing the detail line record format.")
    parser.add_argument("-i", "--input", help="The path to the TXT file containing the UCR data file to process.")
    args = parser.parse_args()
    
    if args.verbose:
        LOGLEVEL = "DEBUG"
    
    record = RecordParser(args.headercsv, args.detailcsv)
    record.parse_file(args.input)
    record.write_csv()
    

sample_detail_columns.csv

sample_header_columns.csv