Construction Expert

建設管理、プロジェクト計画、Building Information Modeling (BIM)、安全コンプライアンス、および最新の建設技術ソリューションに関する専門的なガイダンス。

Core Concepts

建設管理

• プロジェクト計画とスケジューリング
• コスト推定と管理
• リソース管理
• 品質保証
• 契約管理
• リスク管理
• 変更指示書管理

技術

• Building Information Modeling (BIM)
• 建設管理ソフトウェア
• ドローン測量と検査
• 3Dプリンティングと模組化建設
• 監視用IoTセンサー
• 拡張現実による可視化
• 建設ロボティクス

基準と規制

• OSHA安全規制
• 建築基準 (IBC、IRC)
• AIA契約と基準
• LEED認証
• ISO 19650 (BIM基準)
• CSI MasterFormat
• 環境規制

プロジェクト管理システム

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Optional, Dict
from decimal import Decimal
from enum import Enum

class ProjectPhase(Enum):
    PRE_CONSTRUCTION = "pre_construction"
    SITE_PREPARATION = "site_preparation"
    FOUNDATION = "foundation"
    FRAMING = "framing"
    MEP = "mep"  # Mechanical, Electrical, Plumbing
    INTERIOR = "interior"
    EXTERIOR = "exterior"
    FINAL = "final"
    CLOSEOUT = "closeout"

class TaskStatus(Enum):
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    DELAYED = "delayed"
    ON_HOLD = "on_hold"

@dataclass
class ConstructionProject:
    """Construction project information"""
    project_id: str
    project_name: str
    location: dict
    project_type: str  # 'residential', 'commercial', 'industrial'
    owner: str
    general_contractor: str
    start_date: datetime
    planned_end_date: datetime
    actual_end_date: Optional[datetime]
    budget: Decimal
    current_cost: Decimal
    square_footage: float
    current_phase: ProjectPhase

@dataclass
class Task:
    """Construction task/activity"""
    task_id: str
    project_id: str
    name: str
    description: str
    phase: ProjectPhase
    status: TaskStatus
    assigned_to: str  # Subcontractor or crew
    planned_start: datetime
    planned_end: datetime
    actual_start: Optional[datetime]
    actual_end: Optional[datetime]
    budget: Decimal
    actual_cost: Decimal
    predecessors: List[str]  # Task IDs that must complete first
    progress_percent: float

