Pipe-and-Filter

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 동기식 파이프라인 예시
class SyncPipeline:
    def __init__(self):
        self.filters = []
    
    def add_filter(self, filter_func):
        self.filters.append(filter_func)
    
    def execute(self, data):
        result = data
        for filter_func in self.filters:
            result = filter_func(result)
        return result

# 사용 예시
pipeline = SyncPipeline()
pipeline.add_filter(lambda x: x.upper())  # 대문자 변환
pipeline.add_filter(lambda x: x.strip())  # 공백 제거
pipeline.add_filter(lambda x: x.replace(' ', '_'))  # 공백을 언더스코어로 변경

result = pipeline.execute("  hello world  ")
print(result)  # "HELLO_WORLD"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

import asyncio
import queue
import threading

class AsyncPipeline:
    def __init__(self):
        self.filters = []
        self.queues = []
    
    def add_filter(self, filter_func):
        self.filters.append(filter_func)
        self.queues.append(queue.Queue())
    
    def worker(self, filter_func, input_queue, output_queue):
        while True:
            try:
                data = input_queue.get(timeout=1)
                if data is None:  # 종료 신호
                    break
                result = filter_func(data)
                if output_queue:
                    output_queue.put(result)
                input_queue.task_done()
            except queue.Empty:
                continue
    
    def execute_async(self, data_stream):
        # 워커 스레드 시작
        threads = []
        for i, filter_func in enumerate(self.filters):
            input_queue = self.queues[i]
            output_queue = self.queues[i + 1] if i + 1 < len(self.queues) else None
            
            thread = threading.Thread(
                target=self.worker, 
                args=(filter_func, input_queue, output_queue)
            )
            thread.start()
            threads.append(thread)
        
        # 데이터 입력
        for data in data_stream:
            self.queues[0].put(data)
        
        # 스레드 종료
        for i, thread in enumerate(threads):
            self.queues[i].put(None)
            thread.join()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

// Node.js 스트림 기반 파이프라인
const { Transform, Readable, Writable } = require('stream');

class UpperCaseFilter extends Transform {
    _transform(chunk, encoding, callback) {
        // 데이터를 대문자로 변환
        this.push(chunk.toString().toUpperCase());
        callback();
    }
}

class WordCountFilter extends Transform {
    constructor(options) {
        super(options);
        this.wordCount = 0;
    }
    
    _transform(chunk, encoding, callback) {
        // 단어 수 계산
        const words = chunk.toString().split(/\s+/).filter(w => w.length > 0);
        this.wordCount += words.length;
        this.push(`Words: ${this.wordCount}\n`);
        callback();
    }
}

// 파이프라인 구성
const inputStream = new Readable({
    read() {
        this.push('hello world\n');
        this.push('pipe and filter\n');
        this.push(null); // 스트림 종료
    }
});

const outputStream = new Writable({
    write(chunk, encoding, callback) {
        console.log('Output:', chunk.toString());
        callback();
    }
});

// 파이프라인 실행
inputStream
    .pipe(new UpperCaseFilter())
    .pipe(new WordCountFilter())
    .pipe(outputStream);

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365

#!/usr/bin/env python3
"""
Pipe-and-Filter 패턴을 활용한 실시간 주문 처리 시스템 구현

이 예시는 전자상거래 주문 처리를 위한 파이프라인을 보여줍니다.
각 필터는 독립적인 처리 단계를 담당하며, 파이프를 통해 연결됩니다.
"""

import abc
import json
import logging
import time
from typing import Any, Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
import asyncio
from concurrent.futures import ThreadPoolExecutor

# 데이터 모델 정의
@dataclass
class OrderData:
    """주문 데이터 모델"""
    order_id: str
    customer_id: str
    product_id: str
    quantity: int
    price: float
    payment_method: str
    metadata: Dict[str, Any] = None
    
    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}

class ProcessingStatus(Enum):
    """처리 상태 열거형"""
    SUCCESS = "success"
    FAILED = "failed"
    RETRY = "retry"

@dataclass
class ProcessingResult:
    """처리 결과 클래스"""
    status: ProcessingStatus
    data: Optional[OrderData] = None
    error_message: Optional[str] = None
    retry_count: int = 0

