analytics/informix/extend/TimeSeries.6.00.FC8/examples/tsmov.c

/*
 * Licensed Materials - Property of HCL
 *
 * IBM Informix  DataBlade Module
 * (C) Copyright International Business Machines Corporation 2002.
 * (c) Copyright HCL Technologies Ltd. 2017.  All Rights Reserved.
 *
 * COPYRIGHT LICENSE: 
 * This information contains sample application programs in source language,
 * which illustrate programming techniques on various operating platforms.
 * You may copy, modify, and distribute these sample programs in any form
 * without payment to IBM, for the purposes of developing, using, marketing 
 * or distributing application programs conforming to the application 
 * programming interface for the operating platform for which the sample 
 * programs are written. These examples have not been thoroughly tested under 
 * all conditions. IBM, therefore, cannot guarantee or imply reliability, 
 * serviceability, or function of these programs. You may copy, modify, and 
 * distribute these sample programs in any form without payment to IBM for 
 * the purposes of developing, using, marketing, or distributing application 
 * programs conforming to IBM's application programming interfaces. 
 * Each copy or any portion of these sample programs or any derivative work, 
 * must include a copyright notice as follows: 
 * © (your company name) (year). Portions of this code are derived from 
 * IBM Corp. Sample Programs. © Copyright IBM Corp. (enter the year or 
 * years). All rights reserved. 
 *
 */

# include "tseries.h"

/*
 * Sample moving average code
 */

/* base types we can handle */
typedef enum
{
    ACC_INT2,
    ACC_INT4,
    ACC_INT8,
    ACC_SMALLFLOAT,
    ACC_FLOAT,
    ACC_MONEY,
    ACC_DECIMAL
} ts_acc_types;

/* values for accum_flags */
# define ACC_READY	1

/*& accumulator structure */
typedef struct _ts_accum {
    MI_DATUM	*accum_buffer;	/* buffer to hold values */
    mi_integer	 accum_pos;	/* index into buffer, put data here then inc */
    mi_integer	 accum_winsize;	/* number of intervals in window */
    mi_integer	 accum_colnum;	/* column number in element being accumulated */
    mi_integer	 accum_flags;	/* currently says if we have had a full window*/
    MI_DATUM	 accum_accum;	/* accumulator value */
    MI_DATUM	 accum_ret;	/* return value */
    ts_acc_types accum_type;	/* which base type this column is */
    struct _ts_accum *accum_next; /* next column accumulator pointer */
} ts_accum;

/*
 * funciton for doing a moving average.  This might be better split into
 * separate functions for each base type.
 */
