Sensor2pg/utils/generate_tmp.py

109 lines
4.5 KiB
Python

import csv
import random
from datetime import datetime, timedelta
random.seed(51)
start_date = datetime(2025, 1, 6)
end_date = datetime(2025, 1, 10) # exclusive end date
camera_ids = list(range(103, 109)) + list(range(113, 119))
morning_start = timedelta(hours=9, minutes=30)
morning_end = timedelta(hours=12)
afternoon_start = timedelta(hours=14, minutes=30)
afternoon_end = timedelta(hours=19, minutes=30)
original_data = []
original_timestamps = set()
with open('tem.txt', mode='r', newline='') as file:
reader = csv.reader(file)
for row in reader:
stripped_row = [field.strip() for field in row if field.strip()]
if any(stripped_row) and not all(field == '0' for field in stripped_row):
try:
timestamp = datetime.strptime(stripped_row[0], '%Y年%m月%d%H:%M:%S')
original_data.append([timestamp] + stripped_row[1:])
original_timestamps.add(timestamp)
except ValueError:
continue
all_data = []
all_data.extend(original_data)
camera_states = {camera_id: {'positions': None, 'values': None, 'last_update': None} for camera_id in camera_ids}
current_date = start_date
while current_date < end_date:
if current_date.date() == datetime(2025, 1, 6).date():
num_records_per_hour = 1
elif current_date.date() == datetime(2025, 1, 7).date():
num_records_per_hour = random.randint(0, 2)
else:
num_records_per_hour = random.randint(0, 4)
for time_range in [(morning_start, morning_end), (afternoon_start, afternoon_end)]:
current_time = current_date.replace(hour=time_range[0].seconds // 3600,
minute=(time_range[0].seconds // 60) % 60,
second=0)
end_time = current_date.replace(hour=time_range[1].seconds // 3600,
minute=(time_range[1].seconds // 60) % 60,
second=0)
while current_time <= end_time:
for camera_id in camera_ids:
for _ in range(num_records_per_hour):
while True:
random_minutes = random.randint(0, 59)
random_seconds = random.randint(0, 59)
timestamp = current_time + timedelta(minutes=random_minutes, seconds=random_seconds)
if timestamp not in original_timestamps:
break
state = camera_states[camera_id]
positions = state['positions']
values = state['values']
last_update = state['last_update']
new_values = [None, None, None]
if last_update is None or timestamp - last_update > timedelta(hours=2):
positions = [random.choice([True, False]) for _ in range(3)]
values = [random.randint(200, 460) if pos else None for pos in positions]
state['positions'] = positions
state['values'] = values
state['last_update'] = timestamp
new_values = values
else:
for i, value in enumerate(values):
if value is not None:
change = random.randint(-20, 20)
new_value = max(min(value + change, 460), 200)
new_values[i] = new_value
state['values'] = new_values
temp_top = '' if not positions[0] or new_values[0] is None else str(new_values[0])
temp_mid = '' if not positions[1] or new_values[1] is None else str(new_values[1])
temp_bot = '' if not positions[2] or new_values[2] is None else str(new_values[2])
all_data.append([timestamp, camera_id, temp_top, temp_mid, temp_bot])
original_timestamps.add(timestamp)
current_time += timedelta(hours=1)
current_date += timedelta(days=1)
all_data.sort(key=lambda x: x[0])
with open('tmps.csv', mode='w', newline='') as file:
writer = csv.writer(file)
writer.writerow(['时间', '摄像头号', '堆顶', '堆中', '堆底'])
for row in all_data:
timestamp_str = row[0].strftime('%Y年%m月%d%H:%M:%S')
writer.writerow([timestamp_str] + row[1:])
print("CSV 文件已成功创建.")