# 추상 필터 인터페이스
class Filter(abc.ABC):
    """필터 추상 클래스 - 모든 필터가 구현해야 하는 인터페이스"""
    
    def __init__(self, name: str):
        self.name = name
        self.logger = logging.getLogger(f"Filter.{name}")
        self.processed_count = 0
        self.error_count = 0
    
    @abc.abstractmethod
    async def process(self, data: OrderData) -> ProcessingResult:
        """데이터 처리 메서드 - 각 필터에서 구현"""
        pass
    
    def get_stats(self) -> Dict[str, int]:
        """필터 통계 정보 반환"""
        return {
            "processed": self.processed_count,
            "errors": self.error_count,
            "success_rate": (self.processed_count - self.error_count) / max(1, self.processed_count)
        }

# 구체적인 필터 구현들
class OrderValidationFilter(Filter):
    """주문 검증 필터 - 주문 데이터의 유효성을 검사"""
    
    def __init__(self):
        super().__init__("OrderValidation")
    
    async def process(self, data: OrderData) -> ProcessingResult:
        """주문 데이터 검증 처리"""
        self.logger.info(f"Validating order: {data.order_id}")
        
        try:
            # 필수 필드 검증
            if not all([data.order_id, data.customer_id, data.product_id]):
                raise ValueError("Missing required fields")
            
            # 수량 및 가격 검증
            if data.quantity <= 0 or data.price <= 0:
                raise ValueError("Invalid quantity or price")
            
            # 검증 완료 메타데이터 추가
            data.metadata["validated_at"] = time.time()
            data.metadata["validation_status"] = "passed"
            
            self.processed_count += 1
            return ProcessingResult(ProcessingStatus.SUCCESS, data)
            
        except Exception as e:
            self.error_count += 1
            self.logger.error(f"Validation failed for order {data.order_id}: {e}")
            return ProcessingResult(ProcessingStatus.FAILED, error_message=str(e))

class FraudDetectionFilter(Filter):
    """사기 탐지 필터 - 의심스러운 주문을 탐지"""
    
    def __init__(self):
        super().__init__("FraudDetection")
        self.suspicious_customers = {"CUST999", "CUST666"}  # 시뮬레이션용 블랙리스트
    
    async def process(self, data: OrderData) -> ProcessingResult:
        """사기 탐지 처리"""
        self.logger.info(f"Fraud detection for order: {data.order_id}")
        
        try:
            # 시뮬레이션: 머신러닝 모델 호출 (비동기 처리)
            await asyncio.sleep(0.1)  # 모델 추론 시간 시뮬레이션
            
            # 간단한 규칙 기반 사기 탐지
            risk_score = 0
            
            # 블랙리스트 고객 확인
            if data.customer_id in self.suspicious_customers:
                risk_score += 80
            
            # 고액 주문 확인
            if data.price > 10000:
                risk_score += 30
            
            # 대량 주문 확인
            if data.quantity > 100:
                risk_score += 20
            
            data.metadata["risk_score"] = risk_score
            data.metadata["fraud_checked_at"] = time.time()
            
            if risk_score > 70:
                self.error_count += 1
                return ProcessingResult(
                    ProcessingStatus.FAILED, 
                    error_message=f"High fraud risk: {risk_score}"
                )
            
            self.processed_count += 1
            return ProcessingResult(ProcessingStatus.SUCCESS, data)
            
        except Exception as e:
            self.error_count += 1
            self.logger.error(f"Fraud detection failed for order {data.order_id}: {e}")
            return ProcessingResult(ProcessingStatus.FAILED, error_message=str(e))