class ConstructionManagementSystem:
    """Construction project management system"""

    def __init__(self):
        self.projects = {}
        self.tasks = {}
        self.change_orders = []
        self.inspections = []

    def create_project_schedule(self, project_id: str, tasks_data: List[dict]) -> dict:
        """Create project schedule using Critical Path Method"""
        project = self.projects.get(project_id)
        if not project:
            return {'error': 'Project not found'}

        # Create tasks
        tasks = []
        for task_data in tasks_data:
            task = Task(
                task_id=self._generate_task_id(),
                project_id=project_id,
                name=task_data['name'],
                description=task_data.get('description', ''),
                phase=ProjectPhase(task_data['phase']),
                status=TaskStatus.NOT_STARTED,
                assigned_to=task_data['assigned_to'],
                planned_start=task_data['planned_start'],
                planned_end=task_data['planned_end'],
                actual_start=None,
                actual_end=None,
                budget=Decimal(str(task_data['budget'])),
                actual_cost=Decimal('0'),
                predecessors=task_data.get('predecessors', []),
                progress_percent=0.0
            )
            tasks.append(task)
            self.tasks[task.task_id] = task

        # Calculate critical path
        critical_path = self._calculate_critical_path(tasks)

        # Calculate project duration
        if tasks:
            project_end = max(t.planned_end for t in tasks)
            project_duration = (project_end - project.start_date).days
        else:
            project_duration = 0

        return {
            'project_id': project_id,
            'total_tasks': len(tasks),
            'project_duration_days': project_duration,
            'critical_path': [t.task_id for t in critical_path],
            'critical_path_duration': sum(
                (t.planned_end - t.planned_start).days for t in critical_path
            )
        }

    def _calculate_critical_path(self, tasks: List[Task]) -> List[Task]:
        """Calculate critical path through project network"""
        # Simplified critical path calculation
        # In production, would use proper CPM algorithm

        # Find tasks with no predecessors
        start_tasks = [t for t in tasks if not t.predecessors]

        # Find longest path through network
        critical_path = []
        current_tasks = start_tasks

        while current_tasks:
            # Find task with longest duration
            longest_task = max(current_tasks,
                             key=lambda t: (t.planned_end - t.planned_start).days)
            critical_path.append(longest_task)

            # Find successors
            current_tasks = [
                t for t in tasks
                if longest_task.task_id in t.predecessors
            ]

        return critical_path

    def track_progress(self, project_id: str) -> dict:
        """Track project progress and performance"""
        project = self.projects.get(project_id)
        if not project:
            return {'error': 'Project not found'}

        project_tasks = [t for t in self.tasks.values() if t.project_id == project_id]

        # Calculate overall progress
        if project_tasks:
            overall_progress = sum(t.progress_percent for t in project_tasks) / len(project_tasks)
        else:
            overall_progress = 0.0

        # Calculate schedule performance
        total_planned_days = (project.planned_end_date - project.start_date).days
        elapsed_days = (datetime.now() - project.start_date).days
        planned_progress = (elapsed_days / total_planned_days * 100) if total_planned_days > 0 else 0

        schedule_variance = overall_progress - planned_progress

        # Calculate cost performance
        cost_variance = project.budget - project.current_cost
        cost_performance_index = float(project.budget / project.current_cost) if project.current_cost > 0 else 1.0

        # Calculate estimated completion date
        if overall_progress > 0:
            estimated_total_days = elapsed_days / (overall_progress / 100)
            estimated_completion = project.start_date + timedelta(days=estimated_total_days)
        else:
            estimated_completion = project.planned_end_date

        return {
            'project_id': project_id,
            'overall_progress_percent': overall_progress,
            'schedule_variance_percent': schedule_variance,
            'schedule_status': 'ahead' if schedule_variance > 0 else 'behind' if schedule_variance < 0 else 'on_track',
            'cost_variance': float(cost_variance),
            'cost_performance_index': cost_performance_index,
            'budget_status': 'under' if cost_variance > 0 else 'over',
            'estimated_completion': estimated_completion.isoformat(),
            'days_variance': (estimated_completion - project.planned_end_date).days
        }

    def manage_change_order(self, project_id: str, change_data: dict) -> dict:
        """Manage construction change orders"""
        project = self.projects.get(project_id)
        if not project:
            return {'error': 'Project not found'}

        change_order = {
            'co_id': self._generate_co_id(),
            'project_id': project_id,
            'description': change_data['description'],
            'reason': change_data['reason'],
            'cost_impact': Decimal(str(change_data['cost_impact'])),
            'schedule_impact_days': change_data.get('schedule_impact_days', 0),
            'submitted_by': change_data['submitted_by'],
            'submitted_date': datetime.now(),
            'status': 'pending_approval',
            'approved': False
        }

        self.change_orders.append(change_order)

        return {
            'change_order_id': change_order['co_id'],
            'cost_impact': float(change_order['cost_impact']),
            'schedule_impact_days': change_order['schedule_impact_days'],
            'new_budget': float(project.budget + change_order['cost_impact']),
            'new_completion_date': (
                project.planned_end_date + timedelta(days=change_order['schedule_impact_days'])
            ).isoformat()
        }

    def estimate_costs(self, project_type: str, square_footage: float, specifications: dict) -> dict:
        """Estimate construction costs"""
        # Cost per square foot by project type
        base_costs = {
            'residential_basic': Decimal('150'),
            'residential_luxury': Decimal('300'),
            'commercial_office': Decimal('200'),
            'industrial_warehouse': Decimal('75')
        }

        base_cost_per_sf = base_costs.get(project_type, Decimal('150'))

        # Calculate base cost
        base_cost = base_cost_per_sf * Decimal(str(square_footage))

        # Add complexity factors
        complexity_factor = Decimal('1.0')

        if specifications.get('custom_design', False):
            complexity_factor += Decimal('0.15')

        if specifications.get('sustainable_materials', False):
            complexity_factor += Decimal('0.10')

        if specifications.get('complex_site', False):
            complexity_factor += Decimal('0.20')

        adjusted_cost = base_cost * complexity_factor

        # Add contingency (10%)
        contingency = adjusted_cost * Decimal('0.10')

        # Breakdown by category
        breakdown = {
            'site_work': float(adjusted_cost * Decimal('0.08')),
            'foundation': float(adjusted_cost * Decimal('0.12')),
            'structure': float(adjusted_cost * Decimal('0.25')),
            'exterior': float(adjusted_cost * Decimal('0.15')),
            'interior': float(adjusted_cost * Decimal('0.20')),
            'mep': float(adjusted_cost * Decimal('0.20'))
        }

        total_estimate = adjusted_cost + contingency

        return {
            'project_type': project_type,
            'square_footage': square_footage,
            'base_cost_per_sf': float(base_cost_per_sf),
            'complexity_factor': float(complexity_factor),
            'adjusted_cost': float(adjusted_cost),
            'contingency': float(contingency),
            'total_estimate': float(total_estimate),
            'cost_breakdown': breakdown
        }

    def _generate_task_id(self) -> str:
        import uuid
        return f"TASK-{uuid.uuid4().hex[:8].upper()}"

    def _generate_co_id(self) -> str:
        import uuid
        return f"CO-{uuid.uuid4().hex[:6].upper()}"

