Real-world Use Cases for DevOps: Practical Linux Scenarios

# Python: PHI data processing with encryption import hashlib import base64 from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC from fhir.resources.patient import Patient import re class PHIProcessor: def __init__(self, encryption_key): self.cipher = Fernet(encryption_key) def detect_phi(self, text): """Detect PHI in text using regex patterns""" phi_patterns = { 'ssn': r'\b\d{3}[-]?\d{2}[-]?\d{4}\b', 'phone': r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', 'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'medical_record': r'\bMRN[-]?\d{6,}\b', 'date_of_birth': r'\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b' } detected_phi = {} for phi_type, pattern in phi_patterns.items(): matches = re.findall(pattern, text) if matches: detected_phi[phi_type] = matches return detected_phi def encrypt_phi(self, phi_data): """Encrypt PHI data""" if isinstance(phi_data, str): return self.cipher.encrypt(phi_data.encode()).decode() elif isinstance(phi_data, dict): return {k: self.cipher.encrypt(str(v).encode()).decode() for k, v in phi_data.items()} def deidentify_fhir_resource(self, fhir_resource): """De-identify FHIR resource according to HIPAA Safe Harbor""" deidentified = fhir_resource.copy() # Remove direct identifiers (Safe Harbor method) if 'identifier' in deidentified: for identifier in deidentified['identifier']: if identifier.get('system') in ['http://hl7.org/fhir/sid/us-ssn', 'http://hl7.org/fhir/sid/us-medicare']: identifier['value'] = self.hash_identifier(identifier['value']) # Handle patient resource specifically if fhir_resource['resourceType'] == 'Patient': # Remove names (keep first initial for matching) if 'name' in deidentified: for name in deidentified['name']: if 'given' in name: name['given'] = [n[0] + '***' for n in name['given']] if 'family' in name: name['family'] = name['family'][0] + '***' # Remove address details if 'address' in deidentified: for address in deidentified['address']: address['line'] = ['***'] address['city'] = '***' address['postalCode'] = address['postalCode'][:3] + '**' if len(address['postalCode']) >= 5 else '***' # Keep age but obscure birth date if 'birthDate' in deidentified: # Convert to age in years birth_date = datetime.strptime(deidentified['birthDate'], '%Y-%m-%d') age = (datetime.now() - birth_date).days // 365 deidentified['age'] = f"{age // 10 * 10}+" # Round to nearest 10 years return deidentified def hash_identifier(self, identifier, salt=None): """Create irreversible hash of identifier for linking""" if not salt: salt = os.urandom(16) kdf = PBKDF2HMAC( algorithm=hashes.SHA256(), length=32, salt=salt, iterations=100000, ) key = base64.urlsafe_b64encode(kdf.derive(identifier.encode())) return f"{base64.b64encode(salt).decode()}:{key.decode()}"

