83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
from sklearn.cluster import DBSCAN
|
||
from sklearn.preprocessing import StandardScaler
|
||
import os
|
||
|
||
|
||
class GPSCluster:
|
||
def __init__(self, gps_points, output_dir: str, eps=0.01, min_samples=5):
|
||
"""
|
||
初始化GPS聚类器
|
||
|
||
参数:
|
||
eps: DBSCAN的邻域半径参数
|
||
min_samples: DBSCAN的最小样本数参数
|
||
"""
|
||
self.eps = eps
|
||
self.min_samples = min_samples
|
||
self.dbscan = DBSCAN(eps=eps, min_samples=min_samples)
|
||
self.scaler = StandardScaler()
|
||
self.gps_points = gps_points
|
||
|
||
def fit(self):
|
||
"""
|
||
对GPS点进行聚类，只保留最大的类
|
||
|
||
参数:
|
||
gps_points: 包含'lat'和'lon'列的DataFrame
|
||
|
||
返回:
|
||
带有聚类标签的DataFrame，其中最大类标记为1，其他点标记为-1
|
||
"""
|
||
# 提取经纬度数据
|
||
X = self.gps_points[["lon", "lat"]].values
|
||
|
||
# # 数据标准化
|
||
# X_scaled = self.scaler.fit_transform(X)
|
||
|
||
# 执行DBSCAN聚类
|
||
labels = self.dbscan.fit_predict(X)
|
||
|
||
# 找出最大类的标签（排除噪声点-1）
|
||
unique_labels = [l for l in set(labels) if l != -1]
|
||
if unique_labels: # 如果有聚类
|
||
label_counts = [(l, sum(labels == l)) for l in unique_labels]
|
||
largest_label = max(label_counts, key=lambda x: x[1])[0]
|
||
|
||
# 将最大类标记为1，其他都标记为-1
|
||
new_labels = (labels == largest_label).astype(int)
|
||
new_labels[new_labels == 0] = -1
|
||
else: # 如果没有聚类，全部标记为-1
|
||
new_labels = labels
|
||
|
||
# 将聚类结果添加到原始数据中
|
||
result_df = self.gps_points.copy()
|
||
result_df["cluster"] = new_labels
|
||
|
||
return result_df
|
||
|
||
def get_cluster_stats(self, clustered_points):
|
||
"""
|
||
获取聚类统计信息
|
||
|
||
参数:
|
||
clustered_points: 带有聚类标签的DataFrame
|
||
|
||
返回:
|
||
聚类统计信息的字典
|
||
"""
|
||
main_cluster_points = sum(clustered_points["cluster"] == 1)
|
||
stats = {
|
||
"total_points": len(clustered_points),
|
||
"main_cluster_points": main_cluster_points,
|
||
"noise_points": sum(clustered_points["cluster"] == -1),
|
||
}
|
||
|
||
noise_cluster = self.get_noise_cluster(clustered_points)
|
||
return stats
|
||
|
||
def get_main_cluster(self, clustered_points):
|
||
return clustered_points[clustered_points["cluster"] == 1]
|
||
|
||
def get_noise_cluster(self, clustered_points):
|
||
return clustered_points[clustered_points["cluster"] == -1]
|