In [19]:

    

import numpy as np
import scipy.stats
import pandas as pd
from collections import OrderedDict
import matplotlib.mlab as mlab
import matplotlib.pyplot as plt
#import matplotlib.table as tbl
import matplotlib.gridspec as gridspec
import matplotlib.ticker as tkr
%matplotlib inline


    

In [2]:

    

def unbias_const(constant, observation):
    '''Moving range unbiasing constant depends on distance between range points. Tabular look up between 0 and 25, calculated between 26 and 100
    ref: http://support.minitab.com/en-us/minitab-express/1/help-and-how-to/control-charts/how-to/variables-data-in-subgroups/xbar-r-chart/methods-and-formulas/unbiasing-constants-d2-d3-and-d4/
    '''
    d2 = [1.128, 1.693, 2.059, 2.326, 2.534, 2.704, 2.847, 2.97, 3.078, 3.173, 3.258, 3.336, 3.407, 3.472, 3.532, 3.588, 3.64, 3.689, 3.735, 3.778, 3.819, 3.858, 3.895, 3.931, 3.964, 3.997, 4.027, 4.057, 4.086, 4.113, 4.139, 4.165, 4.189, 4.213, 4.236, 4.259, 4.28, 4.301, 4.322, 4.341, 4.361, 4.379, 4.398, 4.415, 4.433, 4.45, 4.466, 4.482, 4.498]
    d3 = [0.8525, 0.8884, 0.8798, 0.8641, 0.848, 0.8332, 0.8198, 0.8078, 0.7971, 0.7873, 0.7785, 0.7704, 0.763, 0.7562, 0.7499, 0.7441, 0.7386, 0.7335, 0.7287, 0.7242, 0.7199, 0.7159, 0.7121, 0.7084]
    d4 = [0.954, 1.588, 1.978, 2.257, 2.472, 2.645, 2.791, 2.915, 3.024, 3.121, 3.207, 3.285, 3.356, 3.422, 3.482, 3.538, 3.591, 3.64, 3.686, 3.73, 3.771, 3.811, 3.847, 3.883]
    
    assert constant in ['d2', 'd3', 'd4']
    assert int(observation) <= 100 or int(observation) >= 2
        #raise ValueError('Observation outside of predictable range')
    
    if constant == 'd2' and observation <= 50:
        return d2[observation - 2]
    elif constant == 'd2' and observation > 50:
        return 3.4873 + 0.0250141 * observation - 0.00009823 * observation **2
    if constant == 'd3' and observation <= 25:
        return d3[observation - 2]
    elif constant == 'd3' and observation > 25:
        return 0.80818 - 0.0051871 * observation + 0.00005098 * observation ** 2 - 0.00000019 * observation ** 3
    if constant == 'd4' and observation <= 25:
        return d4[observation - 2]
    elif constant == 'd4' and observation >25:
        return 2.88606 + 0.051313 * observation -0.00049243 * observation ** 2 + 0.00000188 * observation ** 3
    
    '''
    To do: review source material for calculation
    D. J. Wheeler and D. S. Chambers. (1992). Understanding Statistical Process Control, Second Edition, SPC Press, Inc.
    H. Leon Harter (1960). "Tables of Range and Studentized Range". The Annals of Mathematical Statistics, Vol. 31, No. 4, Institute of Mathematical Statistics, 1122−1147.
    '''


    

In [85]:

    

