AI 실습 공간

AWS Comprehend Medical 예제 코드:

import boto3

comprehend_medical = boto3.client(service_name='comprehendmedical')

text = "환자는 고혈압과 당뇨병이 있으며 하루 아토바스타틴 40mg을 복용하고 있습니다."

result = comprehend_medical.detect_entities(Text=text)

for entity in result['Entities']:
    print(f"Entity: {entity['Text']}, Type: {entity['Type']}, Category: {entity['Category']}")

Azure Text Analytics for Health 예제 코드:

from azure.core.credentials import AzureKeyCredential
from azure.ai.textanalytics import TextAnalyticsClient

endpoint = "YOUR_ENDPOINT"
key = "YOUR_KEY"

text_analytics_client = TextAnalyticsClient(
    endpoint=endpoint, 
    credential=AzureKeyCredential(key)
)

documents = ["환자는 지난 2주간 가슴 통증을 호소하였으며, 심전도에서 ST부분 상승이 관찰되었습니다."]

result = text_analytics_client.analyze_healthcare_entities(documents)

docs = [doc for doc in result if not doc.is_error]

for doc in docs:
    for entity in doc.entities:
        print(f"Entity: {entity.text}, Category: {entity.category}, Subcategory: {entity.subcategory}")

Google Cloud Healthcare API 예제 코드:

from google.cloud import healthcare

client = healthcare.HealthcareClient()

# FHIR 스토어 접근 경로 설정
fhir_store_name = f"projects/{project_id}/locations/{location}/datasets/{dataset_id}/fhirStores/{fhir_store_id}"

# 환자 리소스 검색
resource_type = "Patient"
request = client.projects().locations().datasets().fhirStores().fhir().search(
    parent=fhir_store_name,
    resourceType=resource_type
)

response = request.execute()

# 결과 처리
for entry in response.get('entry', []):
    resource = entry.get('resource', {})
    patient_id = resource.get('id')
    print(f"Patient ID: {patient_id}")

DICOM 의료 이미지 처리 예제 코드:

import pydicom
import matplotlib.pyplot as plt
import numpy as np

# DICOM 파일 읽기
def read_dicom_image(file_path):
    dicom_data = pydicom.dcmread(file_path)
    img = dicom_data.pixel_array
    return img, dicom_data

# 이미지 전처리 및 표시
def process_and_display(img):
    # 이미지 정규화
    img_normalized = (img - np.min(img)) / (np.max(img) - np.min(img))
    
    plt.figure(figsize=(10, 8))
    plt.imshow(img_normalized, cmap='bone')
    plt.colorbar()
    plt.title('DICOM 의료 이미지')
    plt.show()
    
    return img_normalized

임상시험 데이터 시각화 컴포넌트 예제:

import React from 'react';
import { Line } from 'react-chartjs-2';
import {
  Chart as ChartJS,
  CategoryScale,
  LinearScale,
  PointElement,
  LineElement,
  Title,
  Tooltip,
  Legend,
} from 'chart.js';

ChartJS.register(
  CategoryScale,
  LinearScale,
  PointElement,
  LineElement,
  Title,
  Tooltip,
  Legend
);

const ClinicalTrialChart = () => {
  const data = {
    labels: ['Week 0', 'Week 4', 'Week 8', 'Week 12', 'Week 16', 'Week 20'],
    datasets: [
      {
        label: '실험군',
        data: [65, 59, 52, 45, 40, 35],
        borderColor: 'rgb(53, 162, 235)',
        backgroundColor: 'rgba(53, 162, 235, 0.5)',
      },
      {
        label: '대조군',
        data: [65, 62, 60, 57, 55, 53],
        borderColor: 'rgb(255, 99, 132)',
        backgroundColor: 'rgba(255, 99, 132, 0.5)',
      },
    ],
  };

  const options = {
    responsive: true,
    plugins: {
      legend: {
        position: 'top' as const,
      },
      title: {
        display: true,
        text: '임상시험 결과: 실험군 vs 대조군',
      },
    },
  };

  return <Line options={options} data={data} />;
};