static MI_DATUM
ts_mov_avg(ts_accum *accum,
	   MI_DATUM val,
	   mi_boolean nullval,
	   mi_boolean *isNull)
{
    mi_integer	pos;
    MI_DATUM	oldval;
    mi_smallint	int2_tmp;
    mi_integer	int4_tmp;
    mi_int8	int8_tmp;
    mi_decimal	dec_tmp;
    mi_real	real_tmp;
    mi_double_precision	dbl_tmp;

    /* get current index into buffer */
    pos = accum->accum_pos;

    switch (accum->accum_type) {
      case ACC_INT2:
	if ((accum->accum_flags & ACC_READY) || pos + 1 == accum->accum_winsize)
	    /* get old value if there was one */
	    int2_tmp = *((mi_smallint *) &accum->accum_buffer[pos]);

	/* put new value into buffer */
	if (nullval)
	    *(mi_smallint *) &val = 0;

	*((mi_smallint *) &accum->accum_buffer[pos]) = (mi_smallint) val;

	/* add new value to accumulator */
	*(mi_smallint *) &accum->accum_accum += (mi_smallint) val;

	if (++accum->accum_pos == accum->accum_winsize) {
	    /* we have filled our window for the first time */
	    accum->accum_flags |= ACC_READY;
	    accum->accum_pos = 0;
	}

	if (accum->accum_flags & ACC_READY) {
	    /* remove oldest value from accum if there was one */
	    *(mi_smallint *) &accum->accum_accum -= int2_tmp;
	    *(mi_smallint *) &accum->accum_ret = *(mi_smallint *) &accum->accum_accum / accum->accum_winsize;
	}
	break;

      case ACC_INT4:
	if ((accum->accum_flags & ACC_READY) || pos + 1 == accum->accum_winsize)
	    /* get old value if there was one */
	    int4_tmp = *((mi_integer *) &accum->accum_buffer[pos]);

	/* put new value into buffer */
	if (nullval)
	    *(mi_integer *) &val = 0;
	*((mi_integer *) &accum->accum_buffer[pos]) = (mi_integer) val;

	/* add new value to accumulator */
	*(mi_integer *) &accum->accum_accum += (mi_integer) val;

	if (++accum->accum_pos == accum->accum_winsize) {
	    /* we have filled our window for the first time */
	    accum->accum_flags |= ACC_READY;
	    accum->accum_pos = 0;
	}

	if (accum->accum_flags & ACC_READY) {
	    /* remove oldest value from accum if there was one */
	    *(mi_integer *) &accum->accum_accum -= (mi_integer) int4_tmp;
	    *(mi_integer *) &accum->accum_ret = *(mi_integer *) &accum->accum_accum / accum->accum_winsize;
	}
	break;

      case ACC_INT8:
	if ((accum->accum_flags & ACC_READY) || pos + 1 == accum->accum_winsize)
	    /* get old value if there was one */
	    ifx_int8copy((mi_int8 *) accum->accum_buffer[pos], &int8_tmp);

	/* put new value into buffer */
	if (nullval)
	    ifx_int8cvint(0, (mi_int8 *) accum->accum_buffer[pos]);
	else {
	    ifx_int8copy((mi_int8 *) val, (mi_int8 *) accum->accum_buffer[pos]);

	    /* add new value to accumulator */
	    ifx_int8add((mi_int8 *) accum->accum_accum, (mi_int8 *) val, (mi_int8 *) accum->accum_accum);
	}

	if (++accum->accum_pos == accum->accum_winsize) {
	    /* we have filled our window for the first time */
	    accum->accum_flags |= ACC_READY;
	    accum->accum_pos = 0;
	}

	if (accum->accum_flags & ACC_READY) {
	    /* remove oldest value from accum if there was one */
	    ifx_int8sub((mi_int8 *) &accum->accum_accum, &int8_tmp, (mi_int8 *) accum->accum_accum);
	    ifx_int8cvint(accum->accum_winsize, &int8_tmp);
	    ifx_int8div((mi_int8 *) &accum->accum_accum, &int8_tmp, (mi_int8 *) accum->accum_ret);
	}
	break;

      case ACC_SMALLFLOAT:
	if ((accum->accum_flags & ACC_READY) || pos + 1 == accum->accum_winsize)
	    /* get old value if there was one */
	    real_tmp = *((mi_real *) accum->accum_buffer[pos]);

	/* put new value into buffer */
	if (nullval)
	    *((mi_real *) accum->accum_buffer[pos]) = (mi_real) 0;
	else {
	    *((mi_real *) accum->accum_buffer[pos]) = *(mi_real *) val;

	    /* add new value to accumulator */
	    *(mi_real *) accum->accum_accum += *(mi_real *) val;
	}

	if (++accum->accum_pos == accum->accum_winsize) {
	    /* we have filled our window for the first time */
	    accum->accum_flags |= ACC_READY;
	    accum->accum_pos = 0;
	}

	if (accum->accum_flags & ACC_READY) {
	    /* remove oldest value from accum if there was one */
	    *(mi_real *) accum->accum_accum -= real_tmp;
	    *(mi_real *) accum->accum_ret = *(mi_real *) accum->accum_accum / accum->accum_winsize;
	}
	break;

      case ACC_FLOAT:
	if ((accum->accum_flags & ACC_READY) || pos + 1 == accum->accum_winsize)
	    /* get old value if there was one */
	    dbl_tmp = *((mi_double_precision *) accum->accum_buffer[pos]);

	/* put new value into buffer */
	if (nullval)
	    *((mi_double_precision *) accum->accum_buffer[pos]) = (mi_double_precision) 0;
	else {
	    *((mi_double_precision *) accum->accum_buffer[pos]) = *(mi_double_precision *) val;

	    /* add new value to accumulator */
	    *(mi_double_precision *) accum->accum_accum += *(mi_double_precision *) val;
	}

	if (++accum->accum_pos == accum->accum_winsize) {
	    /* we have filled our window for the first time */
	    accum->accum_flags |= ACC_READY;
	    accum->accum_pos = 0;
	}

	if (accum->accum_flags & ACC_READY) {
	    /* remove oldest value from accum if there was one */
	    *(mi_double_precision *) accum->accum_accum -= dbl_tmp;
	    *(mi_double_precision *) accum->accum_ret = *(mi_double_precision *) accum->accum_accum / accum->accum_winsize;
	}
	break;

      case ACC_DECIMAL:
      case ACC_MONEY:
	if ((accum->accum_flags & ACC_READY) || pos + 1 == accum->accum_winsize)
	    /* get old value if there was one */
	    deccopy((mi_decimal *) accum->accum_buffer[pos], &dec_tmp);

	/* put new value into buffer */
	if (nullval)
	    deccvint(0, (mi_decimal *) accum->accum_buffer[pos]);
	else {
	    deccopy((mi_decimal *) val, (mi_decimal *) accum->accum_buffer[pos]);

	    /* add new value to accumulator */
	    decadd((mi_decimal *) accum->accum_accum, (mi_decimal *) val, (mi_decimal *) accum->accum_accum);
	}

	if (++accum->accum_pos == accum->accum_winsize) {
	    /* we have filled our window for the first time */
	    accum->accum_flags |= ACC_READY;
	    accum->accum_pos = 0;
	}

	if (accum->accum_flags & ACC_READY) {
	    /* remove oldest value from accum if there was one */
	    decsub((mi_decimal *) &accum->accum_accum, &dec_tmp, (mi_decimal *) accum->accum_accum);
	    deccvint(accum->accum_winsize, &dec_tmp);
	    decdiv((mi_decimal *) accum->accum_accum, &dec_tmp, (mi_decimal *) accum->accum_ret);
	}
	break;
    }

    /* return null until we have seen winsize number of values */
    *isNull = !(accum->accum_flags & ACC_READY);

    /* return result */
    return(accum->accum_ret);
}