class ControlChart():
    '''
    Tests for special cause
    ruleset invesigation inspired by Michal Nowikowski <godfryd@gmail.com>
    '''
    tests = {
        0: "1 point > n standard deviations from centerline",
        1: "n points in a row on same side of centerline",
        2: "n points in a row, all increasing or all decreasing",
        3: "n points in a row, alternating up and down",
        4: "n of n+1 > 2 standard deviations from center line on same side",
        5: "n of n+1 > 1 standard deviations from center line on same side",
        6: "n points in a row within 1 standard deviation of centerline on either side",
        7: "n points in a row greater than 1 standard deviation from centerline on either side"
        }
    def __init__(self
                 , data
                 , *args
                 , spec_max = None
                 , spec_min = None
                 #, index = 0
                 , sigma_level = 3
                 , on_same_side = 9
                 , trending = 6
                 , alternating = 14
                 , beyond_2sigma = 2
                 , beyond_1sigma = 4
                 , within_1sigma = 15
                 , outside_1sigma = 8
                 , test_group = None
                 , **kwargs
                ):
        self.data = pd.Series(data)
        self.usl = spec_max
        self.lsl = spec_min
        self.sigma_level = sigma_level
        self.on_same_side  = on_same_side 
        self.trending = trending
        self.alternating = alternating
        self.beyond_2sigma = beyond_2sigma
        self.beyond_1sigma = beyond_1sigma
        self.within_1sigma = within_1sigma
        self.outside_1sigma = outside_1sigma
        self.test_group = test_group
        #self.index = index #amount to shift output dataframe index by
    
    def _test_list(self):
        self.test_list = self.test_group if self.test_group else []
        test_matrix = pd.DataFrame(index = self.data.index)
        for i in self.test_list:
            test_matrix[i] = {
                0: self._test0_run,
                1: self._test1_run,
                2: self._test2_run,
                3: self._test3_run,
                4: self._test4_run,
                5: self._test5_run,
                6: self._test6_run,
                7: self._test7_run
            }[i]()
        self.fail_series = self.data[test_matrix.any(axis = 1)]
        
        '''lists of popular groupings    
        RULES_BASIC = [RULES_1_BEYOND_3SIGMA, 
                       RULES_7_ON_ONE_SIDE]
        RULES_PMI = [RULES_1_BEYOND_3SIGMA,
                     RULES_8_ON_ONE_SIDE]
        RULES_WECO = [RULES_1_BEYOND_3SIGMA,
                      RULES_2_OF_3_BEYOND_2SIGMA, 
                      RULES_4_OF_5_BEYOND_1SIGMA, 
                      RULES_8_ON_ONE_SIDE,
                      RULES_6_TRENDING, RULES_14_UP_DOWN]
        RULES_NELSON = [RULES_1_BEYOND_3SIGMA,
                        RULES_9_ON_ONE_SIDE,
                        RULES_6_TRENDING,
                        RULES_14_UP_DOWN,
                        RULES_2_OF_3_BEYOND_2SIGMA,
                        RULES_4_OF_5_BEYOND_1SIGMA,
                        RULES_15_BELOW_1SIGMA,
                        RULES_8_BEYOND_1SIGMA_BOTH_SIDES]

        RULES_ALL = [RULES_1_BEYOND_3SIGMA,
                     RULES_2_OF_3_BEYOND_2SIGMA,
                     RULES_4_OF_5_BEYOND_1SIGMA,
                     RULES_7_ON_ONE_SIDE,
                     RULES_8_ON_ONE_SIDE,
                     RULES_6_TRENDING,
                     RULES_14_UP_DOWN,
                     RULES_15_BELOW_1SIGMA,
                     RULES_8_BEYOND_1SIGMA_BOTH_SIDES]
        '''
    def _test_spec_run(self):
        if self.usl:
            above = self.data > self.usl
            self.usl_series = pd.Series(np.full(len(self.data.index), self.usl, dtype=np.float64), name = 'usl', index = self.data.index)
        else:
            above = np.array([False] * len(self.data.index))
        
        if self.lsl:
            below = self.data < self.lsl
            self.lsl_series = pd.Series(np.full(len(self.data.index), self.lsl, dtype=np.float64), name = 'lsl', index = self.data.index)
        else:
            below = np.array([False] * len(self.data.index))
        
        self.oos_series = pd.Series(self.data[above | below], name = 'oos', index = self.data.index)
        
        if self.usl or self.lsl: self._cpk_calc()
    
    def _cpk_calc(self):
        #Calc CPU
        try:
            cpu = (self.usl - self.centerline) / 3 / self.stddev
        except TypeError:
            cpu = False
        #Calc CPL
        try:
            cpl = (self.centerline - self.lsl) / 3 / self.stddev
        except TypeError:
            cpl = False
        #Take min of existing
        if cpu and cpl:
            self.cpk = min(cpu, cpl)
        elif cpu and not cpl:
            self.cpk = cpu
        elif not cpu and cpl:
            self.cpk = cpl
            
    def _test0_run(self):
        "1 point > n standard deviations from centerline, returns True if out"
        above = self.data > (self.centerline + self.sigma_level * self.stddev)
        below = self.data < (self.centerline - self.sigma_level * self.stddev)
        return above | below
    def _test1_run(self):
        return [False] * len(self.data.index)
    def _test2_run(self):
        return [False] * len(self.data.index)
    def _test3_run(self):
        return [False] * len(self.data.index)
    def _test4_run(self):
        return [False] * len(self.data.index)
    def _test5_run(self):
        return [False] * len(self.data.index)
    def _test6_run(self):
        return [False] * len(self.data.index)
    def _test7_run(self):
        return [False] * len(self.data.index)
    '''tests = {
        0: [_test0_run(), "1 point > n standard deviations from centerline"],
        1: [_test1_run(), "n points in a row on same side of centerline"],
        2: [_test2_run(), "n points in a row, all increasing or all decreasing"],
        3: [_test3_run(), "n points in a row, alternating up and down"],
        4: [_test4_run(), "n of n+1 > 2 standard deviations from center line on same side"],
        5: [_test5_run(), "n of n+1 > 1 standard deviations from center line on same side"],
        6: [_test6_run(), "n points in a row within 1 standard deviation of centerline on either side"],
        7: [_test7_run(), "n points in a row greater than 1 standard deviation from centerline on either side"]
    }'''
    
    
    ###need to concatenate plot points based on the series index entering the class
    def get_plot_points(self):
        '''Returns pandas dataframe for individual data, center line value, upper and lower control limits, upper and lower spec limits, as applicable'''
        plots = pd.concat(
            [
            self.data.rename('pts'),
            self.centerline_series,
            self.lcl_series,
            self.ucl_series
            ], axis = 1)
        if self.usl: plots['usl'] = self.usl_series
        if self.lsl: plots['lsl'] = self.lsl_series
        if self.usl or self.lsl: plots['oos'] = self.oos_series
        if self.test_group: plots['fail'] = self.fail_series
        return plots


    