export default ClinicalTrialChart;

API 테스트 스크립트 예제 (Python/pytest):

import pytest
import requests
import json

API_URL = "http://localhost:8000"
API_KEY = "your-secure-api-key"

# 테스트 환자 데이터
test_patient = {
    "resourceType": "Patient",
    "name": [
        {
            "family": "김",
            "given": ["철수"]
        }
    ],
    "gender": "male",
    "birthDate": "1980-01-01",
    "address": [
        {
            "text": "서울시 강남구"
        }
    ],
    "telecom": [
        {
            "system": "phone",
            "value": "010-1234-5678"
        }
    ]
}

@pytest.fixture
def api_client():
    session = requests.Session()
    session.headers.update({
        "Content-Type": "application/json",
        "X-API-Key": API_KEY
    })
    return session

@pytest.fixture
def created_patient(api_client):
    # 테스트용 환자 생성
    response = api_client.post(f"{API_URL}/fhir/Patient", json=test_patient)
    assert response.status_code == 200
    patient_data = response.json()
    
    yield patient_data
    
    # 테스트 후 정리 (환자 삭제 등)
    # api_client.delete(f"{API_URL}/fhir/Patient/{patient_data['id']}")

def test_create_patient(api_client):
    response = api_client.post(f"{API_URL}/fhir/Patient", json=test_patient)
    assert response.status_code == 200
    data = response.json()
    assert data["resourceType"] == "Patient"
    assert data["name"][0]["family"] == "김"
    assert data["gender"] == "male"

def test_get_patient(api_client, created_patient):
    response = api_client.get(f"{API_URL}/fhir/Patient/{created_patient['id']}")
    assert response.status_code == 200
    data = response.json()
    assert data["id"] == created_patient["id"]
    assert data["name"][0]["family"] == "김"

def test_search_patients(api_client, created_patient):
    # 이름으로 검색
    response = api_client.get(f"{API_URL}/fhir/Patient?name=김")
    assert response.status_code == 200
    data = response.json()
    assert len(data) > 0
    assert any(p["id"] == created_patient["id"] for p in data)
    
    # 성별로 검색
    response = api_client.get(f"{API_URL}/fhir/Patient?gender=male")
    assert response.status_code == 200
    data = response.json()
    assert len(data) > 0
    assert any(p["id"] == created_patient["id"] for p in data)

def test_api_security(api_client):
    # 잘못된 API 키로 요청
    headers = {
        "Content-Type": "application/json",
        "X-API-Key": "wrong-api-key"
    }
    response = requests.get(f"{API_URL}/fhir/Patient", headers=headers)
    assert response.status_code == 403  # Forbidden

def test_data_validation(api_client):
    # 잘못된 환자 데이터 (필수 필드 누락)
    invalid_patient = {
        "resourceType": "Patient",
        # name 필드 누락
        "gender": "unknown"  # 잘못된 성별 값
    }
    response = api_client.post(f"{API_URL}/fhir/Patient", json=invalid_patient)
    assert response.status_code in [400, 422]  # Bad Request 또는 Unprocessable Entity

의료 API 서버 Deployment 예제:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: medical-api
  namespace: healthcare
  labels:
    app: medical-api
    tier: backend
spec:
  replicas: 3
  selector:
    matchLabels:
      app: medical-api
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    metadata:
      labels:
        app: medical-api
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8000"
    spec:
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
      containers:
      - name: medical-api
        image: your-registry/medical-api:1.0.0
        imagePullPolicy: Always
        ports:
        - containerPort: 8000
          name: http
        resources:
          requests:
            cpu: 500m
            memory: 512Mi
          limits:
            cpu: 1000m
            memory: 1Gi
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 15
          periodSeconds: 20
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: medical-api-secrets
              key: database-url
        - name: API_KEY
          valueFrom:
            secretKeyRef:
              name: medical-api-secrets
              key: api-key
        - name: LOG_LEVEL
          value: "info"
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 100
            podAffinityTerm:
              labelSelector:
                matchExpressions:
                - key: app
                  operator: In
                  values:
                  - medical-api
              topologyKey: "kubernetes.io/hostname"
