1from __future__ import print_function 

2from builtins import str 

3import os 

4import unittest 

5import shutil 

6import unittest 

7import yaml 

8from fundamentals.utKit import utKit 

9from fundamentals import tools 

10from os.path import expanduser 

11home = expanduser("~") 

12 

13packageDirectory = utKit("").get_project_root() 

14settingsFile = packageDirectory + "/test_settings.yaml" 

15# settingsFile = home + \ 

16# "/git_repos/_misc_/settings/fundamentals/test_settings.yaml" 

17 

18su = tools( 

19 arguments={"settingsFile": settingsFile}, 

20 docString=__doc__, 

21 logLevel="DEBUG", 

22 options_first=False, 

23 projectName=None, 

24 defaultSettingsFile=False 

25) 

26arguments, settings, log, dbConn = su.setup() 

27 

28# SETUP PATHS TO COMMON DIRECTORIES FOR TEST DATA 

29moduleDirectory = os.path.dirname(__file__) 

30pathToInputDir = moduleDirectory + "/input/" 

31pathToOutputDir = moduleDirectory + "/output/" 

32 

33try: 

34 shutil.rmtree(pathToOutputDir) 

35except: 

36 pass 

37# COPY INPUT TO OUTPUT DIR 

38shutil.copytree(pathToInputDir, pathToOutputDir) 

39 

40# Recursively create missing directories 

41if not os.path.exists(pathToOutputDir): 

42 os.makedirs(pathToOutputDir) 

43 

44 

45# xt-setup-unit-testing-files-and-folders 

46# xt-utkit-refresh-database 

47 

48class test_rolling_window_sigma_clip(unittest.TestCase): 

49 

50 def test_rolling_window_sigma_clip_function(self): 

51 import numpy as np 

52 import random 

53 normalDist = np.random.normal(loc=100, scale=3.0, size=[200]) 

54 for i in range(1, 7): 

55 randNum = random.randint(1, 201) 

56 normalDist[randNum] = normalDist[randNum] * 4.5 

57 normalDist = list(normalDist) 

58 

59 from fundamentals.stats import rolling_window_sigma_clip 

60 arrayMask = rolling_window_sigma_clip( 

61 log=log, 

62 array=normalDist, 

63 clippingSigma=2.2, 

64 windowSize=11) 

65 

66 # JUST KEEP UNMASKED VALUES 

67 try: 

68 normalDist = [e for e, m in zip( 

69 normalDist, arrayMask) if m == False] 

70 except: 

71 normalDist = [] 

72 

73 def test_rolling_window_sigma_clip_function_exception(self): 

74 

75 from fundamentals.stats import rolling_window_sigma_clip 

76 try: 

77 this = rolling_window_sigma_clip( 

78 log=log, 

79 settings=settings, 

80 fakeKey="break the code" 

81 ) 

82 this.get() 

83 assert False 

84 except Exception as e: 

85 assert True 

86 print(str(e)) 

87 

88 # x-print-testpage-for-pessto-marshall-web-object 

89 

90 # x-class-to-test-named-worker-function