[Scipy-svn] r2865 - in trunk/Lib/sandbox/timeseries/io/fame: . tests
scipy-svn at scipy.org
scipy-svn at scipy.org
Thu Mar 22 09:46:19 EDT 2007
Author: mattknox_ca
Date: 2007-03-22 08:46:15 -0500 (Thu, 22 Mar 2007)
New Revision: 2865
Modified:
trunk/Lib/sandbox/timeseries/io/fame/core.py
trunk/Lib/sandbox/timeseries/io/fame/tests/test_fame.py
Log:
updated to work with recent overhaul of timeseries module
Modified: trunk/Lib/sandbox/timeseries/io/fame/core.py
===================================================================
--- trunk/Lib/sandbox/timeseries/io/fame/core.py 2007-03-21 23:05:27 UTC (rev 2864)
+++ trunk/Lib/sandbox/timeseries/io/fame/core.py 2007-03-22 13:46:15 UTC (rev 2865)
@@ -3,6 +3,7 @@
import numpy
import maskedarray as ma
import timeseries as ts
+from timeseries import const as _c
import cfame
from const import *
@@ -14,23 +15,20 @@
def reverse_dict(d):
return dict([(y, x) for x, y in d.iteritems()])
-def convert_dict(d, key_func=lambda x:x, val_func=lambda x:x):
- return dict([(key_func(key), val_func(val)) for key, val in d.iteritems()])
-
-basis_map = { HBSUND:"U",
- HBSDAY:"D",
- HBSBUS:"B"}
+basis_map = { HBSUND:_c.FR_UND,
+ HBSDAY:_c.FR_DAY,
+ HBSBUS:_c.FR_BUS}
basis_revmap = reverse_dict(basis_map)
-observed_map = { HOBUND:"UNDEFINED",
- HOBBEG:"BEGINNING",
- HOBEND:"ENDING",
- HOBAVG:"AVERAGED",
- HOBSUM:"SUMMED",
- HOBANN:"ANNUALIZED",
- HOBFRM:"FORMULA",
- HOBHI:"MAXIMUM",
- HOBLO:"MINIMUM"}
+observed_map = { HOBUND:ts.check_observed("UNDEFINED"),
+ HOBBEG:ts.check_observed("BEGINNING"),
+ HOBEND:ts.check_observed("ENDING"),
+ HOBAVG:ts.check_observed("AVERAGED"),
+ HOBSUM:ts.check_observed("SUMMED"),
+ HOBANN:"ANNUALIZED", #ts.check_observed("ANNUALIZED"),
+ HOBFRM:"FORMULA", #ts.check_observed("FORMULA"),
+ HOBHI:ts.check_observed("MAXIMUM"),
+ HOBLO:ts.check_observed("MINIMUM")}
observed_revmap = reverse_dict(observed_map)
observed_revmap['HIGH'] = HOBHI
observed_revmap['LOW'] = HOBLO
@@ -39,7 +37,7 @@
"translate user specified basis to FAME constant"
if isinstance(basis, str):
- freq = ts.freq_tostr(ts.freq_fromstr(basis))
+ freq = ts.check_freq(basis)
try:
return basis_revmap[freq]
except KeyError:
@@ -47,75 +45,88 @@
"'DAILY', 'BUSINESS', or 'UNDEFINED'")
else:
if basis in basis_map: return basis
- elif basis == ts.freq_fromstr('D'): return HBSDAY
- elif basis == ts.freq_fromstr('B'): return HBSBUS
- elif basis == ts.freq_fromstr('U'): return HBSUND
+ elif basis == _c.FR_DAY: return HBSDAY
+ elif basis == _c.FR_BUS: return HBSBUS
+ elif basis == _c.FR_UND: return HBSUND
else:
raise ValueError("Invalid Basis value")
def translate_observed(observed):
"translate user specified observed to FAME constant"
if isinstance(observed, str):
- return observed_revmap[ts.fmtObserv(observed)]
+ return observed_revmap[ts.check_observed(observed)]
elif observed in (observed_map):
return observed
else:
raise ValueError("Invalid Observed value")
-freq_map = { HDAILY:"D",
- HBUSNS:"B",
- HMONTH:"M",
- HWKSUN:"W",
- HSEC :"S",
- HMIN :"T",
- HHOUR :"H",
- HQTOCT:"Q",
- HQTNOV:"Q",
- HQTDEC:"Q",
- HANJAN:"A",
- HANFEB:"A",
- HANMAR:"A",
- HANAPR:"A",
- HANMAY:"A",
- HANJUN:"A",
- HANJUL:"A",
- HANAUG:"A",
- HANSEP:"A",
- HANOCT:"A",
- HANNOV:"A",
- HANDEC:"A" }
-freq_map = convert_dict(freq_map, val_func=ts.freq_fromstr)
+freq_map = { HDAILY:_c.FR_DAY,
+ HBUSNS:_c.FR_BUS,
+ HMONTH:_c.FR_MTH,
+ HWKSUN:_c.FR_WKSUN,
+ HWKMON:_c.FR_WKMON,
+ HWKTUE:_c.FR_WKTUE,
+ HWKWED:_c.FR_WKWED,
+ HWKTHU:_c.FR_WKTHU,
+ HWKFRI:_c.FR_WKFRI,
+ HWKSAT:_c.FR_WKSAT,
+ HSEC :_c.FR_SEC,
+ HMIN :_c.FR_MIN,
+ HHOUR :_c.FR_HR,
+ HQTOCT:_c.FR_QTR,
+ HQTNOV:_c.FR_QTR,
+ HQTDEC:_c.FR_QTR,
+ HANJAN:_c.FR_ANNJAN,
+ HANFEB:_c.FR_ANNFEB,
+ HANMAR:_c.FR_ANNMAR,
+ HANAPR:_c.FR_ANNAPR,
+ HANMAY:_c.FR_ANNMAY,
+ HANJUN:_c.FR_ANNJUN,
+ HANJUL:_c.FR_ANNJUL,
+ HANAUG:_c.FR_ANNAUG,
+ HANSEP:_c.FR_ANNSEP,
+ HANOCT:_c.FR_ANNOCT,
+ HANNOV:_c.FR_ANNNOV,
+ HANDEC:_c.FR_ANNDEC }
-freq_revmap = { "D" : HDAILY,
- "B" : HBUSNS,
- "M" : HMONTH,
- "W" : HWKSUN,
- "S" : HSEC,
- "T" : HMIN,
- "H" : HHOUR,
- "Q" : HQTDEC,
- "A" : HANDEC}
-freq_revmap = convert_dict(freq_revmap, key_func=ts.freq_fromstr)
+freq_revmap = reverse_dict(freq_map)
+freq_revmap[_c.FR_QTR] = HQTDEC
-date_value_adjust = { 'A':1849,
- 'Q':7396,
- 'M':22188,
- 'W':96477,
- 'B':482381,
- 'D':675333,
- 'H':87648,
- 'T':5258880,
- 'S':315532800}
-date_value_adjust = convert_dict(date_value_adjust, key_func=ts.freq_fromstr)
+date_value_adjust = { _c.FR_ANNJAN:1849,
+ _c.FR_ANNFEB:1849,
+ _c.FR_ANNMAR:1849,
+ _c.FR_ANNAPR:1849,
+ _c.FR_ANNMAY:1849,
+ _c.FR_ANNJUN:1849,
+ _c.FR_ANNJUL:1849,
+ _c.FR_ANNAUG:1849,
+ _c.FR_ANNSEP:1849,
+ _c.FR_ANNOCT:1849,
+ _c.FR_ANNNOV:1849,
+ _c.FR_ANNDEC:1849,
+ _c.FR_QTR:7396,
+ _c.FR_MTH:22188,
+ _c.FR_WKSUN:96477,
+ _c.FR_WKMON:96477,
+ _c.FR_WKTUE:96477,
+ _c.FR_WKWED:96477,
+ _c.FR_WKTHU:96477,
+ _c.FR_WKFRI:96477,
+ _c.FR_WKSAT:96477,
+ _c.FR_BUS:482381,
+ _c.FR_DAY:675333,
+ _c.FR_HR:87648,
+ _c.FR_MIN:5258880,
+ _c.FR_SEC:315532800}
def fametype_fromdata(data):
"""determine fame type code from a data object"""
-
+
if isinstance(data, ts.DateArray) or isinstance(data, ts.Date):
return freq_revmap[data.freq]
elif hasattr(data, 'dtype'):
dtypeStr = str(data.dtype)
-
+
if dtypeStr[:5] == "float":
if int(dtypeStr[5:]) > 32: return HPRECN
else: return HNUMRC
@@ -131,7 +142,7 @@
return HBOOLN
else:
raise ValueError("Unsupported dtype for fame database: %s", dtypeStr)
-
+
elif isinstance(data, str):
return HSTRNG
elif isinstance(data, (int, float)):
@@ -161,19 +172,19 @@
def __init__(self, data={}):
for i, v in data.iteritems():
self[i.upper()] = v
-
+
def __getitem__(self, key):
if hasattr(key, 'upper'): key = key.upper()
return super(CaseInsensitiveDict, self).__getitem__(key)
-
+
def __setitem__(self, key, item):
if hasattr(key, 'upper'): key = key.upper()
super(CaseInsensitiveDict, self).__setitem__(key, item)
-def _famedate_to_tsdate(fame_date, freqstr):
+def _famedate_to_tsdate(fame_date, freq):
"convert integer fame date to a timeseries Date"
- value = fame_date + date_value_adjust[ts.freq_fromstr(freqstr)]
- return ts.Date(freq=freqstr, value=value)
+ value = fame_date + date_value_adjust[freq]
+ return ts.Date(freq=freq, value=value)
def _fame_params_from_pyobj_scalar(pyobj):
@@ -221,7 +232,7 @@
:Construction:
x = FameDb(conn_str, mode='r')
-:Paramaters:
+:Parameters:
- `conn_str` (str) : valid connection string. Can be a physical path,
or channel specification, etc. See FAME documentation on cfmopdb for
valid connection strings.
@@ -258,14 +269,14 @@
else:
raise ValueError, "Database access mode not supported."
self.mode = mode
-
+
self.dbkey = cf_open(conn_str, intmode)
-
+
def read(self, name,
start_date=None, end_date=None,
start_case=None, end_case=None, max_string_len=65):
-
+
"""read specified object(s) from database
:Parameters:
@@ -288,7 +299,7 @@
or series of strings. This is the maximum length of string that can
be read. Lower values result in less memory usage, so you should
specify this as low as is reasonable for your data.
-
+
:Return:
if `name` is a list of strings:
case insensitive dictionary of the objects
@@ -303,7 +314,7 @@
names = name
items = CaseInsensitiveDict()
-
+
#default to -1. This will get the entire range
_start_case = _end_case = -1
_start_date = _end_date = -1
@@ -322,7 +333,7 @@
if start_case is not None: _start_case = start_case
if end_case is not None: _end_case = end_case
-
+
if len(set([_start_case, _end_case, _start_date, _end_date, -1])) != 1:
checkFreq = True
else:
@@ -376,7 +387,7 @@
pyObj = numpyType(pyObj)
elif result['class'] == HSERIE:
-
+
if 'data' in result:
vals = result['data']
mask = result['mask']
@@ -384,7 +395,7 @@
else:
vals = []
mask = ma.nomask
-
+
if result['type'] >= 8: # date type
valadj = date_value_adjust[freq_map[result['type']]]
if len(vals) > 0: vals += valadj
@@ -392,7 +403,7 @@
freq=freq_map[result['type']])
else:
data = numpy.array(vals, dtype=numpyType)
-
+
if result['freq'] == HCASEX:
pyObj = ma.array(data, mask=mask)
else:
@@ -405,7 +416,7 @@
value=result['startindex']+date_value_adjust[freq])
else:
start_date = None
-
+
pyObj = ts.time_series(data, freq=freq,
start_date=start_date,
observed=observed, mask=mask)
@@ -414,7 +425,7 @@
if isSingle:
return items.values()[0]
-
+
return items
#..............................................................................
def write_dict(self, objdict,
@@ -554,7 +565,7 @@
- `end_date` (Date, *[None]*) : If None, data will be written until the end of
`tser`. If specified, only data points on or before end_date will be written.
"""
-
+
if not isinstance(tser, ts.TimeSeries):
raise ValueError("tser is not a valid time series")
elif tser.has_missing_dates():
@@ -569,15 +580,15 @@
if overwrite or not exists: create = True
else: create = False
-
+
fame_params = _fame_params_from_pyobj_tser(tser)
-
+
fame_cls = fame_params['cls']
fame_type = fame_params['type']
fame_freq = fame_params['freq']
fame_basis = fame_params['basis']
fame_observed = fame_params['observed']
-
+
if create:
if exists: self.delete_obj(name)
cf_create(self.dbkey, name, fame_cls, fame_freq, fame_type, fame_basis, fame_observed)
@@ -591,10 +602,10 @@
return bdate
else:
return getattr(tser, attr)
-
+
start_date = get_boundary_date(start_date, "start_date")
end_date = get_boundary_date(end_date, "end_date")
-
+
if start_date is not None:
towrite = tser[start_date:end_date+1]
@@ -645,7 +656,7 @@
- `end_case` (int, *[None]*) : If None, data will be written until the end of
`cser`. If specified, only data points on or before end_case will be written.
"""
-
+
if not isinstance(cser, numpy.ndarray):
raise ValueError("cser is not a valid ndarray")
elif cser.ndim != 1:
@@ -657,9 +668,9 @@
if overwrite or not exists: create = True
else: create = False
-
+
fame_params = _fame_params_from_pyobj_cser(cser)
-
+
fame_cls = fame_params['cls']
fame_type = fame_params['type']
fame_freq = fame_params['freq']
@@ -696,15 +707,15 @@
return zero_represents + cser.size - 1
else:
raise ValueError("unexpected argument: %s " % attr)
-
+
start_case = get_boundary_case(start_case, "start_case")
end_case = get_boundary_case(end_case, "end_case")
- if start_case is not None:
+ if start_case is not None:
# convert integer types to floats since FAME does not have an integer type
s = start_case - zero_represents
e = end_case - zero_represents
-
+
fame_data = fame_data[s:e+1]
fame_mask = fame_mask[s:e+1]
newType = fametype_tonumpy(fame_type)
@@ -725,7 +736,7 @@
- `name` (string) : database key that the object will be written to
- `scalar` : one of the following: string, numpy scalar, int, float,
list of strings (for name lists), Date, boolean"""
-
+
fame_params = _fame_params_from_pyobj_scalar(scalar)
fame_type = fame_params['type']
@@ -746,7 +757,7 @@
fame_data = "{" + ", ".join(scalar) + "}"
else:
raise ValueError("Unrecognized data type")
-
+
if self.obj_exists(name): self.delete_obj(name)
cf_create(self.dbkey, name,
fame_params['cls'],
@@ -759,7 +770,7 @@
newType = fametype_tonumpy(fame_type)
if hasattr(fame_data, 'dtype') and newType != fame_data.dtype:
fame_data = fame_data.astype(newType)
-
+
if fame_type == HNAMEL:
cf_write_namelist(self.dbkey, name, fame_data)
else:
@@ -786,10 +797,10 @@
parameters. Generally speaking, it is easier to use initialize_obj
with a prototype object.
"""
-
+
if cls not in (HSERIE, HSCALA):
raise ValueError("unrecognized object class: "+str(cls))
-
+
if freq is None:
if cls == HSCALA:
freq = HUNDFX
@@ -815,7 +826,7 @@
param_func = _fame_params_from_pyobj_cser
else:
param_func = _fame_params_from_pyobj_scalar
-
+
fame_params = param_func(pyobj)
cf_create(self.dbkey, name,
fame_params['cls'],
@@ -827,7 +838,7 @@
def rename_obj(self, name, new_name):
"""rename fame object in database"""
cf_rename_obj(self.dbkey, name, new_name)
-
+
def copy_obj(self, target_db, source_name, target_name=None):
"""copy fame object to another destination"""
if target_name is None: target_name = source_name
@@ -871,9 +882,9 @@
ts_freq = freq_map[obj_sz['freq']]
except KeyError:
raise DBError("unsupported FAME frequency: %i", fame_freq)
-
+
if obj_sz[date_type+'_year'] == -1: return None
-
+
annDate = ts.Date(freq='A', year=obj_sz[date_type+'_year'])
return annDate.asfreq(ts_freq, relation='BEFORE') + (obj_sz[date_type+'_period'] - 1)
@@ -914,16 +925,16 @@
def obj_end_date(self, name):
"""get end_date of a FAME time series object"""
return self.__ser_date(name, 'end')
-
+
def obj_created(self, name):
"get 'created' attribute of object in database"
fame_date = cf_get_obj_attr(self.dbkey, name, "CREATED")
- return _famedate_to_tsdate(fame_date, 's')
+ return _famedate_to_tsdate(fame_date, _c.FR_SEC)
def obj_modified(self, name):
"get 'modified' attribute of object in database"
fame_date = cf_get_obj_attr(self.dbkey, name, "MODIFIED")
- return _famedate_to_tsdate(fame_date, 's')
+ return _famedate_to_tsdate(fame_date, _c.FR_SEC)
#..............................................................................
def db_desc(self):
"get 'description' attribute of database"
@@ -936,12 +947,12 @@
def db_created(self):
"get 'created' attribute of database"
fame_date = cf_get_db_attr(self.dbkey, "CREATED")
- return _famedate_to_tsdate(fame_date, 's')
+ return _famedate_to_tsdate(fame_date, _c.FR_SEC)
def db_modified(self):
"get 'modified' attribute of database"
fame_date = cf_get_db_attr(self.dbkey, "MODIFIED")
- return _famedate_to_tsdate(fame_date, 's')
+ return _famedate_to_tsdate(fame_date, _c.FR_SEC)
def db_is_open(self):
"returns True if database is open. False otherwise"
@@ -959,7 +970,7 @@
"""performs a wildlist lookup on the database, using Fame syntax
("?" and "^"), returns a normal python list of strings"""
res = cf_wildlist(self.dbkey, exp)
-
+
if wildonly:
exp = exp.replace("?", "(.*)")
exp = exp.replace("^", "(.)")
@@ -973,7 +984,7 @@
"""Closes the database. Changes will be posted."""
if self.db_is_open():
cf_close(self.dbkey)
-
+
def post(self):
cf_post(self.dbkey)
@@ -1002,7 +1013,7 @@
except:
self.fameLock.release()
raise
-
+
return result
cf_open = cFameCall(cfame.open)
@@ -1041,7 +1052,7 @@
:Parameters:
- option (str) : name of the option to set
- setting (str) : value of the option to set
-
+
:Example:
set_option("DBSIZE", "LARGE")
"""
@@ -1049,5 +1060,5 @@
def license_expires():
"""get date that license expires on"""
fame_date = cf_license_expires()
- adj_val = date_value_adjust[ts.freq_fromstr('D')]
- return ts.Date(freq='D', value=fame_date+adj_val)
+ adj_val = date_value_adjust[_c.FR_DAY]
+ return ts.Date(freq=_c.FR_DAY, value=fame_date+adj_val)
Modified: trunk/Lib/sandbox/timeseries/io/fame/tests/test_fame.py
===================================================================
--- trunk/Lib/sandbox/timeseries/io/fame/tests/test_fame.py 2007-03-21 23:05:27 UTC (rev 2864)
+++ trunk/Lib/sandbox/timeseries/io/fame/tests/test_fame.py 2007-03-22 13:46:15 UTC (rev 2865)
@@ -19,6 +19,7 @@
from timeseries.io import fame
from timeseries.io.fame import const
+from timeseries import const as _c
from timeseries import Report
import timeseries as ts
import maskedarray as ma
@@ -85,15 +86,15 @@
data['scalars']['boolean'] = True
for f in data['dates']:
data['scalars']['date_'+f] = data['dates'][f]
-
+
_desc = "my desc\nline 2"
_doc = "my doc\nline 2"
class test_write(NumpyTestCase):
-
+
def setUp(self):
self.db = fame.FameDb("testdb.db",'o')
-
+
def test_main(self):
"execute all the tests. Order is important here"
@@ -127,7 +128,7 @@
"test writing all types of scalar values"
for s in data['scalars']:
self.db.write_scalar('$scalar_'+s, data['scalars'][s])
-
+
def _test_dict_scalars(self):
"test writing multiple scalars at once using write_scalar_dict"
self.db.write_scalar_dict({'$scalar_1':data['scalars']['float32'],
@@ -175,10 +176,10 @@
ser = self.db.read('$tser_'+x)
if str(ser.dtype)[:5] == 'float' and str(data['tser'][x].dtype)[:3] == 'int':
ser = ser.astype(data['tser'][x].dtype)
-
+
assert_mask_equal(ser.mask, data['tser'][x].mask)
assert((ser == data['tser'][x]).all())
-
+
def _test_read_range_tser(self):
"test reading a time series over specified ranges"
src = data['tser']['float32']
@@ -186,32 +187,32 @@
s2 = src.start_date-2
e1 = src.end_date+2
e2 = src.end_date-2
-
+
dateList = [(s1, e1),
(s1, e2),
(s2, e1),
(s2, e2)]
-
+
for s, e in dateList:
res = ts.adjust_endpoints(src, start_date=s, end_date=e)
ser = self.db.read('$tser_float32', start_date=s, end_date=e)
assert_array_equal(res, ser)
-
+
def _test_write_append_tser(self):
"test appending data to an existing time series"
self.db.write_tser('$appendTSer', data['tser']['float32'])
self.db.write_tser('$appendTSer', appendTSer)
-
+
def _test_read_append_tser(self):
"test reading of appended time series"
result = ts.adjust_endpoints(data['tser']['float32'],
start_date=data['tser']['float32'].start_date,
end_date=appendTSer.end_date)
result[appendTSer.start_date:appendTSer.end_date+1] = appendTSer
-
+
ser = self.db.read('$appendTSer')
-
+
assert_array_equal(result, ser)
@@ -223,12 +224,12 @@
def _test_verify_write_range_tser(self):
"verify that _test_write_range_write_tser worked as expected"
-
+
ser = self.db.read('$rangeTSer')
-
+
sDate = ts.Date(freq='A', year=2008)
eDate = ts.Date(freq='A', year=2012)
-
+
assert_array_equal(ser, rangeTSer[sDate:eDate+1])
def _test_write_empty_tser(self):
@@ -240,7 +241,7 @@
"test reading a time series with no data"
ser = self.db.read('$emptyTSer')
assert(ser.start_date is None)
-
+
def _test_overwrite_tser(self):
"test overwriting a time series"
self.db.write_tser('$tser_float32', data['tser']['bool'], overwrite=True)
@@ -255,7 +256,7 @@
except fame.DBError:
exception = True
assert(exception)
-
+
def _test_dict_tser(self):
"test writing multiple time series at once using write_tser_dict"
self.db.write_tser_dict({'$tser_1':data['tser']['float32'],
@@ -275,7 +276,7 @@
ser = self.db.read('$cser_'+x)
if str(ser.dtype)[:5] == 'float' and str(data['cser'][x].dtype)[:3] == 'int':
ser = ser.astype(data['cser'][x].dtype)
-
+
assert_mask_equal(ser.mask, data['cser'][x].mask)
assert((ser == data['cser'][x]).all())
@@ -286,12 +287,12 @@
s2 = 1
e1 = 8
e2 = 4
-
+
caseList = [(s1, e1),
(s1, e2),
(s2, e1),
(s2, e2)]
-
+
for s, e in caseList:
size = (e - s + 1)
res = ma.array([0]*size , np.float32, mask=[1]*size )
@@ -308,14 +309,14 @@
"test appending to an existing case series"
self.db.write_cser('$appendCSer', data['cser']['float32'])
self.db.write_cser('$appendCSer', appendCSer, zero_represents=4)
-
+
def _test_read_append_cser(self):
"test reading of appended case series"
-
+
result = ma.concatenate([data['cser']['float32'][:3], appendCSer])
ser = self.db.read('$appendCSer')
assert_array_equal(result, ser)
-
+
def _test_write_range_cser(self):
"test writing over a specified range"
self.db.write_cser('$rangeCSer', rangeCSer,
@@ -323,11 +324,11 @@
def _test_verify_write_range_cser(self):
"verify that _test_write_range_write_cser worked as expected"
-
+
ser = self.db.read('$rangeCSer')
result = ma.arange(9).astype(np.float32)
result[:4] = ma.masked
-
+
assert_array_equal(ser, result)
def _test_write_empty_cser(self):
@@ -338,13 +339,13 @@
"test reading a case series with no data"
ser = self.db.read('$emptyCSer')
assert_equal(ser.size, 0)
-
+
def _test_overwrite_cser(self):
"test overwriting a case series"
self.db.write_cser('$cser_float32', data['cser']['bool'], overwrite=True)
ser = self.db.read('$cser_float32')
assert_array_equal(ser, data['cser']['bool'])
-
+
def _test_assume_exists_cser(self):
"check to see if the assume_exists flag works for write_cser"
exception = False
@@ -361,17 +362,17 @@
result = self.db.read(['$cser_1', '$cser_2'])
assert_array_equal(result['$cser_1'], data['cser']['float32'])
assert_array_equal(result['$cser_2'], data['cser']['float32'])
-
+
def _test_whats(self):
"test whats method"
# just make sure it doesn't crash for now
what_dict = self.db._whats('$tser_float32')
-
+
def _test_exists(self):
"test exists method"
assert(self.db.obj_exists('$cser_float32'))
assert(not self.db.obj_exists('$fake_series'))
-
+
def _test_delete(self):
"test delete method"
assert(self.db.obj_exists('$cser_1'))
@@ -389,12 +390,12 @@
res2 = sorted([x.upper() for x in list(data['cser'])])
assert_equal(wl1, res1)
assert_equal(wl2, res2)
-
+
def _test_restore(self):
"test restore method"
self.db.close()
self.db = fame.FameDb("testdb.db",'s')
-
+
self.db.delete_obj('$tser_float32')
assert(not self.db.obj_exists('$tser_float32'))
self.db.restore()
@@ -409,12 +410,12 @@
modified = self.db.db_modified()
desc = self.db.db_desc()
doc = self.db.db_doc()
-
+
assert(abs(ts.thisday('s') - created) < 100)
assert(abs(ts.thisday('s') - modified) < 100)
assert_equal(desc, _desc)
assert_equal(doc, _doc)
-
+
assert(self.db.db_is_open())
self.db.close()
assert(not self.db.db_is_open())
@@ -432,13 +433,13 @@
proc.close()
assert_equal(exists, "False")
-
+
self.db.post()
-
+
proc = os.popen('python -c "'+exist_script+'"')
exists = proc.readlines()[0].strip('\n')
proc.close()
-
+
assert_equal(exists, "True")
def _test_copy_rename(self):
@@ -448,17 +449,17 @@
orig_obj = self.db.read("$tser_float32")
copied_obj = db2.read("$copied_obj")
assert_array_equal(orig_obj, copied_obj)
-
+
db2.rename_obj("$copied_obj", "$renamed_obj")
assert(db2.obj_exists("$renamed_obj"))
assert(not db2.obj_exists("$copied_obj"))
-
+
db2.close()
-
+
def _test_obj_attribs(self):
"test getting and setting object attributes"
assert_equal(self.db.obj_freq("$freq_b"), data['freqs']['b'].freq)
-
+
assert_equal(self.db.obj_start_date("$freq_b"),
data['freqs']['b'].start_date)
assert_equal(self.db.obj_end_date("$freq_b"),
@@ -469,24 +470,24 @@
assert(abs(ts.thisday('s') - created) < 100)
assert(abs(ts.thisday('s') - modified) < 100)
-
+
self.db.set_obj_desc("$freq_b", _desc)
self.db.set_obj_doc("$freq_b", _doc)
-
+
desc = self.db.obj_desc("$freq_b")
doc = self.db.obj_doc("$freq_b")
-
+
assert_equal(desc, _desc)
assert_equal(doc, _doc)
-
+
self.db.set_obj_basis("$freq_b", const.HBSDAY)
assert_equal(self.db.obj_basis("$freq_b"), const.HBSDAY)
self.db.set_obj_basis("$freq_b", const.HBSBUS)
assert_equal(self.db.obj_basis("$freq_b"), const.HBSBUS)
-
+
self.db.set_obj_observed("$freq_b", "END")
assert_equal(self.db.obj_observed("$freq_b"), "ENDING")
-
+
self.db.set_obj_observed("$freq_b", "MAX")
assert_equal(self.db.obj_observed("$freq_b"), "MAXIMUM")
@@ -495,16 +496,16 @@
def _test_misc_funcs(self):
"test FAME functions that aren't database methods"
- assert_equal(fame.license_expires().freq, ts.freq_fromstr('D'))
-
+ assert_equal(fame.license_expires().freq, _c.FR_DAY)
+
# just test that this doesn't crash for now
fame.set_option("DBSIZE", "LARGE")
def tearDown(self):
self.db.close()
-
-
+
+
###############################################################################
#------------------------------------------------------------------------------
if __name__ == "__main__":
More information about the Scipy-svn
mailing list