技能名称: open-data-integrator

开放数据集成器

概述

基于DDC方法论（第2.2章），本技能整合来自政府数据库、行业基准、气象服务和地理空间数据等多种来源的开放建筑数据集。

参考书籍：《开放数据主导》/《Open Data Dominance》

快速开始

python
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Optional, Any, Callable
from datetime import datetime, date
import json
import requests
from abc import ABC, abstractmethod

class DataSourceType(Enum):
开放数据源类型
GOVERNMENT = government # 政府统计数据
INDUSTRY_BENCHMARK = benchmark # 行业基准
WEATHER = weather # 气象数据
GEOSPATIAL = geospatial # 地理数据
MATERIALPRICES = materialprices # 材料成本指数
LABORRATES = laborrates # 劳动力成本数据
BUILDING_PERMITS = permits # 许可证数据
ENERGY = energy # 能源价格/数据
ECONOMIC = economic # 经济指标

class UpdateFrequency(Enum):
数据更新频率
REALTIME = realtime
HOURLY = hourly
DAILY = daily
WEEKLY = weekly
MONTHLY = monthly
QUARTERLY = quarterly
ANNUAL = annual

@dataclass
class OpenDataSource:
开放数据源定义
id: str
name: str
source_type: DataSourceType
url: str
apikeyrequired: bool = False
update_frequency: UpdateFrequency = UpdateFrequency.DAILY
format: str = json
license: str = open
description: Optional[str] = None
fields: List[str] = field(default_factory=list)

@dataclass
class DataRecord:
来自数据源的单个数据记录
source_id: str
timestamp: datetime
data: Dict[str, Any]
metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class IntegrationResult:
数据集成结果
source: str
records_fetched: int
records_processed: int
errors: List[str]
last_updated: datetime
sample_data: List[Dict]

@dataclass
class EnrichedData:
用开放数据增强后的数据
original_data: Dict[str, Any]
enrichments: Dict[str, Any]
sources_used: List[str]
confidence: float

class OpenDataConnector(ABC):
开放数据连接器基类

@abstractmethod
def fetch(self, params: Dict) -> List[DataRecord]:
pass

@abstractmethod
def get_metadata(self) -> Dict:
pass

class WeatherDataConnector(OpenDataConnector):
气象数据连接器（例如 OpenWeatherMap）

def init(self, api_key: Optional[str] = None):
self.apikey = apikey
self.base_url = https://api.openweathermap.org/data/2.5

def fetch(
self,
params: Dict
) -> List[DataRecord]:
获取指定位置的气象数据
lat = params.get(lat)
lon = params.get(lon)
startdate = params.get(startdate)
enddate = params.get(enddate)

# 模拟 API 调用（生产环境中使用实际 API）
records = []

# 生成样本历史数据
current = start_date
while current <= end_date:
records.append(DataRecord(
source_id=openweathermap,
timestamp=datetime.combine(current, datetime.min.time()),
data={
date: current.isoformat(),
temp_max: 25.0,
temp_min: 15.0,
precipitation: 0.0,
wind_speed: 10.0,
weather_code: clear
},
metadata={lat: lat, lon: lon}
))
current = date(current.year, current.month, current.day + 1) if current.day < 28 else date(current.year, current.month + 1 if current.month < 12 else 1, 1)

return records[:30] # 演示限制

def get_metadata(self) -> Dict:
return {
source: OpenWeatherMap,
type: DataSourceType.WEATHER.value,
frequency: UpdateFrequency.HOURLY.value,
fields: [tempmax, tempmin, precipitation, wind_speed]
}

class MaterialPriceConnector(OpenDataConnector):
材料价格指数连接器

def init(self, region: str = US):
self.region = region
self.priceindices = self.load_indices()

def loadindices(self) -> Dict[str, Dict]:
加载材料价格指数
return {
concrete: {base: 100, current: 125, trend: up},
steel: {base: 100, current: 145, trend: up},
lumber: {base: 100, current: 180, trend: stable},
copper: {base: 100, current: 135, trend: up},
asphalt: {base: 100, current: 115, trend: stable},
gypsum: {base: 100, current: 110, trend: stable},
glass: {base: 100, current: 105, trend: down},
cement: {base: 100, current: 120, trend: up},
}

def fetch(self, params: Dict) -> List[DataRecord]:
获取材料价格数据
materials = params.get(materials, list(self.price_indices.keys()))

records = []
for material in materials:
if material in self.price_indices:
records.append(DataRecord(
sourceid=materialprices,
timestamp=datetime.now(),
data={
material: material,
region: self.region,
self.price_indices[material]
}
))
return records

def get_metadata(self) -> Dict:
return {
source: Material Price Index,
type: DataSourceType.MATERIAL_PRICES.value,
frequency: UpdateFrequency.MONTHLY.value,
materials: list(self.price_indices.keys())
}

class LaborRateConnector(OpenDataConnector):
劳动力费率数据连接器

def init(self, region: str = US):
self.region = region
self.laborrates = self.load_rates()

def loadrates(self) -> Dict[str, Dict]:
按工种加载劳动力费率
return {
carpenter: {hourly: 45.00, burden_rate: 1.35},
electrician: {hourly: 55.00, burden_rate: 1.40},
plumber: {hourly: 52.00, burden_rate: 1.38},
ironworker: {hourly: 58.00, burden_rate: 1.42},
laborer: {hourly: 32.00, burden_rate: 1.30},
operator: {hourly: 48.00, burden_rate: 1.35},
mason: {hourly: 50.00, burden_rate: 1.36},
painter: {hourly: 38.00, burden_rate: 1.32},
hvactech: {hourly: 54.00, burdenrate: 1.38},
welder: {hourly: 52.00, burden_rate: 1.40},
}

def fetch(self, params: Dict) -> List[DataRecord]:
获取劳动力费率数据
trades = params.get(trades, list(self.labor_rates.keys()))

records = []
for trade in trades:
if trade in self.labor_rates:
ratedata = self.laborrates[trade]
records.append(DataRecord(
sourceid=laborrates,
timestamp=datetime.now(),
data={
trade: trade,
region: self.region,
hourlyrate: ratedata[hourly],
burdenrate: ratedata[burden_rate],
fullyloaded: ratedata[hourly] * ratedata[burdenrate]