In [144]:

    

class MovingRange(ControlChart):
    '''
    Generates Shewhart moving range control chart data, from positional element data, including approximate sigma. Plot using other packages
    observation range, how large of a subset to review range of; default 2
    sigma_level, test for special cause 1 rejection level; default 3
    '''
    
    def __init__(self, *args, observation_range = 2, stddev_only = False, **kwargs):
        super(MovingRange, self).__init__(*args, **kwargs)
        self.observation_range = observation_range
        self._moving_range_array()
        #self._moving_range_median() if use_median else self._moving_range_mean()
        self._mean()
        self._stddev()
        if not stddev_only:
            self._centerline()
            self._lcl()
            self._ucl()
            self._test_list()
            self._test_spec_run()
        #print('sigma level Moving range', self.sigma_level)
        
    def _moving_range_array(self):
        '''Generate an array from the moving average for the given observation range. Overwrites input data, for Class consistancy.
        Need to perform time study to determine if this method or if building comparison method is faster'''

        #add nan to the beginning of each segment
        mr = [] #[np.NaN] * (self.observation_range - 1)
        #pd.Series([np.NaN] * (self.observation_range - 1), index = self.data.index[:self.observation_range], name = 'pts')
        blank = list(range(self.observation_range - 1))
        for i in range(len(self.data.index)):
            if i in blank:
                mr.append(np.NaN)
            else:
                subset = self.data[i - self.observation_range + 1 : i + 1]
                mr.append(max(subset) - min(subset))
        self.data = pd.Series(mr, index = self.data.index, name = 'pts')
        
    def _mean(self):
        self.mean = self.data.mean()#[0] #get value only
    #def _median(self):
    #unable to achieve same stats using this method. More investigation necessary. MR-Bar not the same
    #    self.centerline = self.data.median
    
    def _stddev(self):
        self.stddev = self.mean / unbias_const('d2', self.observation_range)
        
    def _centerline(self, index = 0):
        self.centerline = self.mean
        self.centerline_series = pd.Series(np.full(len(self.data.index), self.centerline, dtype=np.float64)
                                           , name = 'cen'
                                           , index = self.data.index
                                          )
    
    def _lcl(self):
        lcl = self.mean - self.sigma_level * unbias_const('d3', self.observation_range) * self.stddev
        self.lcl = max(0, lcl) 
        self.lcl_series = pd.Series(np.full(len(self.data.index), self.lcl, dtype=np.float64)
                                    , name = 'lcl'
                                    , index = self.data.index
                                   )
        
    def _ucl(self):
        self.ucl = self.mean + self.sigma_level * unbias_const('d3', self.observation_range) * self.stddev
        self.ucl_series = pd.Series(np.full(len(self.data.index), self.ucl, dtype=np.float64)
                                    , name = 'ucl'
                                    , index = self.data.index
                                   )

    #def _test_spec_run(self): #overwrite for moving range
    #    self.oos_series = pd.Series([np.NaN] * len(self.data.index), name = 'oos', index = self.data.index)
        
    def get_S(self):
        return self.stddev


    