安全管理システム

@dataclass
class SafetyIncident:
    """Safety incident report"""
    incident_id: str
    project_id: str
    incident_type: str  # 'injury', 'near_miss', 'property_damage'
    severity: str  # 'minor', 'moderate', 'severe', 'fatal'
    description: str
    location: str
    occurred_at: datetime
    reported_by: str
    injured_person: Optional[str]
    root_cause: Optional[str]
    corrective_actions: List[str]

class SafetyManagementSystem:
    """Construction safety management"""

    def __init__(self):
        self.incidents = []
        self.safety_inspections = []
        self.training_records = []

    def conduct_safety_inspection(self, project_id: str, inspector: str) -> dict:
        """Conduct safety inspection"""
        inspection_items = [
            'Personal protective equipment (PPE)',
            'Fall protection systems',
            'Scaffolding integrity',
            'Electrical safety',
            'Equipment guarding',
            'Housekeeping',
            'Fire prevention',
            'First aid availability',
            'Emergency exits',
            'Signage and barriers'
        ]

        violations = []
        passed_items = []

        # Simulate inspection (in production, would be actual checklist)
        for item in inspection_items:
            # Random pass/fail for demonstration
            import random
            if random.random() < 0.85:  # 85% pass rate
                passed_items.append(item)
            else:
                violations.append({
                    'item': item,
                    'severity': random.choice(['minor', 'major']),
                    'action_required': 'Correct immediately' if random.random() < 0.3 else 'Correct within 24 hours'
                })

        inspection = {
            'inspection_id': self._generate_inspection_id(),
            'project_id': project_id,
            'inspector': inspector,
            'inspection_date': datetime.now(),
            'items_inspected': len(inspection_items),
            'items_passed': len(passed_items),
            'violations': violations,
            'overall_score': (len(passed_items) / len(inspection_items)) * 100,
            'status': 'pass' if len(violations) == 0 else 'fail'
        }

        self.safety_inspections.append(inspection)

        return inspection

    def report_incident(self, incident_data: dict) -> SafetyIncident:
        """Report safety incident"""
        incident = SafetyIncident(
            incident_id=self._generate_incident_id(),
            project_id=incident_data['project_id'],
            incident_type=incident_data['incident_type'],
            severity=incident_data['severity'],
            description=incident_data['description'],
            location=incident_data['location'],
            occurred_at=incident_data['occurred_at'],
            reported_by=incident_data['reported_by'],
            injured_person=incident_data.get('injured_person'),
            root_cause=None,
            corrective_actions=[]
        )

        self.incidents.append(incident)

        # Notify relevant parties
        self._notify_incident(incident)

        return incident

    def calculate_safety_metrics(self, project_id: str, hours_worked: float) -> dict:
        """Calculate safety performance metrics"""
        project_incidents = [
            i for i in self.incidents
            if i.project_id == project_id
        ]

        # Count recordable incidents
        recordable_incidents = [
            i for i in project_incidents
            if i.incident_type == 'injury' and i.severity in ['moderate', 'severe', 'fatal']
        ]

        # OSHA Incident Rate = (Number of incidents × 200,000) / Total hours worked
        if hours_worked > 0:
            incident_rate = (len(recordable_incidents) * 200000) / hours_worked
        else:
            incident_rate = 0

        # Days Away, Restricted, or Transferred (DART) Rate
        dart_incidents = [
            i for i in recordable_incidents
            if i.severity in ['severe', 'fatal']
        ]
        dart_rate = (len(dart_incidents) * 200000) / hours_worked if hours_worked > 0 else 0

        return {
            'project_id': project_id,
            'total_hours_worked': hours_worked,
            'total_incidents': len(project_incidents),
            'recordable_incidents': len(recordable_incidents),
            'incident_rate': incident_rate,
            'dart_rate': dart_rate,
            'safety_rating': 'Excellent' if incident_rate < 1.0 else
                           'Good' if incident_rate < 3.0 else
                           'Needs Improvement'
        }

    def _notify_incident(self, incident: SafetyIncident):
        """Notify stakeholders of incident"""
        # Implementation would send notifications
        pass

    def _generate_inspection_id(self) -> str:
        import uuid
        return f"INS-{uuid.uuid4().hex[:8].upper()}"

    def _generate_incident_id(self) -> str:
        import uuid
        return f"INC-{uuid.uuid4().hex[:8].upper()}"