---
apiVersion: v1
kind: Service
metadata:
  name: medical-api
  namespace: healthcare
spec:
  selector:
    app: medical-api
  ports:
  - port: 80
    targetPort: 8000
    name: http
  type: ClusterIP
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: medical-api
  namespace: healthcare
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: medical-api
  minReplicas: 3
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

GitHub Actions CI/CD 파이프라인 예제:

name: Medical API CI/CD Pipeline

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

env:
  REGISTRY: ghcr.io
  IMAGE_NAME: ${{ github.repository }}/medical-api

jobs:
  test:
    name: Test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.8'
          
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install pytest pytest-cov flake8 bandit
          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
          
      - name: Lint with flake8
        run: flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
        
      - name: Security scan with bandit
        run: bandit -r . -x tests/
        
      - name: Run tests with pytest
        run: pytest --cov=./ --cov-report=xml
        
      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v3
        
  build:
    name: Build and Push
    needs: test
    runs-on: ubuntu-latest
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
    permissions:
      contents: read
      packages: write
    
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v2
        
      - name: Log in to the Container registry
        uses: docker/login-action@v2
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
          
      - name: Extract metadata
        id: meta
        uses: docker/metadata-action@v4
        with:
          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
          
      - name: Build and push
        uses: docker/build-push-action@v4
        with:
          context: .
          push: true
          tags: ${{ steps.meta.outputs.tags }}
          labels: ${{ steps.meta.outputs.labels }}
          cache-from: type=gha
          cache-to: type=gha,mode=max
          
  deploy:
    name: Deploy to Kubernetes
    needs: build
    runs-on: ubuntu-latest
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
    
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up kubeconfig
        uses: azure/k8s-set-context@v3
        with:
          kubeconfig: ${{ secrets.KUBE_CONFIG }}
          
      - name: Deploy to Kubernetes
        uses: azure/k8s-deploy@v4
        with:
          namespace: healthcare
          manifests: |
            kubernetes/deployment.yaml
            kubernetes/service.yaml
            kubernetes/hpa.yaml
          images: |
            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }}
          strategy: canary
          percentage: 20
          
      - name: Verify deployment
        run: |
          kubectl rollout status deployment/medical-api -n healthcare
          
      - name: Promote deployment
        if: success()
        uses: azure/k8s-deploy@v4
        with:
          namespace: healthcare
          manifests: |
            kubernetes/deployment.yaml
            kubernetes/service.yaml
            kubernetes/hpa.yaml
          images: |
            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }}
          strategy: canary
          percentage: 100

Prometheus 모니터링 설정 예제:

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: medical-api-monitor
  namespace: monitoring
  labels:
    release: prometheus
spec:
  selector:
    matchLabels:
      app: medical-api
  namespaceSelector:
    matchNames:
      - healthcare
  endpoints:
  - port: http
    path: /metrics
    interval: 15s
    scrapeTimeout: 10s
---
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: medical-api-alerts
  namespace: monitoring
  labels:
    release: prometheus
