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

.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,10 @@
 <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">
+    <option name="renderExternalDocumentation" value="true" />
+  </component>
 </module>

average.py

@@ -0,0 +1,41 @@
+from algorithm import algorithm
+import json
+import ray
+
+@ray.remote
+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.sum = 0
+        self.average = 0
+
+
+    def set_config(self, config):
+        self.config_ = config
+        self.config_dict_ = json.loads(self.config_)
+        self.average_number_ = self.config_dict_["AVERAGE_NUMBER"]
+
+    def config(self):
+        return self.config_
+
+    def eval(self, value):
+        self.present_number_ = len(self.window_1_)
+        if self.present_number_ < self.average_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_)
+            return self.average
+        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_)
+            return self.average

average_func_test.py

@@ -0,0 +1,10 @@
+from algorithm import algorithm
+import json
+import ray
+import numpy as np
+a = np.zeros(shape=(5,2))
+b = []
+print(a)
+print(type(b))
+print(len(b))
+

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_)

cycle.py

@@ -1,8 +1,9 @@
 from algorithm import algorithm
 import json
 import math
+import ray
 
-
+@ray.remote
 class cycle(algorithm):
     def __init__(self):
         self.config_dict_ = None

cycle_test_csv.py

@@ -1,21 +1,29 @@
 import pandas as pd
 from cycle import cycle
+import ray
 
-filepath = "D:/python_project/1.csv"
+ray.init()
+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]
 disturb_data = origin_data
-algorithm_step = cycle()
+algorithm_step = cycle.remote()
 
 cow_name = "G1.TTXD1_3"
-algorithm_step.set_config('{"CYCLE_AMPLITUDE_BASE": 200 , "CYCLE_TIME_BASE": 5 ,"CYCLE_START_ANGLE_BASE": 3.14}')
+contrast_data = pd.DataFrame()
+contrast_data[cow_name+'_origin'] = origin_data[cow_name]
+algorithm_step.set_config.remote('{"CYCLE_AMPLITUDE_BASE": 200 , "CYCLE_TIME_BASE": 5 ,"CYCLE_START_ANGLE_BASE": 3.14}')
 # algorithm_step.set_config('{"CYCLE_TIME_BASE": 5 }')
 for i in range(0,row_len):
-    disturb_data.loc[i,cow_name]=algorithm_step.eval(origin_data.loc[i,cow_name],i)
-print(disturb_data)
-disturb_data.to_csv("D:/python_project/1_disturb.csv", index=False)
+    futures=algorithm_step.eval.remote(origin_data.loc[i,cow_name],i)
+    disturb_data.loc[i,cow_name]=ray.get(futures)
+
+print(disturb_data.loc[:,cow_name])
+contrast_data[cow_name+'_disturb'] = disturb_data[cow_name]
+disturb_data.to_csv("D:/python_project_data/1_disturb.csv", index=False)
+contrast_data.to_csv("D:/python_project_data/contrast_data.csv", index=False)
 
 
 # 以下均为测试性能用
-print(algorithm_step.config_)
+# 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=",")

excursion.py

@@ -1,7 +1,8 @@
 from algorithm import algorithm
 import json
+import ray
 
-
+@ray.remote
 class excursion(algorithm):
     def __init__(self):
         self.config_dict_ = None

ray_example.py

@@ -0,0 +1,18 @@
+import ray
+ray.init()
+
+@ray.remote
+class Counter(object):
+    def __init__(self):
+        self.n = 0
+
+    def increment(self):
+        self.n += 1
+
+    def read(self):
+        return self.n
+
+counters = [Counter.remote() for i in range(4)]
+[c.increment.remote() for c in counters]
+futures = [c.read.remote() for c in counters]
+print(ray.get(futures))

step.py

@@ -1,7 +1,8 @@
 from algorithm import algorithm
 import json
+import ray
 
-
+@ray.remote
 class step(algorithm):
     def __init__(self):
         self.config_dict_ = None

step_test.py

@@ -1,14 +1,17 @@
 import pandas as pd
 from step import step
+import ray
+ray.init()
 
-filepath = "D:/python_project/TurbineExhaustTemperatureTheromcouple-FourWorkingCondition.xlsx"
-origin_data = pd.read_excel(filepath,sheet_name='s4_ttxd')
+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]
 disturb_data = origin_data
-algorithm_step = step()
-cow_name = "ttxd_12_1"
-algorithm_step.set_config('{"STEP_AMPLITUDE_BASE": 1.3}')
+algorithm_step = step.remote()
+cow_name = "G1.TTXD1_3"
+algorithm_step.set_config.remote('{"STEP_AMPLITUDE_BASE": 1.3}')
 for i in range(0,row_len):
-    disturb_data.loc[i,cow_name]=algorithm_step.eval(origin_data.loc[i,cow_name])
-print(disturb_data)
+    futures=algorithm_step.eval.remote(origin_data.loc[i,cow_name])
+    disturb_data.loc[i, cow_name] = ray.get(futures)
+print(disturb_data.loc[:,cow_name])

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_)

white_noise.py

@@ -2,8 +2,9 @@ from random import random
 import numpy as np
 from algorithm import algorithm
 import json
+import ray
 
-
+@ray.remote
 class white_noise(algorithm):
     def __init__(self):
         self.config_dict_ = None