/*
 * call a function over a moving window.  Currently only avg is implemented
 */
ts_timeseries *
ts_moving_func(MI_CONNECTION *conn,
	       ts_tsdesc *tsdesc,	/* source timeseries */
	       mi_integer window_size,	/* number of intervals in window */
	       mi_integer colnum,	/* col to agg, -1 mean all cols */
	       mi_datetime *tstart,	/* start date, can be null */
	       mi_datetime *tend,	/* end date, can be null */
	       mi_string *func_name)	/* name of function to run */
{
    ts_accum	*accum, *head;
    MI_DATUM	(*func)(ts_accum *, MI_DATUM, mi_boolean, mi_boolean *);
    mi_string	*typename;
    mi_integer	 start, end, i, j;
    ts_typeinfo *tinfo;
    ts_tscan	*scan;
    ts_timeseries *newts, *ts;
    ts_tselem	 elem, newelem;
    ts_tsdesc	*newtsdesc;
    MI_DATUM	 newval;
    mi_boolean	 isNull;
    mi_boolean	*nulls;
    MI_DATUM	*values;

    /* only avg is currently implemented */
    if (strcmp(func_name, "avg") == 0) {
	func = ts_mov_avg;
    } else {
	mi_db_error_raise(NULL, MI_FATAL, "bad function passed in");
    }

    /* setup min and max column numbers */
    if (colnum == -1) {
	/* doing all columns */
	start = 1;
	end = ts_col_cnt(tsdesc);
    } else {
	/* doing just one column */
	start = colnum;
	end = colnum + 1;
    }

    /* make a linked list of all columns involved */
    head = NULL;

    for (i = start; i < end; i++) {
	/*
	 * note: code further on depends on accum
	 * to point to columns in decending order 
	 */
	accum = mi_zalloc(sizeof(ts_accum));

	if (head == NULL) {
	    /* first in list */
	    head = accum;
	    accum->accum_next = NULL;
	} else {
	    /* add to head of list */
	    accum->accum_next = head;
	    head = accum;
	}

	/* setup window buffer for this column */
	accum->accum_buffer = mi_zalloc(window_size * sizeof(MI_DATUM));
	accum->accum_flags = 0;
	accum->accum_pos = 0;
	accum->accum_winsize = window_size;
	accum->accum_colnum = i;

	tinfo = ts_colinfo_number(tsdesc, i);

	/* setup type specific info */
	if (strcmp(tinfo->ti_typename, "smallint") == 0) {
	    accum->accum_type = ACC_INT2;
	} else if (strcmp(tinfo->ti_typename, "integer") == 0) {
	    accum->accum_type = ACC_INT4;
	} else if (strcmp(tinfo->ti_typename, "int8") == 0) {
	    /* int8's are accessed as pointers, so we must allocate space */
	    accum->accum_type = ACC_INT8;
	    accum->accum_accum = (MI_DATUM) mi_zalloc(sizeof(mi_int8));
	    accum->accum_ret = (MI_DATUM) mi_zalloc(sizeof(mi_int8));
	    for (j = 0; j < accum->accum_winsize; j++)
		accum->accum_buffer[j] = (MI_DATUM) mi_zalloc(sizeof(mi_int8));
	} else if (strcmp(tinfo->ti_typename, "smallfloat") == 0) {
	    /* floats are accessed as pointers, so we must allocate space */
	    accum->accum_type = ACC_SMALLFLOAT;
	    accum->accum_accum = (MI_DATUM) mi_zalloc(sizeof(mi_real));
	    accum->accum_ret = (MI_DATUM) mi_zalloc(sizeof(mi_real));
	    for (j = 0; j < accum->accum_winsize; j++)
		accum->accum_buffer[j] = (MI_DATUM) mi_zalloc(sizeof(mi_real));
	} else if (strcmp(tinfo->ti_typename, "float") == 0) {
	    /* floats are accessed as pointers, so we must allocate space */
	    accum->accum_type = ACC_FLOAT;
	    accum->accum_accum = (MI_DATUM) mi_zalloc(sizeof(mi_double_precision));
	    accum->accum_ret = (MI_DATUM) mi_zalloc(sizeof(mi_double_precision));
	    for (j = 0; j < accum->accum_winsize; j++)
		accum->accum_buffer[j] = (MI_DATUM) mi_zalloc(sizeof(mi_double_precision));
	} else if (strcmp(tinfo->ti_typename, "money") == 0) {
	    /* money is accessed as pointers, so we must allocate space */
	    accum->accum_type = ACC_MONEY;
	    accum->accum_accum = (MI_DATUM) mi_zalloc(sizeof(mi_decimal));
	    accum->accum_ret = (MI_DATUM) mi_zalloc(sizeof(mi_decimal));
	    for (j = 0; j < accum->accum_winsize; j++)
		accum->accum_buffer[j] = (MI_DATUM) mi_zalloc(sizeof(mi_decimal));
	} else if (strcmp(tinfo->ti_typename, "decimal") == 0) {
	    /* decimals are accessed as pointers, so we must allocate space */
	    accum->accum_type = ACC_DECIMAL;
	    accum->accum_accum = (MI_DATUM) mi_zalloc(sizeof(mi_decimal));
	    accum->accum_ret = (MI_DATUM) mi_zalloc(sizeof(mi_decimal));
	    for (j = 0; j < accum->accum_winsize; j++)
		accum->accum_buffer[j] = (MI_DATUM) mi_zalloc(sizeof(mi_decimal));
	} else
	    /* we only handle the above column types */
	    mi_db_error_raise(NULL, MI_FATAL, "type not supported");
    }

    /* get the source timeseries */
    ts = ts_get_ts(tsdesc);

    /* create the return timeseriers */
    newts = ts_create(conn,
		      ts_get_calname(ts),
		      tstart == NULL ? ts_get_origin(ts) : tstart,
		      ts_get_threshold(ts),
		      TS_IS_IRREGULAR(ts) ? TSFLAGS_IRR : 0,
		      ts_get_typeid(conn, ts),
		      (tstart != NULL && tend != NULL)
			? ts_cal_index(conn, ts_get_calname(ts), tstart, tend)
			: 0,
		      ts_get_containername(ts));

    /* open the return timeseries */
    newtsdesc = ts_open(conn, newts, ts_get_typeid(conn, ts), 0);

    /* begin the scan of the source timeseries */
    scan = ts_begin_scan(tsdesc, 0, tstart, tend);

    /* allocate the arrays for creating elements */
    values = (MI_DATUM *) mi_zalloc(sizeof(MI_DATUM) * ts_col_cnt(tsdesc));
    nulls = (mi_boolean *) mi_zalloc(sizeof(mi_boolean) * ts_col_cnt(tsdesc));

    newelem = NULL;
    while (ts_next(scan, &elem) != TS_SCAN_EOS) {
	accum = head;
	nulls[0] = MI_TRUE;	/* the datetime column is always null */

	/* loop through each column of each element last col to first */
	for (i = ts_col_cnt(tsdesc) - 1; i >= 1; i--) {
	    /* get column "i" value */
	    if (elem == NULL)
		nulls[i] = MI_TRUE;
	    else
		values[i] = ts_get_col_by_number(tsdesc, elem, i, &nulls[i],-1);

	    if (accum && accum->accum_colnum == i) {
		/* this is a column we are accumulating */
		isNull = MI_FALSE;

		/* call accumulator function */
		values[i] = func(accum, values[i], nulls[i], &isNull);
		nulls[i] = isNull;

		/* on to next column, if there is one */
		accum = accum->accum_next;
	    }
	}

	/* make the element to put in the return timeseries */
	newelem = ts_make_elem_with_buf(newtsdesc, values, nulls, NULL, newelem);

	/* put it into the return timeseries at the given stamp */
	(void) ts_put_elem(newtsdesc, newelem, ts_current_timestamp(scan));
    }

    /* clean up */
    ts_end_scan(scan);
    ts_close(newtsdesc);

    /* all done */
    return(newts);
}

