写了一个白噪声干扰算法的实际应用案例（指读取文件然后随机选列最后在规定位置输出）

.idea/misc.xml

@@ -1,4 +1,4 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
-  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.8" project-jdk-type="Python SDK" />
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.8 (python_vitual_3.85)" project-jdk-type="Python SDK" />
 </project>

.idea/python_project.iml

@@ -2,7 +2,7 @@
 <module type="PYTHON_MODULE" version="4">
   <component name="NewModuleRootManager">
     <content url="file://$MODULE_DIR$" />
-    <orderEntry type="inheritedJdk" />
+    <orderEntry type="jdk" jdkName="Python 3.8 (python_vitual_3.85)" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
   <component name="PyDocumentationSettings">

average.py

@@ -3,14 +3,16 @@ import json
 import ray
 
 @ray.remote
-class step(algorithm):
+class average(algorithm):
     def __init__(self):
         self.config_dict_ = None
         self.config_ = None
         self.average_number_ = 10
         self.present_number_ = 0
         self.window_1_ = []
-        self.window_2_ = []
+        self.sum = 0
+        self.average = 0
+
 
     def set_config(self, config):
         self.config_ = config
@@ -24,7 +26,16 @@ class step(algorithm):
         self.present_number_ = len(self.window_1_)
         if self.present_number_ < self.average_number_:
             self.window_1_.append(value)
-            self.window_2_.append(value)
-            return
+            self.sum = 0
+            for i in self.window_1_:
+                self.sum = self.sum + i
+            self.average = self.sum / len(self.window_1_)
+            return self.average
         else:
-            return value * self.amplitude_base_
+            self.sum = 0
+            del self.window_1_[0]
+            self.window_1_.append(value)
+            for i in self.window_1_:
+                self.sum = self.sum + i
+            self.average = self.sum / len(self.window_1_)
+            return self.average

average_test_csv.py

@@ -0,0 +1,31 @@
+import pandas as pd
+from average import average
+import ray
+
+ray.init()
+
+ray.RAY_DISABLE_MEMORY_MONITOR=1
+filepath = "D:/python_project_data/1.csv"
+origin_data = pd.read_csv(filepath)
+row_len = origin_data.shape[0]
+cow_len = origin_data.shape[1]
+average_data = origin_data
+algorithm_average = average.remote()
+
+cow_name = "G1.TTXD1_3"
+contrast_data = pd.DataFrame()
+contrast_data[cow_name+'_origin'] = origin_data[cow_name]
+algorithm_average.set_config.remote('{"AVERAGE_NUMBER": 5 }')
+# algorithm_step.set_config('{"CYCLE_TIME_BASE": 5 }')
+for i in range(0,row_len):
+    futures=algorithm_average.eval.remote(origin_data.loc[i,cow_name])
+    average_data.loc[i, cow_name]=ray.get(futures)
+
+print(average_data.loc[:, cow_name])
+contrast_data[cow_name+'_average'] = average_data[cow_name]
+# average_data.to_csv("D:/python_project_data/1_disturb.csv", index=False)
+contrast_data.to_csv("D:/python_project_data/average_data.csv", index=False)
+
+
+# 以下均为测试性能用
+# print(algorithm_step.config_)

average_x.py

@@ -0,0 +1,29 @@
+from algorithm import algorithm
+import json
+import ray
+
+@ray.remote
+class average_x(algorithm):
+    def __init__(self):
+        self.config_dict_ = None
+        self.config_ = None
+        self.average_number_ = 10
+        self.present_number_ = 0
+        self.avg_ = 0
+
+    def set_config(self, config):
+        self.config_ = config
+        self.config_dict_ = json.loads(self.config_)
+        self.average_number_ = self.config_dict_["AVERAGE_NUMBER"]
+        self.init_flag_ = False
+
+    def config(self):
+        return self.config_
+
+    def eval(self, value):
+        if self.init_flag_ == False:
+            self.init_flag_ = True
+            self.avg_ = value
+        else:
+            self.avg_ = self.avg_ * (self.average_number_ - 1)/self.average_number_ + value / self.average_number_
+        return self.avg_

average_x_test_csv.py