class InventoryCheckFilter(Filter):
    """재고 확인 필터 - 상품 재고를 확인하고 예약"""
    
    def __init__(self):
        super().__init__("InventoryCheck")
        # 시뮬레이션용 재고 데이터
        self.inventory = {
            "PROD001": 100,
            "PROD002": 50,
            "PROD003": 0  # 품절 상품
        }
    
    async def process(self, data: OrderData) -> ProcessingResult:
        """재고 확인 및 예약 처리"""
        self.logger.info(f"Checking inventory for order: {data.order_id}")
        
        try:
            # 시뮬레이션: 데이터베이스 조회
            await asyncio.sleep(0.05)
            
            current_stock = self.inventory.get(data.product_id, 0)
            
            if current_stock < data.quantity:
                self.error_count += 1
                return ProcessingResult(
                    ProcessingStatus.FAILED,
                    error_message=f"Insufficient stock. Available: {current_stock}, Requested: {data.quantity}"
                )
            
            # 재고 예약 (실제로는 데이터베이스 트랜잭션 처리)
            self.inventory[data.product_id] -= data.quantity
            
            data.metadata["inventory_reserved"] = data.quantity
            data.metadata["remaining_stock"] = self.inventory[data.product_id]
            data.metadata["inventory_checked_at"] = time.time()
            
            self.processed_count += 1
            return ProcessingResult(ProcessingStatus.SUCCESS, data)
            
        except Exception as e:
            self.error_count += 1
            self.logger.error(f"Inventory check failed for order {data.order_id}: {e}")
            return ProcessingResult(ProcessingStatus.FAILED, error_message=str(e))

class PaymentProcessingFilter(Filter):
    """결제 처리 필터 - 결제 게이트웨이와 연동하여 결제 처리"""
    
    def __init__(self):
        super().__init__("PaymentProcessing")
    
    async def process(self, data: OrderData) -> ProcessingResult:
        """결제 처리"""
        self.logger.info(f"Processing payment for order: {data.order_id}")
        
        try:
            # 시뮬레이션: 외부 결제 게이트웨이 호출
            await asyncio.sleep(0.2)  # 네트워크 지연 시뮬레이션
            
            total_amount = data.price * data.quantity
            
            # 간단한 결제 시뮬레이션 (90% 성공률)
            import random
            if random.random() < 0.9:
                transaction_id = f"TXN_{data.order_id}_{int(time.time())}"
                
                data.metadata["payment_status"] = "completed"
                data.metadata["transaction_id"] = transaction_id
                data.metadata["total_amount"] = total_amount
                data.metadata["payment_processed_at"] = time.time()
                
                self.processed_count += 1
                return ProcessingResult(ProcessingStatus.SUCCESS, data)
            else:
                self.error_count += 1
                return ProcessingResult(
                    ProcessingStatus.RETRY,  # 결제 실패는 재시도 가능
                    error_message="Payment gateway error"
                )
                
        except Exception as e:
            self.error_count += 1
            self.logger.error(f"Payment processing failed for order {data.order_id}: {e}")
            return ProcessingResult(ProcessingStatus.FAILED, error_message=str(e))

# 파이프라인 클래스
class OrderProcessingPipeline:
    """주문 처리 파이프라인 - 여러 필터를 연결하여 순차적으로 처리"""
    
    def __init__(self):
        self.filters: List[Filter] = []
        self.logger = logging.getLogger("Pipeline")
        self.executor = ThreadPoolExecutor(max_workers=4)
    
    def add_filter(self, filter_instance: Filter) -> 'OrderProcessingPipeline':
        """필터를 파이프라인에 추가 (체이닝 방식)"""
        self.filters.append(filter_instance)
        return self
    
    async def process_order(self, order: OrderData) -> ProcessingResult:
        """단일 주문을 파이프라인을 통해 처리"""
        self.logger.info(f"Starting pipeline processing for order: {order.order_id}")
        
        current_data = order
        
        for filter_instance in self.filters:
            try:
                result = await filter_instance.process(current_data)
                
                if result.status == ProcessingStatus.SUCCESS:
                    current_data = result.data
                    self.logger.debug(f"Filter {filter_instance.name} completed successfully")
                elif result.status == ProcessingStatus.RETRY:
                    # 재시도 로직 (간단한 구현)
                    self.logger.warning(f"Filter {filter_instance.name} requires retry: {result.error_message}")
                    await asyncio.sleep(1)  # 재시도 대기
                    result = await filter_instance.process(current_data)
                    if result.status != ProcessingStatus.SUCCESS:
                        return result
                    current_data = result.data
                else:
                    self.logger.error(f"Filter {filter_instance.name} failed: {result.error_message}")
                    return result
                    
            except Exception as e:
                self.logger.error(f"Unexpected error in filter {filter_instance.name}: {e}")
                return ProcessingResult(ProcessingStatus.FAILED, error_message=str(e))
        
        self.logger.info(f"Pipeline processing completed for order: {order.order_id}")
        return ProcessingResult(ProcessingStatus.SUCCESS, current_data)
    
    async def process_batch(self, orders: List[OrderData]) -> List[ProcessingResult]:
        """여러 주문을 병렬로 처리"""
        self.logger.info(f"Processing batch of {len(orders)} orders")
        
        tasks = [self.process_order(order) for order in orders]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 예외 처리
        processed_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                processed_results.append(ProcessingResult(
                    ProcessingStatus.FAILED,
                    error_message=f"Batch processing error: {result}"
                ))
            else:
                processed_results.append(result)
        
        return processed_results
    
    def get_pipeline_stats(self) -> Dict[str, Any]:
        """파이프라인 전체 통계 정보 반환"""
        stats = {}
        for filter_instance in self.filters:
            stats[filter_instance.name] = filter_instance.get_stats()
        return stats

