Stream real-time GPIO voltage data based on mathematical equation Expects JSON: { 'equation': str, # Math expression (e.g., "1.65 + 1.65 * sin(x)") 'frequency': float, # Animation update frequency in Hz (1-10) 'duration': float, # Duration in seco
()
| 526 | |
| 527 | @app.route('/api/graph/stream', methods=['POST']) |
| 528 | def graph_stream(): |
| 529 | """Stream real-time GPIO voltage data based on mathematical equation |
| 530 | |
| 531 | Expects JSON: |
| 532 | { |
| 533 | 'equation': str, # Math expression (e.g., "1.65 + 1.65 * sin(x)") |
| 534 | 'frequency': float, # Animation update frequency in Hz (1-10) |
| 535 | 'duration': float, # Duration in seconds |
| 536 | 'pin': int, # GPIO pin number |
| 537 | 'sample_rate': int, # Samples per second |
| 538 | 'x_max': float # X-axis maximum (time span) |
| 539 | } |
| 540 | """ |
| 541 | import math |
| 542 | |
| 543 | def generate(): |
| 544 | pwm_device = None |
| 545 | start_time = time.time() |
| 546 | |
| 547 | try: |
| 548 | data = request.get_json() |
| 549 | equation_str = data.get('equation', '1.65 + 1.65 * sin(x)') |
| 550 | pin = int(data.get('pin', 18)) |
| 551 | animation_freq = float(data.get('frequency', 1.0)) # Animation speed (1-10 Hz) |
| 552 | sample_rate = int(data.get('sample_rate', 60)) |
| 553 | x_max = float(data.get('x_max', 10)) |
| 554 | |
| 555 | # PWM frequency for GPIO hardware (much higher than animation frequency) |
| 556 | PWM_FREQUENCY = 1000 # 1kHz is good for LEDs/bulbs |
| 557 | VOLTAGE_LIMIT = 3.3 |
| 558 | |
| 559 | # Calculate timing based on animation frequency |
| 560 | update_interval = 1.0 / max(animation_freq, 0.1) # Time between updates in seconds |
| 561 | |
| 562 | # Validate pin |
| 563 | if pin not in VALID_PINS: |
| 564 | yield f"data: {json.dumps({'error': f'Invalid pin {pin}'})}\n\n" |
| 565 | return |
| 566 | |
| 567 | # Setup PWM on the pin with proper frequency for hardware |
| 568 | try: |
| 569 | cleanup_pin(pin) |
| 570 | |
| 571 | if GPIO_BACKEND == 'gpiozero': |
| 572 | pwm_device = PWMOutputDevice(pin, frequency=PWM_FREQUENCY) |
| 573 | log_event(f"PWM device created on pin {pin} with {PWM_FREQUENCY}Hz frequency (gpiozero)") |
| 574 | elif GPIO_BACKEND == 'RPi.GPIO': |
| 575 | GPIO.setup(pin, GPIO.OUT) |
| 576 | pwm_device = GPIO.PWM(pin, PWM_FREQUENCY) |
| 577 | pwm_device.start(0) # Start with 0% duty cycle |
| 578 | log_event(f"PWM device created on pin {pin} with {PWM_FREQUENCY}Hz frequency (RPi.GPIO)") |
| 579 | else: # mock |
| 580 | pwm_device = GPIO.PWM(pin, PWM_FREQUENCY) |
| 581 | pwm_device.start(0) |
| 582 | log_event(f"PWM device created on pin {pin} with {PWM_FREQUENCY}Hz frequency (mock)") |
| 583 | |
| 584 | pwm_devices[pin] = { |
| 585 | 'device': pwm_device, |
nothing calls this directly
no test coverage detected