PyAqueduct Application Programming Interface (API) Tutorial¶
In this tutorial, the API of PyAqueduct is introduced by working on a sample experiment. The sample experiment generates some results in the form of different files such as CSV, JSON, HDF5, and image files. Each execution of the experiment generates new set of files and therefore, is treated as a new experiment run.
Experiment: analysis a projectile motion¶
# Install experiment dependecies.
import sys
!{sys.executable} -m pip install numpy pandas matplotlib h5py
import os
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import json
import h5py
def simulate_projectile_motion(v0: float, angle: float, g: float = 9.81) -> pd.DataFrame:
"""
Simulates the projectile motion.
Args:
v0: Initial velocity in m/s.
angle: Launch angle in degrees.
g: Acceleration due to gravity in m/s^2 (default is 9.81).
Returns:
DataFrame containing time, x and y positions.
"""
# Time of flight calculation
t_max = 2 * v0 * np.sin(np.radians(angle)) / g
time_steps = np.linspace(0, t_max, num=50)
# Position calculations
x = v0 * np.cos(np.radians(angle)) * time_steps
y = v0 * np.sin(np.radians(angle)) * time_steps - 0.5 * g * time_steps**2
return pd.DataFrame({"Time (s)": time_steps, "X Position (m)": x, "Y Position (m)": y})
def save_data_formats(data: pd.DataFrame, base_path: str) -> None:
"""
Saves data in different formats: CSV, JSON, and HDF5.
Args:
data: Data to save.
base_path: Base file path without extension.
"""
# CSV
csv_path = f"{base_path}.csv"
data.to_csv(csv_path, index=False)
# JSON
json_path = f"{base_path}.json"
with open(json_path, "w") as json_file:
json.dump(data.to_dict(orient="records"), json_file)
# HDF5
hdf5_path = f"{base_path}.hdf5"
with h5py.File(hdf5_path, "w") as hdf_file:
for column in data.columns:
hdf_file.create_dataset(column, data=data[column].values)
def plot_trajectory(data: pd.DataFrame, image_path: str) -> None:
"""
Plots the trajectory of the projectile motion.
Args:
data: Data containing the trajectory.
image_path: Path to save the plot image.
"""
plt.figure(figsize=(8, 6))
plt.plot(data["X Position (m)"], data["Y Position (m)"])
plt.title("Projectile Motion Trajectory")
plt.xlabel("X Position (m)")
plt.ylabel("Y Position (m)")
plt.grid(True)
plt.savefig(image_path)
# Parameters for the simulation
initial_velocity = 20 # m/s
launch_angle = 45 # degrees
# Simulate the projectile motion
projectile_data = simulate_projectile_motion(initial_velocity, launch_angle)
# Save the data in different formats
base_file_path = os.path.join(os.getcwd(), "projectile_motion")
save_data_formats(projectile_data, base_file_path)
# Plot and save the trajectory
plot_image_path = os.path.join(os.getcwd(), "projectile_motion_plot.png")
plot_trajectory(projectile_data, plot_image_path)
print("Simulation and data processing completed.")
Create experiment on Aqueudct¶
from pyaqueduct import API
api = API("[AQUEDUCT_SERVER_URL_PLACE_HOLDER]", timeout=1)
experiment = api.create_experiment(
title="Motion Simulation Experiment", description="Tutorial experiment: motion simulation."
)
print(f"Experiment created with unique id: {experiment.eid}")
Add tags to experiment¶
Remove tags from experiment¶
Update experiment and file download/upload¶
experiment.title = "Motion Simulation"
experiment.description = "Motion Simulation"
print(f"Experiment title: {experiment.title}")
print(f"Experiment title: {experiment.description}")
print(f"Experiment creation date: {experiment.created_at}")
print(f"Experiment last update date: {experiment.updated_at}")
print(f"Experiment tags: {experiment.tags}")
Upload files¶
experiment.upload_file(file="projectile_motion.csv")
experiment.upload_file(file="projectile_motion.json")
experiment.upload_file(file="projectile_motion.hdf5")
experiment.upload_file(file="projectile_motion_plot.png")
Download files¶
download_dir = os.path.join(os.getcwd(), "downloads")
if not os.path.exists(download_dir):
os.makedirs(download_dir)
experiment.download_file(file_name="projectile_motion.csv", destination_dir=download_dir)
experiment.download_file(file_name="projectile_motion.json", destination_dir=download_dir)
experiment.download_file(file_name="projectile_motion.hdf5", destination_dir=download_dir)
experiment.download_file(file_name="projectile_motion_plot.png", destination_dir=download_dir)
Get specific experiment¶
experiment = api.get_experiment_by_eid(experiment.eid)
experiment = api.get_experiment_by_uuid(experiment.uuid)
print(f"Experiment title: {experiment.title}")
print(f"Experiment title: {experiment.description}")
print(f"Experiment creation date: {experiment.created_at}")
print(f"Experiment last update date: {experiment.updated_at}")
print(f"Experiment tags: {experiment.tags}")
Search through Experiments¶
from datetime import datetime, timedelta
# Search through the experiments by their title or experiment ID. It will return all experiments where search string is a substring of the title or the experiment ID
experiments_list = api.find_experiments(search="[search criteria, experiment title or EID]")
# Find experiments that the ALL of provided tags in the argument are assigned them. It is an AND operation between the tags when searching through experiments.
experiments_list = api.find_experiments(tags=["tag1", "tag2"])
# Find experiments with creation datetime after the specified datetime.
experiments_list = api.find_experiments(start_datetime=datetime.now() - timedelta(days=1))
# Find experiments with creation datetime before the specified datetime.
experiments_list = api.find_experiments(end_datetime=datetime.now() + timedelta(days=1))
# Find experiments with creation datetime between the specified datetimes.
experiments_list = api.find_experiments(
start_datetime=datetime.now() - timedelta(days=1),
end_datetime=datetime.now() + timedelta(days=1),
)
# You can also limit the results via pagination (default limit is 10 items per page) ordered by creation datetime.
# In this example we only retrieve maximum 20 experimets starting from the 10th experiment (inclusive) with the specified tags.
experiments_list = api.find_experiments(limit=20, offset=10, tags=["tag1", "tag2"])