# 사용 예시 및 테스트
async def main():
    """메인 실행 함수 - 파이프라인 설정 및 테스트"""
    
    # 로깅 설정
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    # 파이프라인 구성
    pipeline = OrderProcessingPipeline()
    pipeline.add_filter(OrderValidationFilter()) \
           .add_filter(FraudDetectionFilter()) \
           .add_filter(InventoryCheckFilter()) \
           .add_filter(PaymentProcessingFilter())
    
    # 테스트 주문 데이터 생성
    test_orders = [
        OrderData("ORD001", "CUST001", "PROD001", 2, 100.0, "credit_card"),
        OrderData("ORD002", "CUST002", "PROD002", 1, 50.0, "debit_card"),
        OrderData("ORD003", "CUST999", "PROD001", 1, 200.0, "credit_card"),  # 사기 위험
        OrderData("ORD004", "CUST003", "PROD003", 1, 75.0, "paypal"),  # 재고 부족
        OrderData("ORD005", "CUST004", "PROD001", 150, 25000.0, "credit_card"),  # 고위험 + 대량
    ]
    
    # 배치 처리 실행
    print("=== Starting Order Processing Pipeline ===")
    start_time = time.time()
    
    results = await pipeline.process_batch(test_orders)
    
    processing_time = time.time() - start_time
    
    # 결과 출력
    print(f"\n=== Processing Results (Total time: {processing_time:f}s) ===")
    for i, result in enumerate(results):
        order = test_orders[i]
        print(f"\nOrder {order.order_id}:")
        print(f"  Status: {result.status.value}")
        if result.status == ProcessingStatus.SUCCESS:
            print(f"  Metadata: {json.dumps(result.data.metadata, indent=2)}")
        else:
            print(f"  Error: {result.error_message}")
    
    # 파이프라인 통계 출력
    print("\n=== Pipeline Statistics ===")
    stats = pipeline.get_pipeline_stats()
    for filter_name, filter_stats in stats.items():
        print(f"{filter_name}:")
        print(f"  Processed: {filter_stats['processed']}")
        print(f"  Errors: {filter_stats['errors']}")
        print(f"  Success Rate: {filter_stats['success_rate']:%}")

if __name__ == "__main__":
    # 비동기 실행
    asyncio.run(main())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# Pipe-and-Filter 구조의 간단한 로그 처리 파이프라인 예시

def parse_filter(logs):
    # 파싱: 로그를 구조화된 딕셔너리로 변환
    for log in logs:
        yield {"raw": log, "parsed": log.split(" ")}

def clean_filter(parsed_logs):
    # 정제: 불필요한 정보 제거
    for log in parsed_logs:
        if "ERROR" in log["parsed"]:
            yield log

def analyze_filter(cleaned_logs):
    # 분석: 에러 로그 카운트
    count = 0
    for log in cleaned_logs:
        count += 1
    yield {"error_count": count}

# 파이프라인 실행
logs = ["2023-01-01 INFO Start", "2023-01-01 ERROR Fail", "2023-01-01 ERROR Crash"]
parsed = parse_filter(logs)
cleaned = clean_filter(parsed)
analyzed = analyze_filter(cleaned)
for result in analyzed:
    print(result)  # {'error_count': 2}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408

"""
IoT 센서 데이터 처리 파이프라인 구현 예시
Pipe-and-Filter Architecture를 활용한 실시간 데이터 처리
"""

import json
import queue
import threading
import time
from abc import ABC, abstractmethod
from datetime import datetime
from typing import Any, Dict, List, Optional