@@ -0,0 +1,31 @@
+import pandas as pd
+from average_x import average_x
+import ray
+
+ray.init()
+
+ray.RAY_DISABLE_MEMORY_MONITOR=1
+filepath = "D:/python_project_data/1.csv"
+origin_data = pd.read_csv(filepath)
+row_len = origin_data.shape[0]
+cow_len = origin_data.shape[1]
+average_data = origin_data
+algorithm_average = average_x.remote()
+
+cow_name = "G1.TTXD1_3"
+contrast_data = pd.DataFrame()
+contrast_data[cow_name+'_origin'] = origin_data[cow_name]
+algorithm_average.set_config.remote('{"AVERAGE_NUMBER": 5 }')
+# algorithm_step.set_config('{"CYCLE_TIME_BASE": 5 }')
+for i in range(0,row_len):
+    futures=algorithm_average.eval.remote(origin_data.loc[i,cow_name])
+    average_data.loc[i, cow_name]=ray.get(futures)
+
+print(average_data.loc[:, cow_name])
+contrast_data[cow_name+'_average_x'] = average_data[cow_name]
+# average_data.to_csv("D:/python_project_data/1_disturb.csv", index=False)
+contrast_data.to_csv("D:/python_project_data/average_x_data.csv", index=False)
+
+
+# 以下均为测试性能用
+# print(algorithm_step.config_)

disturb_data_to_csv.py

@@ -0,0 +1,40 @@
+import pandas as pd
+import random
+import numpy
+from white_noise import white_noise
+import ray
+
+ray.init()
+filepath = "E:/Data_sum/ttxd200.csv"
+origin_data = pd.read_csv(filepath)
+row_len = origin_data.shape[0]
+cow_len = origin_data.shape[1]
+columns_index=origin_data.columns
+columns_list=columns_index.tolist()
+
+disturb_data = origin_data
+real_origin_data = origin_data
+algorithm_white_noise = white_noise.remote()
+algorithm_white_noise.set_config.remote('{"WHITE_NOISE_STANDARD_DEVIATION_BASE": 30}')
+
+#cow_name = "ttxd_12_1"
+contrast_data = pd.DataFrame()
+
+
+for i in range(0,row_len-50,10):
+    col_number = random.randint(0,cow_len)
+    cow_name = columns_list[col_number]
+    for j in range(i,i+50):
+        futures = algorithm_white_noise.eval.remote(origin_data.loc[j, cow_name])
+        disturb_data.loc[i, cow_name] = ray.get(futures)
+    contrast_data[cow_name+'_disturb'+str(i)] = disturb_data[cow_name]
+    disturb_data = real_origin_data
+    origin_data = real_origin_data
+
+# print(columns_list)
+contrast_data.to_csv("E:/Data_sum/ttxd200_disturb.csv", index=False)
+contrast_data_only = contrast_data.values
+# contrast_data_only.to_csv("E:/Data_sum/ttxd200_disturb_data.csv", index=False)
+print(contrast_data_only)
+# pd.DataFrame(contrast_data_only).to_csv("E:/Data_sum/ttxd200_disturb_data.csv", index=False)
+numpy.savetxt("E:/Data_sum/ttxd200_disturb_data.csv", contrast_data_only, delimiter=",")

variance.py

@@ -0,0 +1,51 @@
+from algorithm import algorithm
+import json
+import ray
+
+@ray.remote
+class variance(algorithm):
+    def __init__(self):
+        self.config_dict_ = None
+        self.config_ = None
+        self.variance_number_ = 10
+        self.present_number_ = 0
+        self.window_1_ = []
+        self.sum = 0
+        self.average = 0
+        self.variance = 0
+        self.variance_sum = 0
+
+    def set_config(self, config):
+        self.config_ = config
+        self.config_dict_ = json.loads(self.config_)
+        self.variance_number_ = self.config_dict_["VARIANCE_NUMBER"]
+
+    def config(self):
+        return self.config_
+
+    def eval(self, value):
+        self.present_number_ = len(self.window_1_)
+        if self.present_number_ < self.variance_number_:
+            self.window_1_.append(value)
+            self.sum = 0
+            for i in self.window_1_:
+                self.sum = self.sum + i
+            self.average = self.sum / len(self.window_1_)
+            self.variance_sum = 0
+            for i in self.window_1_:
+                self.variance_sum = self.variance_sum + (i-self.average)*(i-self.average)
+            self.variance = self.variance_sum / len(self.window_1_)
+            return self.variance
+        else:
+            self.sum = 0
+            del self.window_1_[0]
+            self.window_1_.append(value)
+            for i in self.window_1_:
+                self.sum = self.sum + i
+            self.average = self.sum / len(self.window_1_)
+            self.variance_sum = 0
+            for i in self.window_1_:
+                self.variance_sum = self.variance_sum + (i-self.average)*(i-self.average)
+            self.variance = self.variance_sum / len(self.window_1_)
+            return self.variance
+

