Initial Commit.

iceyear · iceyear · commit 7f047daea05d · 2023-12-28T11:34:15.000+08:00
diff --git a/evaluate.py b/evaluate.py
@@ -0,0 +1,19 @@
+import numpy as np
+
+def test_generated_gpx(generated_track, mean_coords, std_coords):
+    if generated_track is None or len(generated_track) == 0:
+        print("Generated track is empty.")
+        return
+
+    # 将生成的轨迹数据逆归一化回原始的坐标值
+    generated_track_unnormalized = generated_track * std_coords + mean_coords
+    
+    # 计算生成轨迹的均值和标准差
+    mean_generated = np.mean(generated_track_unnormalized, axis=0)
+    std_generated = np.std(generated_track_unnormalized, axis=0)
+
+    print("Generated GPX track statistics:")
+    print(f"Mean coordinates: {mean_generated}")
+    print(f"Standard deviation of coordinates: {std_generated}")
+
+    # 这里可以添加更多的评估方法，例如与实际轨迹的对比，或者定量评估生成轨迹的质量
diff --git a/main.py b/main.py
@@ -0,0 +1,166 @@
+# main.py
+import argparse
+import glob
+import os
+import sys
+import time
+import numpy as np
+from preprocess import preprocess_gpx
+from model import build_model, train_model, generate_gpx
+
+try:
+    import select
+except ImportError:
+    select = None
+
+try:
+    import msvcrt
+except ImportError:
+    msvcrt = None
+
+# 反标准化函数，将数据恢复到原始的经纬度范围
+def denormalize(lat, lon, mean_coords, std_coords):
+    denorm_lat = (lat * std_coords[0]) + mean_coords[0]
+    denorm_lon = (lon * std_coords[1]) + mean_coords[1]
+    return denorm_lat, denorm_lon
+
+# 保存为 GPX 路线格式的函数
+def save_to_gpx_route(filename, data, mean_coords, std_coords):
+    with open(filename, 'w') as file:
+        # 写入 GPX 文件头
+        file.write('<?xml version="1.0" encoding="UTF-8"?>\n')
+        file.write('<gpx version="1.1" creator="StackAll">\n')
+        file.write(' <rte>\n')
+        file.write('  <name>Generated Route</name>\n')
+        file.write('  <number>1</number>\n')
+        
+        # 写入路线点
+        for point in data:
+            lat, lon = denormalize(point[0], point[1], mean_coords, std_coords)
+            file.write(f'   <rtept lat="{lat}" lon="{lon}"></rtept>\n')
+        
+        # 写入 GPX 文件尾
+        file.write(' </rte>\n')
+        file.write('</gpx>\n')
+
+# 增加了命令行参数解析
+def parse_args():
+    parser = argparse.ArgumentParser(description='GPX Route Generation')
+    parser.add_argument('-ni', '--no-interaction', action='store_true',
+        help='Run without interactive input, using default values')
+    parser.add_argument('-e', '--epochs', type=int, default=100,
+        help='Number of epochs to train the model')
+    parser.add_argument('-p', '--num-points', type=int, default=100,
+        help='Number of points to generate in each GPX route')
+    parser.add_argument('-n', '--num-files', type=int, default=10,
+        help='Number of GPX files to generate')
+    return parser.parse_args()
+
+def cross_platform_input_with_timeout(prompt, timeout, default):
+    print(prompt, end='', flush=True)
+    input_str = ''
+    start_time = time.time()
+
+    while True:
+        if msvcrt:
+            while (time.time() - start_time) < timeout:
+                if msvcrt.kbhit():
+                    char = msvcrt.getche()
+                    if ord(char) == 13:
+                        return input_str or default
+                    if char.decode().isdigit():
+                        input_str += char.decode()
+                    else:
+                        print("\n请输入数字。")
+                        return cross_platform_input_with_timeout(prompt, timeout, default)
+                time.sleep(0.05)
+        else:
+            if (time.time() - start_time) < timeout:
+                rlist, _, _ = select.select([sys.stdin], [], [], timeout)
+                if rlist:
+                    input_str = sys.stdin.readline().rstrip('\n')
+                    if input_str.isdigit() or input_str == '':
+                        return input_str or default
+                    else:
+                        print("请输入数字。")
+                        return cross_platform_input_with_timeout(prompt, timeout, default)
+                timeout -= (time.time() - start_time)
+            else:
+                print()
+                return default
+
+def ensure_directories():
+    if not os.path.exists('./input'):
+        os.makedirs('./input')
+        print("创建了 './input' 文件夹。请将您的 GPX 文件导入到这个文件夹。")
+    if not os.path.exists('./output'):
+        os.makedirs('./output')
+        print("创建了 './output' 文件夹。")
+
+def ensure_numeric_args(args):
+    if not str(args.epochs).isdigit() or not str(args.num_points).isdigit() or not str(args.num_files).isdigit():
+        print("参数必须为数字。现在进入交互式。")
+        return False
+    return True
+
+def main():
+    ensure_directories()
+    args = parse_args()
+
+    if not ensure_numeric_args(args):
+        args.no_interaction = False
+
+    # 默认使用命令行参数设置变量
+    epochs = args.epochs
+    num_points = args.num_points
+    num_files = args.num_files
+
+    # 如果启用了交互模式，则使用用户输入覆盖
+    if not args.no_interaction:
+        epochs_str = cross_platform_input_with_timeout(
+            "请输入训练模型的迭代次数（默认为 100，10s 后采用默认值）：",
+            10,
+            str(args.epochs)
+        )
+        epochs = int(epochs_str) if epochs_str.isdigit() else args.epochs
+        
+        num_points_str = cross_platform_input_with_timeout(
+            "请输入每个 GPX 路线要生成的点数（默认为 100，10s 后采用默认值）：",
+            10,
+            str(args.num_points)
+        )
+        num_points = int(num_points_str) if num_points_str.isdigit() else args.num_points
+
+    if not args.no_interaction:
+        num_files_str = cross_platform_input_with_timeout(
+            "请输入要生成的 GPX 数量（默认为 10，10s 后采用默认值）：",
+            10,
+            str(args.num_files)
+        )
+        num_files = int(num_files_str) if num_files_str.isdigit() else args.num_files
+
+    # 在input文件夹中查找所有.gpx文件
+    gpx_files = glob.glob('./input/*.gpx')
+
+    # 数据预处理
+    sequences, next_points, mean_coords, std_coords = preprocess_gpx(gpx_files)
+
+    if sequences is not None:
+        # 构建模型
+        input_shape = (sequences.shape[1], sequences.shape[2])  # 序列长度和特征数
+        model = build_model(input_shape)
+
+        # 训练模型
+        train_model(model, sequences, next_points, epochs=epochs)
+
+        # 生成并保存新的GPX路线
+        seed = sequences[:1]  # 使用现有序列作为种子
+        for i in range(num_files):  # 使用 num_files 控制生成的 GPX 文件数量
+            generated_route = generate_gpx(model, seed, num_points=num_points)
+            # 反标准化并将生成的路径保存到 GPX 文件
+            filename = f'./output/generated_route_{i+1}.gpx'
+            save_to_gpx_route(filename, generated_route, mean_coords, std_coords)
+            print(f'生成的GPX路线已保存至 {filename}')
+
+if __name__ == '__main__':
+    main()
diff --git a/model.py b/model.py
@@ -0,0 +1,33 @@
+# model.py
+import numpy as np
+from tensorflow.keras.models import Sequential
+from tensorflow.keras.layers import LSTM, Dense
+from tensorflow.keras.callbacks import ModelCheckpoint
+
+# 设计 LSTM 模型
+def build_model(input_shape, units=64, dropout=0.3):
+    model = Sequential()
+    model.add(LSTM(units, input_shape=input_shape, return_sequences=True))
+    model.add(LSTM(units, return_sequences=False))
+    model.add(Dense(units, activation='relu'))
+    model.add(Dense(2))  # 预测纬度和经度
+    model.compile(optimizer='adam', loss='mse')
+    return model
+
+# 训练模型并保存
+# epochs 代表训练模型时整个数据集将被遍历迭代的次数。每一次遍历完整个数据集并进行一次前向传播和后向传播的过程被称为一个epoch。
+def train_model(model, sequences, next_points, epochs=100, batch_size=64, model_path='model.h5'):
+    checkpoint = ModelCheckpoint(model_path, save_best_only=True, monitor='val_loss', mode='min')
+    model.fit(sequences, next_points, batch_size=batch_size, epochs=epochs, callbacks=[checkpoint], validation_split=0.2)
+    model.save(model_path)
+    return model
+
+# 使用模型生成轨迹，num_points 代表生成的轨迹的点的数量
+def generate_gpx(model, seed, num_points=100):
+    generated_points = []
+    current_seq = seed
+    for _ in range(num_points):
+        predicted_point = model.predict(current_seq)
+        generated_points.append(predicted_point[0])
+        current_seq = np.concatenate((current_seq[0][1:], predicted_point.reshape(1, -1)), axis=0).reshape(1, current_seq.shape[1], current_seq.shape[2])
+    return np.array(generated_points)
diff --git a/preprocess.py b/preprocess.py
@@ -0,0 +1,63 @@
+import gpxpy
+import numpy as np
+
+# Loads GPX data
+def load_gpx_data(gpx_files):
+    tracks = []
+    for file in gpx_files:
+        print(f"Loading GPX file: {file}")
+        try:
+            with open(file, 'r') as f:
+                gpx = gpxpy.parse(f)
+            # Check for tracks
+            for track in gpx.tracks:
+                for segment in track.segments:
+                    if segment.points:
+                        tracks.append(np.array([(point.latitude, point.longitude) for point in segment.points]))
+                    else:
+                        print("No points in segment.")
+            # Check for routes
+            for route in gpx.routes:
+                if route.points:
+                    tracks.append(np.array([(point.latitude, point.longitude) for point in route.points]))
+                else:
+                    print("No points in route.")
+        except Exception as e:
+            print(f"Error loading GPX file {file}: {e}")
+    return tracks
+
+# Normalizes the tracks
+def normalize_tracks(tracks):
+    all_points = np.concatenate(tracks, axis=0)
+    mean_lat, mean_lon = np.mean(all_points, axis=0)
+    std_lat, std_lon = np.std(all_points, axis=0)
+
+    normalized_tracks = []
+    for track in tracks:
+        normalized_track = (track - [mean_lat, mean_lon]) / [std_lat, std_lon]
+        normalized_tracks.append(normalized_track)
+    return normalized_tracks, (mean_lat, mean_lon), (std_lat, std_lon)
+
+# Prepares sequences
+def prepare_sequences(tracks, sequence_length):
+    sequences = []
+    next_points = []
+    for track in tracks:
+        for i in range(len(track) - sequence_length):
+            sequences.append(track[i:i + sequence_length])
+            next_points.append(track[i + sequence_length])
+    return np.array(sequences), np.array(next_points)
+
+# Main function to preprocess GPX files
+def preprocess_gpx(files, sequence_length=5):
+    tracks = load_gpx_data(files)
+    if not tracks:
+        print("No tracks to normalize. Exiting preprocessing.")
+        return None, None, None, None
+
+    print(f"Normalizing {len(tracks)} tracks.")
+    normalized_tracks, mean_coords, std_coords = normalize_tracks(tracks)
+
+    print(f"Preparing sequences from normalized tracks.")
+    sequences, next_points = prepare_sequences(normalized_tracks, sequence_length)
+    return sequences, next_points, mean_coords, std_coords
diff --git a/requirements.txt b/requirements.txt
@@ -0,0 +1,4 @@
+numpy
+gpxpy
+tensorflow
+scikit-learn