# 파이프 인터페이스 정의
class Pipe:
    """필터 간 데이터 전송을 담당하는 파이프 클래스"""
    
    def __init__(self, buffer_size: int = 100):
        self.queue = queue.Queue(maxsize=buffer_size)
        self.closed = False
    
    def send(self, data: Any) -> bool:
        """데이터를 파이프에 전송"""
        if self.closed:
            return False
        try:
            self.queue.put(data, block=False)
            return True
        except queue.Full:
            return False
    
    def receive(self, timeout: float = 1.0) -> Optional[Any]:
        """파이프에서 데이터를 수신"""
        try:
            return self.queue.get(timeout=timeout)
        except queue.Empty:
            return None
    
    def close(self):
        """파이프 종료"""
        self.closed = True

# 필터 기본 클래스
class Filter(ABC):
    """모든 필터의 기본 클래스"""
    
    def __init__(self, name: str):
        self.name = name
        self.input_pipe: Optional[Pipe] = None
        self.output_pipe: Optional[Pipe] = None
        self.running = False
        self.thread: Optional[threading.Thread] = None
    
    def connect_input(self, pipe: Pipe):
        """입력 파이프 연결"""
        self.input_pipe = pipe
    
    def connect_output(self, pipe: Pipe):
        """출력 파이프 연결"""
        self.output_pipe = pipe
    
    @abstractmethod
    def process(self, data: Any) -> Optional[Any]:
        """데이터 처리 로직 (서브클래스에서 구현)"""
        pass
    
    def run(self):
        """필터 실행 메인 루프"""
        self.running = True
        while self.running:
            if self.input_pipe:
                data = self.input_pipe.receive()
                if data is not None:
                    processed_data = self.process(data)
                    if processed_data is not None and self.output_pipe:
                        success = self.output_pipe.send(processed_data)
                        if not success:
                            print(f"[{self.name}] Output pipe full, dropping data")
            else:
                time.sleep(0.1)
    
    def start(self):
        """필터 시작"""
        self.thread = threading.Thread(target=self.run, name=self.name)
        self.thread.start()
    
    def stop(self):
        """필터 종료"""
        self.running = False
        if self.thread:
            self.thread.join()

# 소스 필터 - MQTT 센서 데이터 시뮬레이션
class SensorSourceFilter(Filter):
    """IoT 센서 데이터를 생성하는 소스 필터"""
    
    def __init__(self, sensor_id: str, sensor_type: str):
        super().__init__(f"SensorSource-{sensor_id}")
        self.sensor_id = sensor_id
        self.sensor_type = sensor_type
        self.data_counter = 0
    
    def process(self, data: Any) -> Optional[Any]:
        # 소스 필터는 입력 데이터를 처리하지 않음
        return None
    
    def run(self):
        """센서 데이터 생성 및 전송"""
        self.running = True
        while self.running:
            # 센서 데이터 시뮬레이션
            sensor_data = {
                "sensor_id": self.sensor_id,
                "sensor_type": self.sensor_type,
                "timestamp": datetime.now().isoformat(),
                "value": self._generate_sensor_value(),
                "sequence": self.data_counter
            }
            
            if self.output_pipe:
                success = self.output_pipe.send(sensor_data)
                if success:
                    self.data_counter += 1
                    print(f"[{self.name}] Generated: {sensor_data}")
                else:
                    print(f"[{self.name}] Output pipe full")
            
            time.sleep(2)  # 2초마다 데이터 생성
    
    def _generate_sensor_value(self) -> float:
        """센서 타입에 따른 값 생성"""
        import random
        if self.sensor_type == "temperature":
            return round(random.uniform(20.0, 30.0), 2)
        elif self.sensor_type == "humidity":
            return round(random.uniform(40.0, 80.0), 2)
        elif self.sensor_type == "pressure":
            return round(random.uniform(1000.0, 1020.0), 2)
        return 0.0

