feat: Add oscillator sensor visualization and data export scripts
- demo_oscillator_simple.py: Visualizes oscillator waveforms in terminal - oscillator_data_export.py: Exports oscillator data as JSON - Supports all waveforms: sine, square, sawtooth, triangle, noise - Real-time visualization with phase tracking - Configurable frequency, sample rate, and duration
This commit is contained in:
134
scripts/demo_oscillator_simple.py
Normal file
134
scripts/demo_oscillator_simple.py
Normal file
@@ -0,0 +1,134 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Simple Oscillator Sensor Demo
|
||||
|
||||
This script demonstrates the oscillator sensor by:
|
||||
1. Creating an oscillator sensor with various waveforms
|
||||
2. Printing the waveform data in real-time
|
||||
|
||||
Usage:
|
||||
uv run python scripts/demo_oscillator_simple.py --waveform sine --frequency 1.0
|
||||
uv run python scripts/demo_oscillator_simple.py --waveform square --frequency 2.0
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import math
|
||||
import time
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add mainline to path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from engine.sensors.oscillator import OscillatorSensor, register_oscillator_sensor
|
||||
|
||||
|
||||
def render_waveform(width: int, height: int, osc: OscillatorSensor, frame: int) -> str:
|
||||
"""Render a waveform visualization."""
|
||||
# Get current reading
|
||||
current_reading = osc.read()
|
||||
current_value = current_reading.value if current_reading else 0.0
|
||||
|
||||
# Generate waveform data - sample the waveform function directly
|
||||
# This shows what the waveform looks like, not the live reading
|
||||
samples = []
|
||||
waveform_fn = osc.WAVEFORMS[osc._waveform]
|
||||
|
||||
for i in range(width):
|
||||
# Sample across one complete cycle (0 to 1)
|
||||
phase = i / width
|
||||
value = waveform_fn(phase)
|
||||
samples.append(value)
|
||||
|
||||
# Build visualization
|
||||
lines = []
|
||||
|
||||
# Header with sensor info
|
||||
header = (
|
||||
f"Oscillator: {osc.name} | Waveform: {osc.waveform} | Freq: {osc.frequency}Hz"
|
||||
)
|
||||
lines.append(header)
|
||||
lines.append("─" * width)
|
||||
|
||||
# Waveform plot (scaled to fit height)
|
||||
num_rows = height - 3 # Header, separator, footer
|
||||
for row in range(num_rows):
|
||||
# Calculate the sample value that corresponds to this row
|
||||
# 0.0 is bottom, 1.0 is top
|
||||
row_value = 1.0 - (row / (num_rows - 1)) if num_rows > 1 else 0.5
|
||||
|
||||
line_chars = []
|
||||
for x, sample in enumerate(samples):
|
||||
# Determine if this sample should be drawn in this row
|
||||
# Map sample (0.0-1.0) to row (0 to num_rows-1)
|
||||
# 0.0 -> row 0 (bottom), 1.0 -> row num_rows-1 (top)
|
||||
sample_row = int(sample * (num_rows - 1))
|
||||
if sample_row == row:
|
||||
# Use different characters for waveform vs current position marker
|
||||
# Check if this is the current reading position
|
||||
if abs(x / width - (osc._phase % 1.0)) < 0.02:
|
||||
line_chars.append("◎") # Current position marker
|
||||
else:
|
||||
line_chars.append("█")
|
||||
else:
|
||||
line_chars.append(" ")
|
||||
lines.append("".join(line_chars))
|
||||
|
||||
# Footer with current value and phase info
|
||||
footer = f"Value: {current_value:.3f} | Frame: {frame} | Phase: {osc._phase:.2f}"
|
||||
lines.append(footer)
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def demo_oscillator(waveform: str = "sine", frequency: float = 1.0, frames: int = 0):
|
||||
"""Run oscillator demo."""
|
||||
print(f"Starting oscillator demo: {waveform} wave at {frequency}Hz")
|
||||
if frames > 0:
|
||||
print(f"Running for {frames} frames")
|
||||
else:
|
||||
print("Press Ctrl+C to stop")
|
||||
print()
|
||||
|
||||
# Create oscillator sensor
|
||||
register_oscillator_sensor(name="demo_osc", waveform=waveform, frequency=frequency)
|
||||
osc = OscillatorSensor(name="demo_osc", waveform=waveform, frequency=frequency)
|
||||
osc.start()
|
||||
|
||||
# Run demo loop
|
||||
try:
|
||||
frame = 0
|
||||
while frames == 0 or frame < frames:
|
||||
# Render waveform
|
||||
visualization = render_waveform(80, 20, osc, frame)
|
||||
|
||||
# Print with ANSI escape codes to clear screen and move cursor
|
||||
print("\033[H\033[J" + visualization)
|
||||
|
||||
time.sleep(0.05) # 20 FPS
|
||||
frame += 1
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n\nDemo stopped by user")
|
||||
|
||||
finally:
|
||||
osc.stop()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description="Oscillator sensor demo")
|
||||
parser.add_argument(
|
||||
"--frames",
|
||||
type=int,
|
||||
default=0,
|
||||
help="Number of frames to render (0 = infinite until Ctrl+C)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--frequency", type=float, default=1.0, help="Oscillator frequency in Hz"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--frames", type=int, default=100, help="Number of frames to render"
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
demo_oscillator(args.waveform, args.frequency, args.frames)
|
||||
111
scripts/oscillator_data_export.py
Normal file
111
scripts/oscillator_data_export.py
Normal file
@@ -0,0 +1,111 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Oscillator Data Export
|
||||
|
||||
Exports oscillator sensor data in JSON format for external use.
|
||||
|
||||
Usage:
|
||||
uv run python scripts/oscillator_data_export.py --waveform sine --frequency 1.0 --duration 5.0
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import time
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
# Add mainline to path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from engine.sensors.oscillator import OscillatorSensor, register_oscillator_sensor
|
||||
|
||||
|
||||
def export_oscillator_data(
|
||||
waveform: str = "sine",
|
||||
frequency: float = 1.0,
|
||||
duration: float = 5.0,
|
||||
sample_rate: float = 60.0,
|
||||
output_file: str | None = None,
|
||||
):
|
||||
"""Export oscillator data to JSON."""
|
||||
print(f"Exporting oscillator data: {waveform} wave at {frequency}Hz")
|
||||
print(f"Duration: {duration}s, Sample rate: {sample_rate}Hz")
|
||||
|
||||
# Create oscillator sensor
|
||||
register_oscillator_sensor(
|
||||
name="export_osc", waveform=waveform, frequency=frequency
|
||||
)
|
||||
osc = OscillatorSensor(name="export_osc", waveform=waveform, frequency=frequency)
|
||||
osc.start()
|
||||
|
||||
# Collect data
|
||||
data = {
|
||||
"waveform": waveform,
|
||||
"frequency": frequency,
|
||||
"duration": duration,
|
||||
"sample_rate": sample_rate,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"samples": [],
|
||||
}
|
||||
|
||||
sample_interval = 1.0 / sample_rate
|
||||
num_samples = int(duration * sample_rate)
|
||||
|
||||
print(f"Collecting {num_samples} samples...")
|
||||
|
||||
for i in range(num_samples):
|
||||
reading = osc.read()
|
||||
if reading:
|
||||
data["samples"].append(
|
||||
{
|
||||
"index": i,
|
||||
"timestamp": reading.timestamp,
|
||||
"value": reading.value,
|
||||
"phase": osc._phase,
|
||||
}
|
||||
)
|
||||
time.sleep(sample_interval)
|
||||
|
||||
osc.stop()
|
||||
|
||||
# Export to JSON
|
||||
if output_file:
|
||||
with open(output_file, "w") as f:
|
||||
json.dump(data, f, indent=2)
|
||||
print(f"Data exported to {output_file}")
|
||||
else:
|
||||
print(json.dumps(data, indent=2))
|
||||
|
||||
return data
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description="Export oscillator sensor data")
|
||||
parser.add_argument(
|
||||
"--waveform",
|
||||
choices=["sine", "square", "sawtooth", "triangle", "noise"],
|
||||
default="sine",
|
||||
help="Waveform type",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--frequency", type=float, default=1.0, help="Oscillator frequency in Hz"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--duration", type=float, default=5.0, help="Duration to record in seconds"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--sample-rate", type=float, default=60.0, help="Sample rate in Hz"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output", "-o", type=str, help="Output JSON file (default: print to stdout)"
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
export_oscillator_data(
|
||||
waveform=args.waveform,
|
||||
frequency=args.frequency,
|
||||
duration=args.duration,
|
||||
sample_rate=args.sample_rate,
|
||||
output_file=args.output,
|
||||
)
|
||||
Reference in New Issue
Block a user