BIM統合

class BIMManagement:
    """Building Information Modeling management"""

    def __init__(self):
        self.models = {}
        self.clash_detections = []

    def perform_clash_detection(self, model_ids: List[str]) -> dict:
        """Detect clashes between BIM models"""
        # Simulate clash detection between disciplines
        # In production, would use BIM software APIs (Revit, Navisworks)

        clashes = [
            {
                'clash_id': 'CLASH-001',
                'type': 'hard',  # 'hard' or 'soft'
                'disciplines': ['structural', 'mep'],
                'description': 'Steel beam conflicts with HVAC duct',
                'location': 'Level 3, Grid B-4',
                'severity': 'high',
                'status': 'open'
            },
            {
                'clash_id': 'CLASH-002',
                'type': 'soft',
                'disciplines': ['architectural', 'mep'],
                'description': 'Insufficient clearance for plumbing access',
                'location': 'Level 2, Grid C-2',
                'severity': 'medium',
                'status': 'open'
            }
        ]

        return {
            'models_analyzed': model_ids,
            'total_clashes': len(clashes),
            'hard_clashes': len([c for c in clashes if c['type'] == 'hard']),
            'soft_clashes': len([c for c in clashes if c['type'] == 'soft']),
            'clashes': clashes
        }

    def extract_quantities(self, model_id: str) -> dict:
        """Extract material quantities from BIM model"""
        # Simulate quantity takeoff
        # In production, would extract from actual BIM model

        quantities = {
            'concrete': {
                'unit': 'cubic_yards',
                'quantity': 1250,
                'cost_per_unit': 150,
                'total_cost': 187500
            },
            'rebar': {
                'unit': 'tons',
                'quantity': 85,
                'cost_per_unit': 800,
                'total_cost': 68000
            },
            'structural_steel': {
                'unit': 'tons',
                'quantity': 120,
                'cost_per_unit': 1200,
                'total_cost': 144000
            }
        }

        total_cost = sum(item['total_cost'] for item in quantities.values())

        return {
            'model_id': model_id,
            'quantities': quantities,
            'total_estimated_cost': total_cost
        }

ベストプラクティス

プロジェクト管理

• クリティカルパス法を使用したスケジューリング
• 定期的な進捗レビューの実施
• 詳細なドキュメンテーションの保守
• 統合プロジェクト配信 (IPD) の使用
• リーン建設原則の実装
• 主要な成績評価指標 (KPI) の追跡
• 定期的なステークホルダー会議の実施

コスト管理

• 詳細な見積もりの作成
• 継続的なコスト追跡
• 変更指示書の効果的な管理
• バリューエンジニアリングの実装
• コストコーディングシステムの実装
• キャッシュフロー監視
• 定期的な監査の実施

安全管理

• 包括的な安全プログラムの実装
• 定期的なツールボックストークの実施
• 適切なPPEの提供
• OSHA準拠の維持
• すべてのインシデントの調査
• 安全メトリクスの追跡
• 安全文化の推進

BIM実装

• クラッシュ検出のためのBIM使用
• 4Dスケジューリングの実装
• モデルからの数量抽出
• コラボレーションの実現
• モデル調整の維持
• ファシリティ管理のためのBIM使用
• ISO 19650基準に従う

アンチパターン

❌ 不十分なプロジェクト計画 ❌ 不十分なコスト追跡 ❌ 安全プログラムなし ❌ 悪いコミュニケーション ❌ 変更指示書の無視 ❌ 品質管理なし ❌ 不十分なドキュメンテーション ❌ 下請業者管理の不備 ❌ リスク管理なし

リソース

• AIA (American Institute of Architects): https://www.aia.org/
• AGC (Associated General Contractors): https://www.agc.org/
• OSHA: https://www.osha.gov/construction
• International Building Code: https://www.iccsafe.org/
• buildingSMART (BIM): https://www.buildingsmart.org/
• PMI Construction Extension: https://www.pmi.org/
• LEED Certification: https://www.usgbc.org/leed