In [130]:

    

class Individual(ControlChart):
    '''
    Generates plot data for base indvidual data. Accepts modifications to mean, stdev, UCL & LCL for historical comparison.
    observation range, how large of a subset to review range of, only applicable to moving range sub calculation; default 2
    sigma_level, test for special cause 1 rejection level; default 3
    '''
    def __init__(self, *args, process_mean = None, stddev = None, observation_range = 2, **kwargs):
        super(Individual, self).__init__(*args, **kwargs)
        self.observation_range = observation_range
        self._mean(process_mean)
        self.stddev = stddev if stddev else self.get_S()
        self._centerline()
        self._lcl()
        self._ucl()
        self._test_list()
        self._test_spec_run()
        self._ADnormality()
                
    def _mean(self, process_mean):
        self.mean = process_mean if process_mean else self.data.mean()
    
    def _centerline(self):
        self.centerline = self.mean #add future use of median if necessary
        self.centerline_series = pd.Series(np.full(len(self.data.index), self.centerline, dtype=np.float64)
                                           , name = 'cen' 
                                           , index = self.data.index
                                          )
        
    def _lcl(self):
        self.lcl = self.centerline - self.sigma_level * self.stddev
        self.lcl_series = pd.Series(np.full(len(self.data.index), self.lcl, dtype=np.float64)
                                    , name = 'lcl' 
                                    , index = self.data.index
                                    )
        
    def _ucl(self):
        self.ucl = self.centerline + self.sigma_level * self.stddev
        self.ucl_series = pd.Series(np.full(len(self.data.index), self.ucl, dtype=np.float64)
                                    , name = 'ucl' 
                                    , index = self.data.index
                                    )
    
    def get_S(self):
        mr = MovingRange(data, observation_range = self.observation_range, 
                         sigma_level = self.sigma_level, stddev_only = True)
        S = mr.get_S()
        del mr
        return S
    
    def _ADnormality(self):
        self.ADnorm, crit, sig_lvl = scipy.stats.anderson(self.data)


    

In [161]:

    

'''
random datasets to trial out for unit tests
'''

np.random.seed(2)
#uniform random data
data = np.random.uniform(0, 20, size=50)
#normal random data
data_norm = np.random.normal(loc = 10, scale = 3, size = 50)

#normal random data, with stage column
size = 5
df1 = pd.DataFrame({'data': np.random.normal(loc = 10, scale = 3, size = size)
                   , 'stage': ['stage0'] * size
                   }, index = range(size)
                  )
df2 = pd.DataFrame({'data': np.random.normal(loc = 20, scale = 3, size = size)
                   , 'stage': ['stage1'] * size
                   }, index = range(size, size*2)
                  )
df3 = pd.DataFrame({'data': np.random.normal(loc = 10, scale = 3, size = size)
                   , 'stage': ['stage0'] * size
                   }, index = range(size * 2, size * 3)
                  )
data_stage = pd.concat([df1, df2, df3])


    

In [162]:

    

## test code
ser1 = pd.Series(['A0', 'A1', 'A2', 'A3'], name = 'A', index = [0,1,2,3])
ser2 = pd.Series(['B0', 'B1', 'B2', 'B3'], name = 'B', index = [1,2,3,4]) 
ser3 = pd.Series(['C0', 'C1', 'C2', 'C3'], name = 'C', index = [1,2,3,4])
ser4 = pd.Series(['D0', 'D1', 'D2', 'D3'], name = 'D', index = [0,1,2,4])
ser5 = pd.Series(['D0', 'D1', 'D2', 'D3'], name = 'D', index = [5,6,7,8])
ser6 = pd.Series(np.full(4, 10, dtype=np.float64),name = 'E', index = ser2.index)
a = pd.concat([ser2, ser3, ser6], axis = 1)


    

