| import time |
| import os |
| from urllib.parse import urlparse, parse_qs |
| from datetime import datetime |
| from selenium import webdriver |
| from selenium.webdriver.chrome.service import Service |
| from selenium.webdriver.chrome.options import Options |
| from selenium.webdriver.common.by import By |
| from selenium.webdriver.common.keys import Keys |
|
|
| |
| download_path = '/Users/noamcohen/Downloads/data collection/data' |
| ROADS = {'405': ['N', 'S'], 101: ['N', 'S'], 5: ['N', 'S'], 110: ['N', 'S'], 170: ['N', 'S'], 118: ['E', 'W'], |
| 134: ['E', 'W'], 605: ['N', 'S'], 210: ['E', 'W']} |
| |
| chrome_options = Options() |
| chrome_prefs = { |
| "download.default_directory": download_path, |
| "download.prompt_for_download": False, |
| "download.directory_upgrade": True, |
| } |
| chrome_options.add_experimental_option("prefs", chrome_prefs) |
|
|
| |
| driver = webdriver.Chrome(service=Service('/opt/homebrew/bin/chromedriver'), options=chrome_options) |
|
|
| |
|
|
| |
| url = "https://pems.dot.ca.gov/" |
| driver.get(url) |
|
|
| |
| time.sleep(20) |
|
|
| |
| username_field = driver.find_element(By.ID, "username") |
| password_field = driver.find_element(By.ID, "password") |
|
|
| |
| username_field.send_keys("amitomer1912@gmail.com") |
| password_field.send_keys("5^applel?X") |
|
|
| |
| login_button = driver.find_element(By.NAME, "login") |
| login_button.click() |
|
|
| |
| time.sleep(20) |
| i = 1 |
| s_time_id=1740787200 |
| while(i <= 1): |
| |
| if i <= 9: |
| day = f"0{i}" |
| else: |
| day = i |
| for road in ROADS.keys(): |
| for dir in ROADS.get(road): |
|
|
| report_url = f"https://pems.dot.ca.gov/?report_form=1&dnode=Freeway&content=spatial&tab=contours&export=&fwy={road}&dir={dir}&s_time_id={s_time_id}&s_time_id_f=03%2F{day}%2F2025&from_hh=0&to_hh=23&start_pm=.0&end_pm=1000.09&lanes=&station_type=ml&q=speed&colormap=30%2C31%2C32&sc=auto&ymin=&ymax=&view_d=2&chart.x=93&chart.y=20" |
| driver.get(report_url) |
| time.sleep(60) |
|
|
| |
| export_button = driver.find_element(By.NAME, "xls") |
|
|
| |
| export_button.click() |
|
|
| |
| time.sleep(100) |
|
|
| |
| parsed_url = urlparse(report_url) |
| query_params = parse_qs(parsed_url.query) |
| highway_number = query_params.get('fwy', ['unknown'])[0] |
| date_taken = query_params.get('s_time_id_f', ['unknown_date'])[0] |
|
|
| |
| new_file_name = rf"{highway_number}_{dir}_{date_taken.replace("/", "*")}.xlsx" |
| new_file_path = os.path.join(download_path, new_file_name) |
|
|
| |
| original_file_path = os.path.join(download_path, 'pems_output.xlsx') |
|
|
| |
| if os.path.exists(original_file_path): |
| os.rename(original_file_path, new_file_path) |
| print(f"File renamed to: {new_file_path}") |
| else: |
| print("Original file not found!") |
|
|
| print(f"Download {i} to March complete!") |
| i+=1 |
| s_time_id += 86400 |
| |
| driver.quit() |
|
|