Analyze large-scale construction datasets. Process thousands of projects for patterns, benchmarks, and predictive insights.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
설치하기 전에 에이전트에게 전달되는 지시 내용을 확인할 수 있습니다.
Large-scale data analysis challenges:
Scalable big data analysis framework for construction data using efficient data structures and parallel processing patterns.
import pandas as pd
from typing import Dict, Any, List, Optional, Callable, Iterator
from dataclasses import dataclass, field
from datetime import datetime, date
from enum import Enum
import json
class AnalysisType(Enum):
BENCHMARK = "benchmark"
TREND = "trend"
ANOMALY = "anomaly"
CORRELATION = "correlation"
CLUSTERING = "clustering"
AGGREGATION = "aggregation"
class MetricType(Enum):
COST_PER_SF = "cost_per_sf"
DURATION_PER_SF = "duration_per_sf"
PRODUCTIVITY = "productivity"
CHANGE_ORDER_RATE = "change_order_rate"
SAFETY_RATE = "safety_rate"
QUALITY_SCORE = "quality_score"
@dataclass
class ProjectRecord:
project_id: str
name: str
project_type: str
location: str
size_sf: float
duration_days: int
total_cost: float
start_date: date
metrics: Dict[str, float] = field(default_factory=dict)
attributes: Dict[str, Any] = field(default_factory=dict)
@dataclass
class BenchmarkResult:
metric: str
mean: float
median: float
std: float
min_val: float
max_val: float
percentile_25: float
percentile_75: float
sample_size: int
class BigDataAnalyzer:
"""Analyze large-scale construction datasets."""
def __init__(self, name: str = "Construction Analytics"):
self.name = name
self.projects: List[ProjectRecord] = []
self.df: Optional[pd.DataFrame] = None
self.benchmarks: Dict[str, BenchmarkResult] = {}
def load_from_dataframe(self, df: pd.DataFrame):
"""Load project data from DataFrame."""
self.df = df.copy()
self.projects = []
for _, row in df.iterrows():
project = ProjectRecord(
project_id=str(row.get('project_id', '')),
name=str(row.get('name', '')),
project_type=str(row.get('project_type', '')),
location=str(row.get('location', '')),
size_sf=float(row.get('size_sf', 0)),
duration_days=int(row.get('duration_days', 0)),
total_cost=float(row.get('total_cost', 0)),
start_date=pd.to_datetime(row.get('start_date')).date() if pd.notna(row.get('start_date')) else date.today()
)
# Add calculated metrics
if project.size_sf > 0:
project.metrics['cost_per_sf'] = project.total_cost / project.size_sf
project.metrics['duration_per_1000sf'] = project.duration_days / (project.size_sf / 1000)
self.projects.append(project)
def load_from_parquet(self, path: str):
"""Load data from Parquet file."""
df = pd.read_parquet(path)
self.load_from_dataframe(df)
def stream_process(self, file_path: str, chunk_size: int = 10000,
processor: Callable = None) -> Iterator[Dict[str, Any]]:
"""Process large file in chunks."""
for chunk in pd.read_csv(file_path, chunksize=chunk_size):
if processor:
result = processor(chunk)
yield result
else:
yield {'rows': len(chunk), 'columns': list(chunk.columns)}
def calculate_benchmarks(self, metric_column: str,
group_by: str = None) -> Dict[str, BenchmarkResult]:
"""Calculate benchmarks for a metric."""
if self.df is None or self.df.empty:
return {}
results = {}
if group_by and group_by in self.df.columns:
groups = self.df.groupby(group_by)
for group_name, group_df in groups:
values = group_df[metric_column].dropna()
if len(values) > 0:
results[str(group_name)] = self._calculate_stats(values, metric_column)
else:
values = self.df[metric_column].dropna()
if len(values) > 0:
results['all'] = self._calculate_stats(values, metric_column)
self.benchmarks.update(results)
return results
def _calculate_stats(self, values: pd.Series, metric: str) -> BenchmarkResult:
"""Calculate statistics for a series."""
return BenchmarkResult(
metric=metric,
mean=round(values.mean(), 2),
median=round(values.median(), 2),
std=round(values.std(), 2),
min_val=round(values.min(), 2),
max_val=round(values.max(), 2),
percentile_25=round(values.quantile(0.25), 2),
percentile_75=round(values.quantile(0.75), 2),
sample_size=len(values)
)
def find_anomalies(self, metric_column: str,
threshold_std: float = 2.0) -> pd.DataFrame:
"""Find anomalies based on standard deviation threshold."""
if self.df is None or self.df.empty:
return pd.DataFrame()
values = self.df[metric_column]
mean = values.mean()
std = values.std()
lower = mean - (threshold_std * std)
upper = mean + (threshold_std * std)
anomalies = self.df[(values < lower) | (values > upper)].copy()
anomalies['anomaly_type'] = anomalies[metric_column].apply(
lambda x: 'high' if x > upper else 'low'
)
anomalies['deviation'] = ((anomalies[metric_column] - mean) / std).round(2)
return anomalies
def analyze_trends(self, metric_column: str,
date_column: str,
period: str = 'Y') -> pd.DataFrame:
"""Analyze trends over time."""
if self.df is None or self.df.empty:
return pd.DataFrame()
df = self.df.copy()
df[date_column] = pd.to_datetime(df[date_column])
df['period'] = df[date_column].dt.to_period(period)
trends = df.groupby('period').agg({
metric_column: ['mean', 'median', 'count', 'std']
}).round(2)
trends.columns = ['mean', 'median', 'count', 'std']
trends = trends.reset_index()
trends['period'] = trends['period'].astype(str)
# Calculate year-over-year change
trends['yoy_change'] = trends['mean'].pct_change().round(4) * 100
return trends
def calculate_correlations(self, columns: List[str]) -> pd.DataFrame:
"""Calculate correlations between metrics."""
if self.df is None or self.df.empty:
return pd.DataFrame()
available_cols = [c for c in columns if c in self.df.columns]
return self.df[available_cols].corr().round(3)
def segment_analysis(self, metric_column: str,
segment_column: str) -> pd.DataFrame:
"""Analyze metric by segments."""
if self.df is None or self.df.empty:
return pd.DataFrame()
results = self.df.groupby(segment_column).agg({
metric_column: ['count', 'mean', 'median', 'std', 'min', 'max']
}).round(2)
results.columns = ['count', 'mean', 'median', 'std', 'min', 'max']
results = results.reset_index()
# Calculate percentage of total
total_count = results['count'].sum()
results['pct_of_total'] = (results['count'] / total_count * 100).round(1)
return results.sort_values('count', ascending=False)
def percentile_rank(self, project_id: str, metric_column: str) -> Dict[str, Any]:
"""Get percentile rank for a specific project."""
if self.df is None or self.df.empty:
return {}
project = self.df[self.df['project_id'] == project_id]
if project.empty:
return {'error': 'Project not found'}
value = project[metric_column].values[0]
all_values = self.df[metric_column].dropna()
percentile = (all_values < value).sum() / len(all_values) * 100
benchmark = self.benchmarks.get('all') or self._calculate_stats(all_values, metric_column)
return {
'project_id': project_id,
'metric': metric_column,
'value': round(value, 2),
'percentile': round(percentile, 1),
'comparison': {
'mean': benchmark.mean,
'median': benchmark.median,
'vs_mean': round((value / benchmark.mean - 1) * 100, 1),
'vs_median': round((value / benchmark.median - 1) * 100, 1)
}
}
def generate_summary_stats(self) -> Dict[str, Any]:
"""Generate summary statistics for the dataset."""
if self.df is None or self.df.empty:
return {}
numeric_cols = self.df.select_dtypes(include=['number']).columns
return {
'total_projects': len(self.df),
'date_range': {
'min': str(self.df['start_date'].min()) if 'start_date' in self.df.columns else None,
'max': str(self.df['start_date'].max()) if 'start_date' in self.df.columns else None
},
'project_types': self.df['project_type'].nunique() if 'project_type' in self.df.columns else 0,
'locations': self.df['location'].nunique() if 'location' in self.df.columns else 0,
'total_value': self.df['total_cost'].sum() if 'total_cost' in self.df.columns else 0,
'total_sf': self.df['size_sf'].sum() if 'size_sf' in self.df.columns else 0,
'numeric_columns': list(numeric_cols)
}
def export_analysis(self, output_path: str,
metrics: List[str] = None) -> str:
"""Export analysis results to Excel."""
metrics = metrics or ['cost_per_sf', 'duration_days']
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Summary
summary = self.generate_summary_stats()
summary_df = pd.DataFrame([{
'Total Projects': summary.get('total_projects', 0),
'Project Types': summary.get('project_types', 0),
'Locations': summary.get('locations', 0),
'Total Value ($)': summary.get('total_value', 0),
'Total SF': summary.get('total_sf', 0)
}])
summary_df.to_excel(writer, sheet_name='Summary', index=False)
# Benchmarks
for metric in metrics:
if metric in self.df.columns:
benchmarks = self.calculate_benchmarks(metric, 'project_type')
if benchmarks:
bench_data = [{
'Segment': k,
'Mean': v.mean,
'Median': v.median,
'Std': v.std,
'Min': v.min_val,
'Max': v.max_val,
'P25': v.percentile_25,
'P75': v.percentile_75,
'Count': v.sample_size
} for k, v in benchmarks.items()]
bench_df = pd.DataFrame(bench_data)
sheet_name = f"Benchmark_{metric}"[:31]
bench_df.to_excel(writer, sheet_name=sheet_name, index=False)
# Anomalies
for metric in metrics:
if metric in self.df.columns:
anomalies = self.find_anomalies(metric)
if not anomalies.empty:
sheet_name = f"Anomalies_{metric}"[:31]
anomalies.to_excel(writer, sheet_name=sheet_name, index=False)
return output_path
# Create analyzer
analyzer = BigDataAnalyzer("Multi-Project Analysis")
# Load data
df = pd.DataFrame([
{'project_id': 'P001', 'name': 'Office A', 'project_type': 'Office',
'location': 'NYC', 'size_sf': 50000, 'duration_days': 365,
'total_cost': 15000000, 'start_date': '2023-01-01'},
{'project_id': 'P002', 'name': 'Office B', 'project_type': 'Office',
'location': 'LA', 'size_sf': 75000, 'duration_days': 400,
'total_cost': 20000000, 'start_date': '2023-03-01'},
{'project_id': 'P003', 'name': 'Warehouse', 'project_type': 'Industrial',
'location': 'Chicago', 'size_sf': 100000, 'duration_days': 200,
'total_cost': 12000000, 'start_date': '2023-06-01'}
])
# Add calculated metric
df['cost_per_sf'] = df['total_cost'] / df['size_sf']
analyzer.load_from_dataframe(df)
# Calculate benchmarks
benchmarks = analyzer.calculate_benchmarks('cost_per_sf', 'project_type')
for segment, stats in benchmarks.items():
print(f"{segment}: ${stats.mean:.2f}/SF (median: ${stats.median:.2f}/SF)")
anomalies = analyzer.find_anomalies('cost_per_sf', threshold_std=2.0)
print(f"Found {len(anomalies)} anomalous projects")
trends = analyzer.analyze_trends('cost_per_sf', 'start_date', 'Y')
print(trends)
ranking = analyzer.percentile_rank('P001', 'cost_per_sf')
print(f"Project is at {ranking['percentile']}th percentile")
segments = analyzer.segment_analysis('cost_per_sf', 'project_type')
print(segments)
아직 리뷰가 없습니다. 사용해 본 소감을 남겨 주세요.
로그인 없이 쓸 수 있습니다. 사용 후기, 질문, 작동하지 않은 환경 등을 편하게 남겨 주세요.
아직 댓글이 없습니다. 첫 댓글을 남겨 보세요.
Create 4D construction simulations by linking BIM elements to project schedules. Generate time-based visualizations, sequence analysis, and construction phasing with Gantt integration.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Analyze 5000+ IFC and Revit projects at scale for patterns, benchmarks, and insights. Big data analysis for construction.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Check construction AI systems against the EU AI Act (Regulation 2024/1689): classify risk of estimation, scheduling, CV and agent tools, document transparency, keep human oversight. Use when deploying or auditing AI in construction.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Orchestrate multiple AI agents for construction workflows: estimator, scheduler, document, QA and safety agents coordinated by a supervisor agent, with human checkpoints. Use when automating end-to-end project processes with agentic AI.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Create Apache Airflow DAGs for construction data pipelines. Orchestrate ETL, validation, and reporting workflows.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Automate as-built documentation and digital handover for construction. Compile project records, generate O&M manuals, create asset databases, and ensure complete project closeout.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.