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.
日本語の概要は準備中です。原文の説明を表示しています。