/*
 * SQL entry point
 *    create function TsMovAvg(TimeSeries,
 *    			 integer,
 *    			 integer,
 *    			 datetime year to fraction(5) default NULL,
 *    			 datetime year to fraction(5) default NULL)
 *        returns timeseries with (handlesnulls)
 *        external name
 *        '$INFORMIXDIR/functions/tsmov.bld(ts_moving_avg_sql)'
 *        language c not variant;
 */
#ifdef NT
__declspec(dllexport)
#endif
ts_timeseries *
ts_moving_avg_sql(ts_timeseries *ts,
	       mi_integer window_size,	/* number of intervals in window */
	       mi_integer colnum,	/* col to agg, -1 means all */
	       mi_datetime *tstart,	/* start time, can be null */
	       mi_datetime *tend,	/* end time, can be null */
	       MI_FPARAM *fp)
{
    MI_CONNECTION	*conn;
    ts_timeseries	*ret;
    ts_tsdesc		*tsdesc;

    /* get a connections */
    conn = mi_open(NULL, NULL, NULL);

    /* see if start and end were passed as null values */
    if (mi_fp_argisnull(fp, 3))
	tstart = 0;
    if (mi_fp_argisnull(fp, 4))
	tend = 0;

    /* open the source timeseries */
    tsdesc = ts_open(conn, ts, mi_fp_argtype(fp, 0), 0);

    /* go do the work! */
    ret = ts_moving_func(conn, tsdesc, window_size, colnum, tstart, tend, "avg");

    /* clean up */
    ts_close(tsdesc);
    mi_close(conn);

    return(ret);
}