@datadrivenconstruction/compliance-tracker

---
name: "compliance-tracker"
description: "Track regulatory compliance for construction projects. Monitor permits, certifications, inspections, and regulatory requirements with automated alerts and reporting."
homepage: "https://datadrivenconstruction.io"
metadata: {"openclaw": {"emoji": "🦺", "os": ["darwin", "linux", "win32"], "homepage": "https://datadrivenconstruction.io", "requires": {"bins": ["python3"]}}}
---
# Compliance Tracker

## Overview

Track and manage regulatory compliance across construction projects. Monitor permits, licenses, certifications, inspections, and regulatory requirements. Automated alerts for expirations and deadlines.

> "Proactive compliance tracking prevents costly project delays and penalties" — DDC Community

## Compliance Categories

```
┌─────────────────────────────────────────────────────────────────┐
│                    COMPLIANCE TRACKING                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  Permits          Certifications    Inspections    Training     │
│  ───────          ──────────────    ───────────    ────────     │
│  🏗️ Building      👷 Workers         📋 Fire        🎓 OSHA      │
│  🔥 Fire          🏢 Company         ⚡ Electrical   🦺 Safety    │
│  ⚡ Electrical    🔧 Equipment       🔧 Mechanical   🏗️ Trade     │
│  🚰 Plumbing      📜 Insurance       🏗️ Structural  🚗 Equipment │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘
```

## Technical Implementation