spec:
  groups:
  - name: medical-api
    rules:
    - alert: HighErrorRate
      expr: sum(rate(http_requests_total{job="medical-api",status=~"5.."}[5m])) / sum(rate(http_requests_total{job="medical-api"}[5m])) > 0.05
      for: 5m
      labels:
        severity: critical
        service: medical-api
      annotations:
        summary: "High error rate on Medical API"
        description: "Medical API error rate is above 5% (current value: {{ $value }})"
        
    - alert: SlowResponseTime
      expr: histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket{job="medical-api"}[5m])) by (le)) > 0.5
      for: 5m
      labels:
        severity: warning
        service: medical-api
      annotations:
        summary: "Slow response time on Medical API"
        description: "Medical API 95th percentile response time is above 500ms (current value: {{ $value }}s)"
        
    - alert: HighCPUUsage
      expr: sum(rate(container_cpu_usage_seconds_total{container="medical-api"}[5m])) / sum(kube_pod_container_resource_limits_cpu_cores{container="medical-api"}) > 0.8
      for: 10m
      labels:
        severity: warning
        service: medical-api
      annotations:
        summary: "High CPU usage on Medical API"
        description: "Medical API CPU usage is above 80% of limit (current value: {{ $value }})"
        
    - alert: HighMemoryUsage
      expr: sum(container_memory_working_set_bytes{container="medical-api"}) / sum(kube_pod_container_resource_limits_memory_bytes{container="medical-api"}) > 0.8
      for: 10m
      labels:
        severity: warning
        service: medical-api
      annotations:
        summary: "High memory usage on Medical API"
        description: "Medical API memory usage is above 80% of limit (current value: {{ $value }})"
        
    - alert: PodCrashLooping
      expr: increase(kube_pod_container_status_restarts_total{container="medical-api"}[15m]) > 3
      for: 10m
      labels:
        severity: critical
        service: medical-api
      annotations:
        summary: "Medical API pod is crash looping"
        description: "Pod {{ $labels.pod }} has restarted {{ $value }} times in the last 15 minutes"

데이터베이스 백업 CronJob 예제:

apiVersion: batch/v1
kind: CronJob
metadata:
  name: medical-db-backup
  namespace: healthcare
spec:
  schedule: "0 2 * * *"  # 매일 오전 2시에 실행
  concurrencyPolicy: Forbid
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 3
  jobTemplate:
    spec:
      backoffLimit: 3
      template:
        spec:
          securityContext:
            runAsNonRoot: true
            runAsUser: 1000
          containers:
          - name: db-backup
            image: your-registry/db-backup:1.0.0
            imagePullPolicy: IfNotPresent
            command:
            - /bin/sh
            - -c
            - |
              # 백업 파일명 생성 (날짜 포함)
              BACKUP_FILE="medical_db_$(date +%Y%m%d_%H%M%S).sql.gz"
              
              # 데이터베이스 덤프 생성 및 압축
              PGPASSWORD=$DB_PASSWORD pg_dump -h $DB_HOST -U $DB_USER -d $DB_NAME | gzip > /backups/$BACKUP_FILE
              
              # 백업 파일 암호화 (HIPAA 준수)
              gpg --symmetric --batch --passphrase "$ENCRYPTION_KEY" --cipher-algo AES256 /backups/$BACKUP_FILE
              
              # 클라우드 스토리지에 업로드
              aws s3 cp /backups/$BACKUP_FILE.gpg s3://$S3_BUCKET/backups/
              
              # 성공 로그
              echo "Backup completed and uploaded to S3: $BACKUP_FILE.gpg"
              
              # 오래된 로컬 백업 정리 (7일 이상)
              find /backups -type f -name "*.gpg" -mtime +7 -delete
            env:
            - name: DB_HOST
              valueFrom:
                secretKeyRef:
                  name: medical-db-credentials
                  key: host
            - name: DB_USER
              valueFrom:
                secretKeyRef:
                  name: medical-db-credentials
                  key: username
            - name: DB_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: medical-db-credentials
                  key: password
            - name: DB_NAME
              valueFrom:
                secretKeyRef:
                  name: medical-db-credentials
                  key: database
            - name: ENCRYPTION_KEY
              valueFrom:
                secretKeyRef:
                  name: backup-encryption
                  key: key
            - name: S3_BUCKET
              valueFrom:
                configMapKeyRef:
                  name: backup-config
                  key: s3-bucket
            volumeMounts:
            - name: backup-volume
              mountPath: /backups
          volumes:
          - name: backup-volume
            persistentVolumeClaim:
              claimName: backup-pvc
          restartPolicy: OnFailure