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

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

# 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

pip install (no external dependencies)