```python
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
from datetime import datetime, timedelta
import json

class ComplianceType(Enum):
    PERMIT = "permit"
    LICENSE = "license"
    CERTIFICATION = "certification"
    INSURANCE = "insurance"
    INSPECTION = "inspection"
    TRAINING = "training"
    SUBMITTAL = "submittal"

class ComplianceStatus(Enum):
    ACTIVE = "active"
    PENDING = "pending"
    EXPIRED = "expired"
    EXPIRING_SOON = "expiring_soon"
    NOT_APPLICABLE = "not_applicable"
    REJECTED = "rejected"

class Priority(Enum):
    CRITICAL = "critical"  # Work stoppage if missing
    HIGH = "high"          # Significant impact
    MEDIUM = "medium"      # Moderate impact
    LOW = "low"            # Administrative

@dataclass
class ComplianceItem:
    id: str
    name: str
    compliance_type: ComplianceType
    category: str
    description: str

    # Dates
    issue_date: Optional[datetime] = None
    expiration_date: Optional[datetime] = None
    renewal_date: Optional[datetime] = None

    # Status
    status: ComplianceStatus = ComplianceStatus.PENDING
    priority: Priority = Priority.MEDIUM

    # Ownership
    responsible_party: str = ""
    issuing_authority: str = ""

    # Documentation
    document_url: str = ""
    reference_number: str = ""
    notes: str = ""

    # Tracking
    alert_days_before: int = 30
    last_checked: Optional[datetime] = None

@dataclass
class ComplianceAlert:
    id: str
    compliance_item_id: str
    alert_type: str  # expiring, expired, action_required
    message: str
    due_date: datetime
    acknowledged: bool = False
    acknowledged_by: str = ""

class ComplianceTracker:
    """Track construction regulatory compliance."""

    # Standard compliance requirements by project type
    STANDARD_REQUIREMENTS = {
        "commercial": [
            {"name": "Building Permit", "type": ComplianceType.PERMIT, "category": "Building", "priority": Priority.CRITICAL},
            {"name": "Fire Permit", "type": ComplianceType.PERMIT, "category": "Fire", "priority": Priority.CRITICAL},
            {"name": "Electrical Permit", "type": ComplianceType.PERMIT, "category": "Electrical", "priority": Priority.HIGH},
            {"name": "Plumbing Permit", "type": ComplianceType.PERMIT, "category": "Plumbing", "priority": Priority.HIGH},
            {"name": "Mechanical Permit", "type": ComplianceType.PERMIT, "category": "Mechanical", "priority": Priority.HIGH},
            {"name": "General Liability Insurance", "type": ComplianceType.INSURANCE, "category": "Insurance", "priority": Priority.CRITICAL},
            {"name": "Workers Comp Insurance", "type": ComplianceType.INSURANCE, "category": "Insurance", "priority": Priority.CRITICAL},
            {"name": "OSHA 10/30 Training", "type": ComplianceType.TRAINING, "category": "Safety", "priority": Priority.HIGH},
        ],
        "residential": [
            {"name": "Building Permit", "type": ComplianceType.PERMIT, "category": "Building", "priority": Priority.CRITICAL},
            {"name": "Electrical Permit", "type": ComplianceType.PERMIT, "category": "Electrical", "priority": Priority.HIGH},
            {"name": "Plumbing Permit", "type": ComplianceType.PERMIT, "category": "Plumbing", "priority": Priority.HIGH},
            {"name": "General Liability Insurance", "type": ComplianceType.INSURANCE, "category": "Insurance", "priority": Priority.CRITICAL},
        ]
    }

    # Required inspections by permit type
    REQUIRED_INSPECTIONS = {
        "Building": ["Foundation", "Framing", "Insulation", "Final"],
        "Electrical": ["Rough-in", "Service", "Final"],
        "Plumbing": ["Underground", "Rough-in", "Final"],
        "Mechanical": ["Rough-in", "Final"],
        "Fire": ["Underground", "Rough-in", "Final", "Alarm"]
    }

    def __init__(self, project_id: str, project_name: str):
        self.project_id = project_id
        self.project_name = project_name
        self.compliance_items: Dict[str, ComplianceItem] = {}
        self.alerts: List[ComplianceAlert] = []

    def initialize_requirements(self, project_type: str = "commercial") -> List[ComplianceItem]:
        """Initialize standard compliance requirements."""
        requirements = self.STANDARD_REQUIREMENTS.get(project_type, [])
        created = []

        for req in requirements:
            item = self.add_compliance_item(
                name=req["name"],
                compliance_type=req["type"],
                category=req["category"],
                description=f"Standard {req['name']} requirement",
                priority=req["priority"]
            )
            created.append(item)

        return created

    def add_compliance_item(self, name: str, compliance_type: ComplianceType,
                           category: str, description: str = "",
                           priority: Priority = Priority.MEDIUM,
                           expiration_date: datetime = None,
                           responsible_party: str = "",
                           issuing_authority: str = "") -> ComplianceItem:
        """Add compliance item to track."""
        item_id = f"COMP-{datetime.now().strftime('%Y%m%d%H%M%S')}-{len(self.compliance_items)}"

        item = ComplianceItem(
            id=item_id,
            name=name,
            compliance_type=compliance_type,
            category=category,
            description=description,
            priority=priority,
            expiration_date=expiration_date,
            responsible_party=responsible_party,
            issuing_authority=issuing_authority
        )

        self.compliance_items[item_id] = item
        return item

    def update_status(self, item_id: str, status: ComplianceStatus,
                     issue_date: datetime = None,
                     expiration_date: datetime = None,
                     reference_number: str = "",
                     document_url: str = "") -> ComplianceItem:
        """Update compliance item status."""
        if item_id not in self.compliance_items:
            raise ValueError(f"Compliance item {item_id} not found")

        item = self.compliance_items[item_id]
        item.status = status
        item.last_checked = datetime.now()

        if issue_date:
            item.issue_date = issue_date
        if expiration_date:
            item.expiration_date = expiration_date
        if reference_number:
            item.reference_number = reference_number
        if document_url:
            item.document_url = document_url

        return item

    def check_all_status(self) -> List[ComplianceAlert]:
        """Check status of all items and generate alerts."""
        new_alerts = []
        today = datetime.now()

        for item in self.compliance_items.values():
            # Skip non-applicable items
            if item.status == ComplianceStatus.NOT_APPLICABLE:
                continue

            # Check for expired
            if item.expiration_date and item.expiration_date < today:
                if item.status != ComplianceStatus.EXPIRED:
                    item.status = ComplianceStatus.EXPIRED
                    alert = self._create_alert(item, "expired",
                        f"EXPIRED: {item.name} expired on {item.expiration_date.strftime('%Y-%m-%d')}")
                    new_alerts.append(alert)

            # Check for expiring soon
            elif item.expiration_date:
                days_until = (item.expiration_date - today).days
                if days_until <= item.alert_days_before:
                    if item.status != ComplianceStatus.EXPIRING_SOON:
                        item.status = ComplianceStatus.EXPIRING_SOON
                        alert = self._create_alert(item, "expiring",
                            f"EXPIRING: {item.name} expires in {days_until} days")
                        new_alerts.append(alert)

            # Check pending items
            elif item.status == ComplianceStatus.PENDING:
                if item.priority == Priority.CRITICAL:
                    alert = self._create_alert(item, "action_required",
                        f"ACTION REQUIRED: {item.name} is pending - Critical priority")
                    new_alerts.append(alert)

        self.alerts.extend(new_alerts)
        return new_alerts

    def _create_alert(self, item: ComplianceItem, alert_type: str, message: str) -> ComplianceAlert:
        """Create compliance alert."""
        return ComplianceAlert(
            id=f"ALERT-{datetime.now().strftime('%Y%m%d%H%M%S')}",
            compliance_item_id=item.id,
            alert_type=alert_type,
            message=message,
            due_date=item.expiration_date or datetime.now()
        )

    def get_compliance_summary(self) -> Dict:
        """Get compliance status summary."""
        total = len(self.compliance_items)
        by_status = {}
        by_type = {}
        by_priority = {}

        for item in self.compliance_items.values():
            # By status
            status = item.status.value
            by_status[status] = by_status.get(status, 0) + 1

            # By type
            comp_type = item.compliance_type.value
            by_type[comp_type] = by_type.get(comp_type, 0) + 1

            # By priority
            priority = item.priority.value
            by_priority[priority] = by_priority.get(priority, 0) + 1

        # Calculate compliance rate
        active = by_status.get("active", 0)
        compliance_rate = (active / total * 100) if total else 0

        critical_missing = len([i for i in self.compliance_items.values()
                               if i.priority == Priority.CRITICAL
                               and i.status in [ComplianceStatus.PENDING, ComplianceStatus.EXPIRED]])

        return {
            "total_items": total,
            "compliance_rate": compliance_rate,
            "by_status": by_status,
            "by_type": by_type,
            "by_priority": by_priority,
            "critical_missing": critical_missing,
            "active_alerts": len([a for a in self.alerts if not a.acknowledged])
        }

    def get_expiring_items(self, days: int = 30) -> List[ComplianceItem]:
        """Get items expiring within specified days."""
        cutoff = datetime.now() + timedelta(days=days)
        return [item for item in self.compliance_items.values()
                if item.expiration_date and item.expiration_date <= cutoff
                and item.status != ComplianceStatus.EXPIRED]

    def get_required_inspections(self, permit_category: str) -> List[str]:
        """Get required inspections for permit type."""
        return self.REQUIRED_INSPECTIONS.get(permit_category, [])

    def schedule_inspection(self, permit_id: str, inspection_name: str,
                           scheduled_date: datetime) -> ComplianceItem:
        """Schedule required inspection."""
        if permit_id not in self.compliance_items:
            raise ValueError(f"Permit {permit_id} not found")

        permit = self.compliance_items[permit_id]

        inspection = self.add_compliance_item(
            name=f"{permit.category} - {inspection_name} Inspection",
            compliance_type=ComplianceType.INSPECTION,
            category=permit.category,
            description=f"Required inspection for {permit.name}",
            priority=Priority.HIGH,
            expiration_date=scheduled_date
        )

        return inspection

    def generate_compliance_report(self) -> str:
        """Generate compliance status report."""
        summary = self.get_compliance_summary()

        lines = [
            f"# Compliance Status Report",
            f"",
            f"**Project:** {self.project_name}",
            f"**Date:** {datetime.now().strftime('%Y-%m-%d')}",
            f"**Compliance Rate:** {summary['compliance_rate']:.1f}%",
            f"",
            f"## Summary",
            f"",
            f"| Status | Count |",
            f"|--------|-------|",
        ]

        for status, count in summary['by_status'].items():
            lines.append(f"| {status.title()} | {count} |")

        # Critical items
        critical_pending = [i for i in self.compliance_items.values()
                          if i.priority == Priority.CRITICAL
                          and i.status == ComplianceStatus.PENDING]
        if critical_pending:
            lines.extend([
                f"",
                f"## Critical Items Pending",
                f""
            ])
            for item in critical_pending:
                lines.append(f"- **{item.name}** - {item.responsible_party or 'Unassigned'}")

        # Expiring items
        expiring = self.get_expiring_items(30)
        if expiring:
            lines.extend([
                f"",
                f"## Items Expiring in 30 Days",
                f""
            ])
            for item in expiring:
                days = (item.expiration_date - datetime.now()).days
                lines.append(f"- **{item.name}** - Expires in {days} days ({item.expiration_date.strftime('%Y-%m-%d')})")

        # Active alerts
        active_alerts = [a for a in self.alerts if not a.acknowledged]
        if active_alerts:
            lines.extend([
                f"",
                f"## Active Alerts ({len(active_alerts)})",
                f""
            ])
            for alert in active_alerts[:10]:
                lines.append(f"- {alert.message}")

        return "\n".join(lines)
```

## Quick Start

```python
# Initialize tracker
tracker = ComplianceTracker("PRJ-001", "Office Tower")

# Initialize standard requirements
items = tracker.initialize_requirements("commercial")
print(f"Initialized {len(items)} compliance items")

# Update permit status
building_permit = [i for i in tracker.compliance_items.values()
                   if i.name == "Building Permit"][0]

tracker.update_status(
    building_permit.id,
    status=ComplianceStatus.ACTIVE,
    issue_date=datetime.now(),
    expiration_date=datetime.now() + timedelta(days=365),
    reference_number="BP-2024-12345"
)

# Check all status and generate alerts
alerts = tracker.check_all_status()
print(f"Generated {len(alerts)} alerts")

# Get summary
summary = tracker.get_compliance_summary()
print(f"Compliance Rate: {summary['compliance_rate']:.1f}%")
print(f"Critical Missing: {summary['critical_missing']}")

# Generate report
print(tracker.generate_compliance_report())
```

## Requirements

```bash
pip install (no external dependencies)
```