# JSON 파싱 필터
class JsonParseFilter(Filter):
    """JSON 데이터를 파싱하고 검증하는 필터"""
    
    def __init__(self):
        super().__init__("JsonParser")
    
    def process(self, data: Any) -> Optional[Any]:
        """JSON 데이터 파싱 및 구조 검증"""
        try:
            if isinstance(data, str):
                parsed_data = json.loads(data)
            else:
                parsed_data = data
            
            # 필수 필드 검증
            required_fields = ["sensor_id", "sensor_type", "timestamp", "value"]
            for field in required_fields:
                if field not in parsed_data:
                    print(f"[{self.name}] Missing field: {field}")
                    return None
            
            print(f"[{self.name}] Parsed: {parsed_data['sensor_id']}")
            return parsed_data
            
        except json.JSONDecodeError as e:
            print(f"[{self.name}] JSON parse error: {e}")
            return None
        except Exception as e:
            print(f"[{self.name}] Processing error: {e}")
            return None

# 데이터 검증 필터
class ValidationFilter(Filter):
    """센서 데이터 값의 유효성을 검증하는 필터"""
    
    def __init__(self):
        super().__init__("Validator")
        # 센서 타입별 유효 범위 정의
        self.valid_ranges = {
            "temperature": (0.0, 50.0),
            "humidity": (0.0, 100.0),
            "pressure": (950.0, 1050.0)
        }
    
    def process(self, data: Any) -> Optional[Any]:
        """데이터 값 범위 검증"""
        try:
            sensor_type = data["sensor_type"]
            value = float(data["value"])
            
            if sensor_type in self.valid_ranges:
                min_val, max_val = self.valid_ranges[sensor_type]
                if min_val <= value <= max_val:
                    print(f"[{self.name}] Valid: {sensor_type}={value}")
                    return data
                else:
                    print(f"[{self.name}] Invalid range: {sensor_type}={value}")
                    return None
            else:
                print(f"[{self.name}] Unknown sensor type: {sensor_type}")
                return None
                
        except (KeyError, ValueError, TypeError) as e:
            print(f"[{self.name}] Validation error: {e}")
            return None

# 데이터 변환 필터
class TransformFilter(Filter):
    """데이터 형식을 표준화하는 변환 필터"""
    
    def __init__(self):
        super().__init__("Transformer")
    
    def process(self, data: Any) -> Optional[Any]:
        """데이터를 표준 형식으로 변환"""
        try:
            # 표준 형식으로 데이터 변환
            transformed_data = {
                "id": data["sensor_id"],
                "type": data["sensor_type"],
                "timestamp": data["timestamp"],
                "measurement": {
                    "value": data["value"],
                    "unit": self._get_unit(data["sensor_type"])
                },
                "metadata": {
                    "sequence": data.get("sequence", 0),
                    "processed_at": datetime.now().isoformat()
                }
            }
            
            print(f"[{self.name}] Transformed: {transformed_data['id']}")
            return transformed_data
            
        except Exception as e:
            print(f"[{self.name}] Transform error: {e}")
            return None
    
    def _get_unit(self, sensor_type: str) -> str:
        """센서 타입에 따른 단위 반환"""
        units = {
            "temperature": "°C",
            "humidity": "%",
            "pressure": "hPa"
        }
        return units.get(sensor_type, "")

# 라우터 필터
class RouterFilter(Filter):
    """센서 타입에 따라 데이터를 다른 경로로 라우팅"""
    
    def __init__(self):
        super().__init__("Router")
        self.output_pipes: Dict[str, Pipe] = {}
    
    def add_route(self, sensor_type: str, pipe: Pipe):
        """특정 센서 타입에 대한 출력 파이프 추가"""
        self.output_pipes[sensor_type] = pipe
    
    def process(self, data: Any) -> Optional[Any]:
        """센서 타입에 따른 라우팅"""
        try:
            sensor_type = data["type"]
            if sensor_type in self.output_pipes:
                success = self.output_pipes[sensor_type].send(data)
                if success:
                    print(f"[{self.name}] Routed {sensor_type} data")
                else:
                    print(f"[{self.name}] Route pipe full for {sensor_type}")
            else:
                print(f"[{self.name}] No route for {sensor_type}")
            
            # 라우터는 원본 데이터를 변경하지 않고 전달만 함
            return None
            
        except Exception as e:
            print(f"[{self.name}] Routing error: {e}")
            return None