variance_test_csv.py

@@ -0,0 +1,31 @@
+import pandas as pd
+from variance import variance
+import ray
+
+ray.init()
+
+ray.RAY_DISABLE_MEMORY_MONITOR=1
+filepath = "D:/python_project_data/1.csv"
+origin_data = pd.read_csv(filepath)
+row_len = origin_data.shape[0]
+cow_len = origin_data.shape[1]
+variance_data = origin_data
+algorithm_variance = variance.remote()
+
+cow_name = "G1.TTXD1_3"
+contrast_data = pd.DataFrame()
+contrast_data[cow_name+'_origin'] = origin_data[cow_name]
+algorithm_variance.set_config.remote('{"VARIANCE_NUMBER": 5 }')
+# algorithm_step.set_config('{"CYCLE_TIME_BASE": 5 }')
+for i in range(0,row_len):
+    futures=algorithm_variance.eval.remote(origin_data.loc[i,cow_name])
+    variance_data.loc[i, cow_name]=ray.get(futures)
+
+print(variance_data.loc[:, cow_name])
+contrast_data[cow_name+'_variance'] = variance_data[cow_name]
+# average_data.to_csv("D:/python_project_data/1_disturb.csv", index=False)
+contrast_data.to_csv("D:/python_project_data/variance_data.csv", index=False)
+
+
+# 以下均为测试性能用
+# print(algorithm_step.config_)

variance_x.py

@@ -0,0 +1,33 @@
+from algorithm import algorithm
+import json
+import ray
+
+# @ray.remote
+@ray.remote
+class variance_x(algorithm):
+    def __init__(self):
+        self.config_dict_ = None
+        self.config_ = None
+        self.window_length_ = 10
+        self.avg_ = 0
+        self.var_ = 0
+
+    def set_config(self, config):
+        self.config_ = config
+        self.config_dict_ = json.loads(self.config_)
+        self.window_length_ = self.config_dict_["WINDOW_LENGTH"]
+        self.init_flag_ = False
+
+    def config(self):
+        return self.config_
+
+    def eval(self, value):
+        if self.init_flag_ == False:
+            self.init_flag_ = True
+            self.avg_ = value
+            self.var_ = 0
+        else:
+            present_avg_ = self.avg_ * (self.window_length_ - 1)/self.window_length_ + value / self.window_length_
+            self.var_ = (self.window_length_ - 1)/self.window_length_ * self.var_ + (self.window_length_ - 1) / (self.window_length_ ** 2) * ((value - self.avg_)**2)
+            self.avg_ = present_avg_
+        return self.var_

variance_x_test_csv.py

@@ -0,0 +1,31 @@
+import pandas as pd
+from variance_x import variance_x
+import ray
+
+ray.init()
+
+RAY_DISABLE_MEMORY_MONITOR=1
+filepath = "D:/python_project_data/1.csv"
+origin_data = pd.read_csv(filepath)
+row_len = origin_data.shape[0]
+cow_len = origin_data.shape[1]
+variance_data = origin_data
+algorithm_variance_x = variance_x.remote()
+
+cow_name = "G1.TTXD1_3"
+contrast_data = pd.DataFrame()
+contrast_data[cow_name+'_origin'] = origin_data[cow_name]
+algorithm_variance_x.set_config.remote('{"WINDOW_LENGTH": 5 }')
+# algorithm_step.set_config('{"CYCLE_TIME_BASE": 5 }')
+for i in range(0,row_len):
+    futures=algorithm_variance_x.eval.remote(origin_data.loc[i,cow_name])
+    variance_data.loc[i, cow_name]=ray.get(futures)
+
+print(variance_data.loc[:, cow_name])
+contrast_data[cow_name+'_variance_x'] = variance_data[cow_name]
+# average_data.to_csv("D:/python_project_data/1_disturb.csv", index=False)
+contrast_data.to_csv("D:/python_project_data/variance_data_x.csv", index=False)
+
+ray.shutdown()
+# 以下均为测试性能用
+# print(algorithm_step.config_)