# predictive-maintenance-pipeline.py import json import numpy as np import pandas as pd from datetime import datetime, timedelta from sklearn.ensemble import RandomForestClassifier from sklearn.preprocessing import StandardScaler import joblib import boto3 import psycopg2 from kafka import KafkaConsumer, KafkaProducer import pickle class PredictiveMaintenance: def __init__(self): self.model = None self.scaler = StandardScaler() self.s3_client = boto3.client('s3') self.model_bucket = 'iot-ml-models' def load_training_data(self, device_type, lookback_days=90): """Load historical device data for training""" conn = psycopg2.connect( host=os.getenv('DB_HOST'), database=os.getenv('DB_NAME'), user=os.getenv('DB_USER'), password=os.getenv('DB_PASSWORD') ) query = f""" SELECT device_id, temperature, vibration_x, vibration_y, vibration_z, power_consumption, operating_hours, error_codes, maintenance_required, timestamp FROM device_metrics WHERE device_type = %s AND timestamp >= NOW() - INTERVAL '%s days' AND maintenance_required IS NOT NULL ORDER BY device_id, timestamp """ df = pd.read_sql_query(query, conn, params=(device_type, lookback_days)) conn.close() return df def preprocess_data(self, df): """Preprocess IoT sensor data for ML""" # Feature engineering df['vibration_magnitude'] = np.sqrt( df['vibration_x']**2 + df['vibration_y']**2 + df['vibration_z']**2 ) # Create rolling statistics df['temp_rolling_avg'] = df.groupby('device_id')['temperature'].transform( lambda x: x.rolling(window=24, min_periods=1).mean() ) df['vibration_std'] = df.groupby('device_id')['vibration_magnitude'].transform( lambda x: x.rolling(window=24, min_periods=1).std() ) # Time-based features df['operating_streak'] = df.groupby('device_id')['operating_hours'].diff() df['hours_since_maintenance'] = df.groupby('device_id').cumcount() # Remove NaN values df = df.dropna() return df def train_model(self, device_type): """Train predictive maintenance model""" print(f"Training model for {device_type}...") # Load and preprocess data df = self.load_training_data(device_type) df = self.preprocess_data(df) # Prepare features and labels features = [ 'temperature', 'vibration_magnitude', 'power_consumption', 'operating_hours', 'temp_rolling_avg', 'vibration_std', 'operating_streak', 'hours_since_maintenance' ] X = df[features].values y = df['maintenance_required'].values # Scale features X_scaled = self.scaler.fit_transform(X) # Train model self.model = RandomForestClassifier( n_estimators=100, max_depth=10, random_state=42, n_jobs=-1 ) self.model.fit(X_scaled, y) # Calculate feature importance importance = pd.DataFrame({ 'feature': features, 'importance': self.model.feature_importances_ }).sort_values('importance', ascending=False) print("Feature importance:") print(importance) # Save model model_data = { 'model': self.model, 'scaler': self.scaler, 'features': features, 'trained_at': datetime.now().isoformat() } model_bytes = pickle.dumps(model_data) self.s3_client.put_object( Bucket=self.model_bucket, Key=f'{device_type}/model.pkl', Body=model_bytes ) print(f"Model saved for {device_type}") return self.model def predict_maintenance(self, device_data): """Predict maintenance needs for a device""" if self.model is None: # Load latest model self.load_model(device_data['device_type']) # Prepare features features = self.preprocess_single_device(device_data) features_scaled = self.scaler.transform([features]) # Make prediction prediction = self.model.predict(features_scaled)[0] probability = self.model.predict_proba(features_scaled)[0][1] return { 'device_id': device_data['device_id'], 'maintenance_predicted': bool(prediction), 'confidence': float(probability), 'timestamp': datetime.now().isoformat(), 'recommended_action': self.get_recommendation(probability, device_data) } def stream_predictions(self): """Real-time streaming predictions""" consumer = KafkaConsumer( 'iot-device-metrics', bootstrap_servers=['kafka:9092'], value_deserializer=lambda x: json.loads(x.decode('utf-8')), group_id='predictive-maintenance' ) producer = KafkaProducer( bootstrap_servers=['kafka:9092'], value_serializer=lambda x: json.dumps(x).encode('utf-8') ) print("Starting real-time prediction stream...") for message in consumer: device_data = message.value try: prediction = self.predict_maintenance(device_data) # Send prediction to alerts topic if maintenance needed if prediction['maintenance_predicted'] and prediction['confidence'] > 0.8: producer.send('maintenance-alerts', value=prediction) print(f"Alert: Device {device_data['device_id']} needs maintenance") # Send to predictions topic for dashboard producer.send('maintenance-predictions', value=prediction) except Exception as e: print(f"Error processing device {device_data['device_id']}: {e}") # Send to DLQ for manual inspection producer.send('predictions-dlq', value={ 'device_data': device_data, 'error': str(e), 'timestamp': datetime.now().isoformat() }) # Main execution if __name__ == "__main__": pm = PredictiveMaintenance() # Train initial models for all device types device_types = ['sensor_v1', 'sensor_v2', 'gateway_v1', 'controller_v1'] for device_type in device_types: pm.train_model(device_type) # Start real-time predictions pm.stream_predictions()