# 싱크 필터
class DatabaseSinkFilter(Filter):
    """처리된 데이터를 데이터베이스에 저장하는 싱크 필터"""
    
    def __init__(self, db_name: str):
        super().__init__(f"DatabaseSink-{db_name}")
        self.db_name = db_name
        self.stored_count = 0
    
    def process(self, data: Any) -> Optional[Any]:
        """데이터베이스에 데이터 저장 (시뮬레이션)"""
        try:
            # 실제 구현에서는 데이터베이스 연결 및 저장 수행
            print(f"[{self.name}] Storing to {self.db_name}: {data['id']} = {data['measurement']['value']}")
            self.stored_count += 1
            
            # 싱크 필터는 출력이 없음
            return None
            
        except Exception as e:
            print(f"[{self.name}] Storage error: {e}")
            return None

# 파이프라인 관리자
class Pipeline:
    """전체 파이프라인을 관리하는 클래스"""
    
    def __init__(self):
        self.filters: List[Filter] = []
        self.pipes: List[Pipe] = []
    
    def add_filter(self, filter_instance: Filter):
        """필터를 파이프라인에 추가"""
        self.filters.append(filter_instance)
    
    def add_pipe(self, pipe: Pipe):
        """파이프를 파이프라인에 추가"""
        self.pipes.append(pipe)
    
    def connect(self, source_filter: Filter, target_filter: Filter, pipe: Pipe):
        """두 필터를 파이프로 연결"""
        source_filter.connect_output(pipe)
        target_filter.connect_input(pipe)
        self.add_pipe(pipe)
    
    def start(self):
        """전체 파이프라인 시작"""
        print("Starting pipeline…")
        for filter_instance in self.filters:
            filter_instance.start()
        print("Pipeline started!")
    
    def stop(self):
        """전체 파이프라인 종료"""
        print("Stopping pipeline…")
        for filter_instance in self.filters:
            filter_instance.stop()
        for pipe in self.pipes:
            pipe.close()
        print("Pipeline stopped!")

# 메인 실행 예제
def main():
    """IoT 데이터 처리 파이프라인 실행 예제"""
    
    # 파이프라인 생성
    pipeline = Pipeline()
    
    # 필터 인스턴스 생성
    temp_source = SensorSourceFilter("TEMP_01", "temperature")
    humidity_source = SensorSourceFilter("HUM_01", "humidity")
    json_parser = JsonParseFilter()
    validator = ValidationFilter()
    transformer = TransformFilter()
    router = RouterFilter()
    temp_sink = DatabaseSinkFilter("TemperatureDB")
    humidity_sink = DatabaseSinkFilter("HumidityDB")
    
    # 파이프 생성
    temp_to_parser = Pipe()
    humidity_to_parser = Pipe()
    parser_to_validator = Pipe()
    validator_to_transformer = Pipe()
    transformer_to_router = Pipe()
    temp_route_pipe = Pipe()
    humidity_route_pipe = Pipe()
    
    # 필터들을 파이프라인에 추가
    pipeline.add_filter(temp_source)
    pipeline.add_filter(humidity_source)
    pipeline.add_filter(json_parser)
    pipeline.add_filter(validator)
    pipeline.add_filter(transformer)
    pipeline.add_filter(router)
    pipeline.add_filter(temp_sink)
    pipeline.add_filter(humidity_sink)
    
    # 파이프라인 연결
    pipeline.connect(temp_source, json_parser, temp_to_parser)
    pipeline.connect(humidity_source, json_parser, humidity_to_parser)
    pipeline.connect(json_parser, validator, parser_to_validator)
    pipeline.connect(validator, transformer, validator_to_transformer)
    pipeline.connect(transformer, router, transformer_to_router)
    
    # 라우터 설정
    router.add_route("temperature", temp_route_pipe)
    router.add_route("humidity", humidity_route_pipe)
    temp_sink.connect_input(temp_route_pipe)
    humidity_sink.connect_input(humidity_route_pipe)
    pipeline.add_pipe(temp_route_pipe)
    pipeline.add_pipe(humidity_route_pipe)
    
    try:
        # 파이프라인 시작
        pipeline.start()
        
        # 30초간 실행
        print("Pipeline running for 30 seconds…")
        time.sleep(30)
        
    except KeyboardInterrupt:
        print("\nReceived interrupt signal")
    finally:
        # 파이프라인 종료
        pipeline.stop()

if __name__ == "__main__":
    main()