In [160]:

    

class ControlChartFigureWrapper():
    '''
    Data, and stage must be provided as a single pandas series, or dataframe with only one column.
    '''
    
    #formatting defaults
    fmt_pts  = {'color': 'black', 'linestyle': 'solid', 'marker': 's'} #data
    fmt_fail = {'color': 'green', 'linestyle': None,    'marker': 's'} #failed points
    fmt_spec = {'color': 'red',   'linestyle': None,    'marker': 's'} #out_of_spec
    fmt_cen  = {'color': 'green', 'linestyle': 'dashed', 'marker': None} #centerlines
    fmt_cl   = {'color': 'green', 'linestyle': 'solid', 'marker': None} #control limits
    fmt_sl   = {'color': 'red',   'linestyle': 'solid', 'marker': None} #spec limits
    
    stats = {'centerline': True, 'stddev': True, 'ucl': True, 'lcl': True, 'usl': False, 'lsl': False, 'cpk': False,
            'A-D normality': False
            }
    def __init__(self, data, charts = 'H-I-MR', *args, **kwargs):
        self.data_stages, self.stage_breaks = self._stage_data(data, *args, **kwargs)
        self.stage_breaks = [max(data.index) + 0.5 for stage, data in self.data_stages][:-1]
        #plt.close('all')
      
        #try: ylabel = kwarg.pop('ylabel')
        #except: ylabel = ''
        figsize = kwargs.get('figsize', (10,10))
        self.charts = charts
        self.fmt_pts.update(kwargs.get('fmt_pts', {}))
        self.fmt_fail.update(kwargs.get('fmt_fail', {}))
        self.fmt_spec.update(kwargs.get('fmt_spec', {}))
        self.fmt_cen.update(kwargs.get('fmt_cen', {}))
        self.fmt_cl.update(kwargs.get('fmt_cl', {}))
        self.fmt_sl.update(kwargs.get('fmt_sl', {}))
        self.stats.update(kwargs.get('stats', {}))
        {
            'H-I-MR': self._figure_HIMR,
            'I': self._figure_Individual,
            'I-MR': self._figure_IMR
            #'XR'
            #'XS'
            #'H'
        }[charts](figsize = figsize, *args, **kwargs)

        
    def _figure_HIMR(self, figsize, *args, **kwargs):
        self.gs = gridspec.GridSpec(2, 2,
                       width_ratios=[1,3],
                       height_ratios=[5,3]
                       )
        plt.figure(1, figsize=figsize)
        self._plot_mr(*args, gs_pos = 3, **kwargs)
        self._plot_individual(*args, gs_pos = 1, **kwargs)
        self._plot_hist(*args, gs_pos = 0, **kwargs)
        self._plot_stats(*args, gs_pos = 2, **kwargs)
        
    def _figure_Individual(self, figsize, *args, **kwargs):
        self.gs = gridspec.GridSpec(1,1)
        
        plt.figure(2, figsize=figsize)
        self._plot_individual(*args, gs_pos = 0, **kwargs)
    
    def _figure_IMR(self, figsize, *args, **kwargs):
        self.gs = gridspec.GridSpec(2, 1,
                       height_ratios=[5,3]
                       )
        plt.figure(3, figsize=figsize)
        self._plot_mr(*args, gs_pos = 1, **kwargs)
        self._plot_individual(*args, gs_pos = 0, **kwargs)
        
    def _plot_individual(self, gs_pos, *args, **kwargs):
        #Initialize plot, grab kwargs
        spec_min = kwargs.get('spec_min', False)
        spec_max = kwargs.get('spec_max', False)
        ylabel = kwargs.get('ylabel', False)
        special_cause_tests = kwargs.get('test_group', False)
        out_of_spec = spec_min or spec_max
        
        try:
            self.individuals = [Individual(data, stddev = stddev, *args, **kwargs)
                for (stage, data), stddev in zip(self.data_stages, self.mr_stddev)
                ]
        except ValueError:
            self.individuals = [Individual(data, *arg, **kwargs)
                for stage, data in self.data_stages
                ]
        
        plot_points = [ind.get_plot_points() for ind in self.individuals]
        ind = pd.concat(plot_points)
        for index in self.stage_breaks:
            ind.loc[index] = np.NaN
        self.ind = ind.sort_index()

        self.ind_axes = plt.subplot(self.gs[gs_pos])
        plt.plot(self.ind['lcl'], **self.fmt_cl)
        plt.plot(self.ind['ucl'], **self.fmt_cl)
        if spec_min:
            plt.plot(self.ind['lsl'], **self.fmt_sl)
        if spec_max:
            plt.plot(self.ind['usl'], **self.fmt_sl)
        plt.plot(self.ind['cen'], **self.fmt_cen)
        plt.plot(self.ind['pts'], **self.fmt_pts)
        if special_cause_tests:
            plt.plot(self.ind['fail'], **self.fmt_fail)
        if out_of_spec:
            plt.plot(self.ind['oos'], **self.fmt_spec)
        plt.title('Individual')
    
    def _plot_mr(self, *args, gs_pos, **kwargs):
        #Initialize plot, grab kwargs
        ylabel = kwargs.get('ylabel', False)
        special_cause_tests = kwargs.get('test_group', False)
        
        moving_ranges = [MovingRange(data, *args, **kwargs) for stage, data in self.data_stages]
        self.mr_stddev = [mr.get_S() for mr in moving_ranges]
        plot_points = [mr.get_plot_points() for mr in moving_ranges]
        mr = pd.concat(plot_points)
        for index in self.stage_breaks:
            mr.loc[index] = np.NaN
        self.mr = mr.sort_index()
        
        plt.subplot(self.gs[gs_pos])
        plt.plot(self.mr['lcl'], **self.fmt_cl)
        plt.plot(self.mr['ucl'], **self.fmt_cl)
        plt.plot(self.mr['cen'], **self.fmt_cen)
        plt.plot(self.mr['pts'], **self.fmt_pts)
        if special_cause_tests:
            plt.plot(self.mr['fail'], **self.fmt_fail)
        plt.title('Moving Range')
        if ylabel:
            plt.ylabel(ylabel)
    
    def _plot_hist(self, *args, gs_pos, num_bins = 10, orientation = 'horizontal', **kwargs):
        #Initialize plot, grab kwargs
        spec_min = kwargs.get('spec_min', False)
        spec_max = kwargs.get('spec_max', False)
        ylabel = kwargs.get('ylabel', False)
        #charts = kwargs.get('charts', 'H-I-MR')
        
        if self.charts == 'H-I-MR':
            self.hist_axes = plt.subplot(self.gs[gs_pos], sharey = self.ind_axes)
            #single histogram for all data, need to implement something that plots groups on each other, maybe.
            data = np.array([item for stage, group in self.data_stages for item in group])
            plt.hist(data, num_bins, normed=True, orientation = orientation, color = 'orange')
            
            x_array = np.linspace(*self.hist_axes.get_ylim(), num = 50)
            y = scipy.stats.norm.pdf(x_array, data.mean(), data.std())
            
            plt.plot(y, x_array, **self.fmt_cen)
            
            plt.xticks(np.linspace(*self.hist_axes.get_xlim(), num = 5) )
            if spec_min:
                plt.axhline(y=spec_min, **self.fmt_sl)
            if spec_max:
                plt.axhline(y=spec_max, **self.fmt_sl)
            if ylabel:
                plt.ylabel(ylabel)
            formatter = tkr.FormatStrFormatter('%1.2f')
            self.hist_axes.xaxis.set_major_formatter(formatter)
            plt.gca().invert_xaxis()
        else:
            self.hist_axes = plt.subplot(gs[gs_pos])
            plt.hist(self.data_stages, num_bins, normed=1, orientation = 'vertical', color = 'orange')
        
    def _plot_stats(self, *args, gs_pos, **kwargs):
        plt.subplot(self.gs[gs_pos])
        plt.axis('off')
        stat_result = []
        if self.stats['centerline']: 
            cen = ['{:1.4g}'.format(ind.mean) for ind in self.individuals]
            stat_result.append('Centerline: ' + ', '.join(cen))
        if self.stats['stddev']:
            stddev = ['{:1.4f}'.format(ind.stddev) for ind in self.individuals]
            stat_result.append('Std Dev: ' + ', '.join(stddev))
        if self.stats['ucl']: 
            ucl = ['{:1.4g}'.format(ind.ucl) for ind in self.individuals]
            stat_result.append('UCL: ' + ', '.join(ucl)) 
        if self.stats['lcl']: 
            lcl = ['{:1.4g}'.format(ind.lcl) for ind in self.individuals]
            stat_result.append('LCL: ' + ', '.join(lcl))
        if self.stats['usl']: 
            usl = ['{:1.4g}'.format(ind.usl) for ind in self.individuals]
            stat_result.append('USL: ' + ', '.join(usl))
        if self.stats['lsl']: 
            lsl = ['{:1.4g}'.format(ind.lsl) for ind in self.individuals]
            stat_result.append('LSL: ' + ', '.join(lsl))
        if self.stats['cpk']: 
            cpk = ['{:1.4g}'.format(ind.cpk) for ind in self.individuals]
            stat_result.append('CPk: ' + ', '.join(cpk))
        if self.stats['A-D normality']: 
            norm = ['{:1.4g}'.format(ind.ADnorm) for ind in self.individuals]
            stat_result.append('Normality: ' + ', '.join(norm))
        plt.text(0.0, 0.5, '\n'.join(stat_result))
    
    def _stage_data(self, data, *args, stages = pd.Series(), split_method = None, **kwargs): 
        '''
        Restructures the data given into an orderedDict of stage name keys
        and stage dataframe subsets
        '''
        if split_method and (len(data) != len(stages)):
            raise ValueError('Provided data and Stage length differ')
        if stages.any() and not split_method:
            raise NameError('No split method provided with stage list')

        data = pd.Series(data)
        stages = pd.Series(stages)
        stage_frames = []
        
        def _none():
            key = data.name
            values = data
            stage_frames.append((key, values))
            stage_breaks = []
            return stage_frames, stage_breaks

        def _unique():
            #obtain unique stages
            keys = stages.unique()
            for key in keys:
                values = data[stages == key]
                stage_frames.append((key, values))
            #obtain stage breaks
            init = True
            key = None
            values = None
            stage_breaks = []

            for n in range(len(data.index)):
                if init: #init only true once
                    key = stages[n]
                    values = data[n:n+1]
                    init = False
                elif not init and key == stages[n]:
                    values = data[n:n+1]
                else:
                    stage_breaks.append(max(values.index) + 0.5)
                    key = stages[n]
                    values = data[n:n+1]
            return stage_frames, stage_breaks
                
        def _change():
            init = True
            key = None
            values = None
            stage_breaks = []
            #for value, stage in zip(data, stages):
            for n in range(len(data.index)):
                if init: #init only true once
                    key = stages[n]
                    values = pd.Series(data[n:n+1])
                    init = False
                elif not init and key == stages[n]:
                    values = values.append(data[n:n+1])
                else:
                    stage_frames.append((key, values)) #save last round data
                    stage_breaks.append(max(values.index) + 0.5)
                    key = stages[n]
                    values = pd.Series(data[n:n+1])
            stage_frames.append((key, values)) #save last data point
            return stage_frames, stage_breaks
        
        return {
            None: _none,
            'unique': _unique,
            'change': _change
        }[split_method]()

        
#unit test?
#data_df = pd.DataFrame(data)
stats = {'centerline': True, 'stddev': True, 'ucl': True, 'lcl': True, 'usl': False, 'lsl': False, 'cpk': False,
            'A-D normality': True
            }
ControlChartFigureWrapper(data_stage['data']
                          , charts = 'H-I-MR'
                          , ylabel = 'Random'
                          , stats = stats
                          , test_group = [0]
                          , spec_min = 5
                          , stages = data_stage['stage']
                         , split_method = 'change'
                          , observation_range = 2
                          )


    

    
Out[160]:





<__main__.ControlChartFigureWrapper at 0x1fc75c90>






    

In [ ]:

    




    

In [ ]:

    

http://docs.scipy.org/doc/scipy/reference/stats.html