Python 人工智能实战：智能安防

    class SensorData:
    def __init__(self):
        self.name = ''
        self.data_type = ''
        self.data_value = []
    
    # 获取传感器名称
    def get_sensor_name(self):
        return self.name
    
    # 设置传感器名称
    def set_sensor_name(self, name):
        self.name = name
    
    # 获取传感器数据类型
    def get_data_type(self):
        return self.data_type
    
    # 设置传感器数据类型
    def set_data_type(self, data_type):
        self.data_type = data_type
    
    # 添加传感器数据
    def add_data(self, value):
        if isinstance(value, list):
            for v in value:
                self.data_value.append(v)
        else:
            self.data_value.append(value)
    
    # 获取传感器数据
    def get_data(self):
        return self.data_value
    
    class MonitorSystem:
    def __init__(self):
        self.sensors = {}
        pass
    
    # 添加传感器
    def add_sensor(self, sensor):
        name = sensor.get_sensor_name()
        self.sensors[name] = sensor
    
    # 获取传感器数据
    def get_sensor_data(self, sensor_name):
        return self.sensors[sensor_name].get_data()
    
    if __name__ == '__main__':
    system = MonitorSystem()
    
    # 创建传感器1
    s1 = SensorData()
    s1.set_sensor_name('Sensor1')
    s1.set_data_type('electricity')
    s1.add_data([230, 120])
    system.add_sensor(s1)
    
    # 创建传感器2
    s2 = SensorData()
    s2.set_sensor_name('Sensor2')
    s2.set_data_type('water pressure')
    s2.add_data([1.05, 1.03, 1.02])
    system.add_sensor(s2)
    
    
    print('传感器1：', system.get_sensor_data('Sensor1'))    #[230, 120]
    print('传感器2：', system.get_sensor_data('Sensor2'))   #[1.05, 1.03, 1.02]

    import numpy as np
    
    class DataProcessor:
    def process_raw_data(self, raw_data):
        # 清洗数据
        cleaned_data = self._clean_data(raw_data)
    
        # 归一化数据
        normalized_data = self._normalize_data(cleaned_data)
    
        # 离散化数据
        discrete_data = self._discretize_data(normalized_data)
    
        return discrete_data
    
    @staticmethod
    def _clean_data(data):
        cleaned_data = [x for x in data if x!= None and x >= 0]
        return cleaned_data
    
    @staticmethod
    def _normalize_data(data):
        max_val = max(data)
        min_val = min(data)
        norm_data = [(x - min_val) / (max_val - min_val + 1e-8) for x in data]
        return norm_data
    
    @staticmethod
    def _discretize_data(data):
        num_bins = int((max(data) - min(data)) // 0.1 + 1)
        bins = np.linspace(min(data), max(data), num=num_bins+1, endpoint=True)[1:]
        hist, bin_edges = np.histogram(np.array(data), bins=bins)
        discretized_data = hist / sum(hist).astype(float)
        return discretized_data
    
    class DataStorage:
    def __init__(self):
        self.storage = {}
    
    def store_data(self, device_id, data):
        processor = DataProcessor()
        processed_data = processor.process_raw_data(data)
        self.storage[device_id] = processed_data
    
    def get_stored_data(self, device_id):
        return self.storage[device_id]
    
    if __name__ == '__main__':
    storage = DataStorage()
    storage.store_data('Device1', [230, 120])
    storage.store_data('Device2', [1.05, 1.03, 1.02])
    print('设备1：', storage.get_stored_data('Device1'))      #[0.0, 0.0, 1.0]
    print('设备2：', storage.get_stored_data('Device2'))     #[0.0, 0.0, 1.0, 0.0]

    from sklearn import svm
    from sklearn.model_selection import train_test_split
    from sklearn.metrics import accuracy_score
    
    class ModelTrainer:
    def train_model(self, input_data, output_data, model='svm'):
        if not input_data or not output_data:
            raise ValueError("input_data 或 output_data 为None")
    
        X_train, X_test, y_train, y_test = train_test_split(input_data, output_data, test_size=0.2, random_state=42)
    
        if model=='svm':
            clf = svm.SVC(kernel='linear', C=1.0)
        elif model=='rf':
            from sklearn.ensemble import RandomForestClassifier
            clf = RandomForestClassifier(n_estimators=100, max_depth=2, random_state=42)
        else:
            raise ValueError('Unsupported model type.')
    
        clf.fit(X_train, y_train)
    
        y_pred = clf.predict(X_test)
    
        accu = accuracy_score(y_true=y_test, y_pred=y_pred)
    
        return accu
    
    class RiskAnalyzer:
    def analyze_risk(self, stored_data):
        risk_values = []
    
        for key, values in sorted(stored_data.items()):
            print('分析设备:', key)
    
            timesteps = len(values)
    
            feature_list = ['max'] * timesteps 
            feature_list += ['min'] * timesteps 
    
            features = [[i, j] for i, j in zip(feature_list[:-1], feature_list[1:])]
    
            model_trainer = ModelTrainer()
            score = model_trainer.train_model(features[:timesteps//2], values[:timesteps//2],'svm')
    
            print('\t风险值:', score)
            risk_values.append(score)
    
        avg_risk = sum(risk_values)/len(risk_values)
        print('平均风险值:', avg_risk)
    
        return avg_risk
    
    if __name__ == '__main__':
    analyzer = RiskAnalyzer()
    input_data = {'Device1': [0.0, 0.0, 1.0]}
    output_data = {'Device1': [1]}
    result = analyzer.analyze_risk(input_data)