feat: Implement DMuffler application script for Raspberry Pi

[barkleesanders]

This commit introduces the dmuffler_pi.py script and a corresponding requirements.txt file to run a dynamic engine sound generation application (DMuffler) on your Raspberry Pi 4 with a PiCAN HAT and an I2S Audio DAC.

Key features and implementation steps:

1. Script Generation (dmuffler_pi.py):

   • Comprehensive command-line argument parsing for runtime configuration (CAN settings, audio paths, device indices, RPM parameters, test mode, etc.).
   • Detailed setup instructions embedded as comments covering:
     • System package installation (apt).
     • Python virtual environment setup.
     • Python package installation via requirements.txt.
     • Hardware configuration for PiCAN HAT (MCP2515) in /boot/config.txt, driver loading, and ip link setup.
     • Hardware configuration for I2S Audio DACs (e.g., HiFiBerry) in /boot/config.txt and ALSA verification.
   • Script usage instructions and expected console output examples.

2. Core Functionality:

   • CAN Interface: Initializes the specified SocketCAN interface (python-can), including bringing the interface up via sudo ip link and applying CAN ID filters. Includes error handling and fallback to a test mode if CAN setup fails.
   • Audio Output: Initializes PyAudio, attempts to auto-detect a suitable I2S DAC, and opens an audio stream. Supports manual device index specification. Handles errors gracefully, with an option to run without audio.
   • Main Loop:
     • Operates in either live CAN mode or a test mode with simulated/random RPMs.
     • Extracts RPM from CAN messages based on user-defined ID, byte index, scaling, and offset.
     • Selects WAV audio files from a specified directory based on the current RPM and a configurable RPM-to-filename map.
     • Implements non-blocking audio playback by dispatching WAV file playback to separate daemon threads, allowing the main loop to remain responsive.
   • Error Handling: Includes try-except blocks for KeyboardInterrupt, CAN errors during receive, and general robustness.
   • Resource Cleanup: Ensures graceful shutdown of the CAN bus, PyAudio instance, and active audio threads in a finally block.

3. Dependency Management:

   • Generated requirements.txt with pinned/minimum versions for python-can, pyaudio, pygame, numpy, and scipy.

4. Refinements:

   • Added comments to clarify the scope of current CAN parameter extraction (RPM-focused) and potential extensions for other parameters (throttle, speed).
   • Noted audio synthesis as a potential advanced alternative to WAV file playback.
   • Clarified CAN ID filter assumptions (standard vs. extended).

The script is designed to be configurable and provides substantial inline documentation to guide you through setup and operation on a Raspberry Pi.

[gemini-code-assist]

Hello @barkleesanders, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

Summary of Changes

Hello team, gemini-code-assist here with a summary of this pull request. This PR introduces a new Python script, dmuffler_pi.py, specifically designed to run on a Raspberry Pi equipped with a PiCAN HAT and an I2S Audio DAC. The core purpose is to read engine RPM data from a vehicle's CAN bus and dynamically play corresponding engine sound WAV files through the DAC, simulating a dynamic exhaust system. The script includes comprehensive command-line arguments for configuration, detailed inline setup instructions for hardware and software, and implements threaded audio playback to maintain responsiveness.

Highlights

• New Raspberry Pi Application Script: Adds a new Python script (dmuffler_pi.py) tailored for the Raspberry Pi platform to implement the DMuffler dynamic sound generation application.
• CAN Bus Integration: Integrates with SocketCAN interfaces (like those provided by a PiCAN HAT) using the python-can library to read vehicle data, specifically focusing on engine RPM.
• Audio Output via I2S DAC: Utilizes the pyaudio library to output dynamic engine sounds through an I2S Audio DAC connected to the Raspberry Pi, with auto-detection and manual device selection options.
• RPM-Based Audio Playback: Selects and plays specific WAV audio files from a configured directory based on the current engine RPM, allowing for different sounds at different engine speeds.
• Threaded Audio Playback: Implements audio playback in separate daemon threads to prevent the main loop (responsible for reading CAN data) from blocking, ensuring smooth operation.
• Configurable via Command Line: Provides extensive command-line arguments to configure CAN interface, RPM data parsing (ID, byte index, scale, offset), audio path, audio device, RPM thresholds, bitrate, and enable debug or test modes.
• Detailed Inline Documentation: Includes comprehensive comments within the script covering system setup, virtual environment creation, Python package installation, and hardware configuration steps for both the CAN interface and the I2S DAC.
• Test Mode: Includes a --test-mode option to run the application without a live CAN bus, simulating RPM changes for easier testing of the audio logic.

Changelog

Click here to see the changelog

• dmuffler_pi.py
  • Added a new Python script for the DMuffler application on Raspberry Pi.
  • Implemented command-line argument parsing for various configuration options.
  • Included detailed inline instructions for system, Python, and hardware setup.
  • Added functions for loading audio files based on RPM filenames.
  • Implemented logic for initializing PyAudio and auto-detecting/selecting audio output devices.
  • Added a function to parse RPM data from CAN messages (with user-configurable logic and warnings).
  • Integrated subprocess calls to bring up the SocketCAN interface using ip link.
  • Implemented the main application loop to read CAN data (or simulate RPM in test mode).
  • Added logic to select the appropriate audio file based on current RPM.
  • Implemented threaded playback of audio files using threading and pyaudio.
  • Included error handling for CAN and audio initialization/operation.
  • Added a finally block for graceful cleanup of CAN bus, PyAudio instance, and audio threads on exit.
• requirements.txt
  • Replaced the previous extensive list of dependencies with a minimal set required for the DMuffler script.
  • Added python-can==4.1.0 for CAN bus communication.
  • Added pyaudio>=0.2.11 for audio output.
  • Added pygame>=2.1.0 (not directly used in the core logic but included as per description).
  • Added numpy>=1.20.0 and scipy>=1.7.0 (not directly used in the core logic but included as per description).

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point in your pull request via creating an issue comment (i.e. comment on the pull request page) using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in issue comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist is currently in preview and may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments to provide feedback.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

---

CAN bus whispers low,
RPM climbs, the sounds will grow,
Pi sings exhaust's might.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

This pull request introduces a Python script (dmuffler_pi.py) designed to run on a Raspberry Pi with a PiCAN HAT and an I2S DAC, dynamically adjusting audio playback based on CAN bus data (engine RPM). The script includes detailed setup instructions, argument parsing, CAN interface initialization, audio output, and a main loop for processing RPM data and playing corresponding audio files. Overall, the script provides a good foundation for the DMuffler application. However, there are several areas that could be improved for robustness, maintainability, and clarity.

Summary of Findings

• Robustness of Audio Stream Handling: The code lacks comprehensive error handling around audio stream operations, potentially leading to crashes or unexpected behavior. Adding try...except blocks around critical audio operations (e.g., temp_stream.wait_done()) and logging exceptions can improve the script's resilience.
• CAN Bus Initialization Security: The use of shell=True in subprocess.run during CAN bus initialization poses a security risk. While the risk is mitigated by using argparse-derived values, it's best to avoid shell=True and use a list of arguments instead.
• Thread Synchronization: The method for stopping the audio playback thread relies on setting a global variable to None, which is a cooperative but potentially unreliable approach. Using a threading.Event and waiting for the thread to join provides a more robust way to ensure the thread has stopped before cleanup.
• Fallback Logic in RPM Extraction: The fallback logic in get_rpm_from_can_message that attempts to use a single byte when the expected number of bytes is not available can lead to incorrect RPM values. It's better to return None or raise an exception if the expected number of bytes is not present.

Merge Readiness

The pull request introduces a functional DMuffler application script for Raspberry Pi. However, several improvements are needed to enhance the script's robustness, security, and maintainability. I recommend addressing the issues related to audio stream handling, CAN bus initialization security, thread synchronization, and RPM extraction fallback logic before merging. I am unable to approve this pull request, and users should have others review and approve this code before merging.

[gemini-code-assist]

This fallback logic introduces a potential issue: if required_bytes is 2 (as initially assumed), but only one byte is available, the code will proceed with a single byte. This could lead to incorrect RPM values and unexpected behavior. It would be better to either return None or raise a more specific exception if the expected number of bytes is not available. Also, the warning message should indicate the expected number of bytes.

            elif len(msg.data) >= byte_index + 1: # Fallback example if only 1 byte is available (less likely for standard RPM)
                if debug:
                    print(f"Debug: WARNING - Only {len(msg.data)} bytes available but {required_bytes} bytes needed starting at index {byte_index}.  CAN data format may be incorrect. Data: {msg.data.hex()}")
                return None # Or raise an exception
                # raw_rpm_value = msg.data[byte_index] # Treat the single byte as the raw value, user might need to adjust logic

[gemini-code-assist]

It's good to check if p_audio_instance is available, but it would be more robust to check if p_audio_instance is not None AND if it's actually an instance of pyaudio.PyAudio. This adds a layer of type safety. Consider adding a check like isinstance(p_audio_instance, pyaudio.PyAudio).

    if args.no_audio or not isinstance(p_audio_instance, pyaudio.PyAudio): # Check if PyAudio is available

[gemini-code-assist]

Consider adding a try...except block around temp_stream.wait_done() to catch potential exceptions during the waiting process. This can prevent the entire function from crashing if there's an issue with the audio stream.

        try:
            temp_stream.wait_done() # Wait for all frames to be played
        except Exception as e:
            print(f"[Blocking] Error waiting for stream to finish: {e}")

[gemini-code-assist]

Instead of just breaking the loop, consider setting thread_stream to None and adding a check at the beginning of the while loop to see if thread_stream is None. This can prevent the thread from continuing to process data after it has been interrupted. This is especially important if thread_stream.write(data) could raise an exception after the thread should have stopped. It would also be good to add a debug log message here.

                if debug_flag:
                    print(f"[AudioThread:{current_thread_obj.name}] Playback of {wav_path} interrupted by new audio thread.")
                thread_stream = None # Prevent further writes
                break

[gemini-code-assist]

It's good to catch exceptions, but consider logging the exception message itself using str(e) or repr(e). This can provide more context for debugging. Also, consider using logger.exception to automatically include the traceback in the log message, especially in debug mode.

        # Catching PyAudioError specifically can be useful too: `except pyaudio.PyAudioError as e:`
        print(f"[AudioThread:{current_thread_obj.name}] ERROR playing {wav_path}: {e}")
        if debug_flag: 
            import traceback
            traceback.print_exc() # Consider using logger.exception(e) instead

[gemini-code-assist]

Checking thread_stream.is_active() before stopping is a good practice. However, it might be safer to also check if thread_stream is not None before calling is_active(). This prevents potential errors if thread_stream was not properly initialized or was set to None due to an earlier error.

                if thread_stream and thread_stream.is_active(): # Check if stream is active before stopping
                    thread_stream.stop_stream()

[gemini-code-assist]

Using shell=True can introduce security vulnerabilities if the can_interface_name or args.can_bitrate variables are derived from untrusted sources. Since these values are derived from argparse, the risk is lower, but it's still a good practice to avoid shell=True when possible. Consider using subprocess.run with a list of arguments instead.

            # Note: Using 'sudo' directly in script is generally discouraged for security reasons.
            # Consider if the user should run the script with sudo, or configure /etc/sudoers.d/
            # or run these ip link commands manually before starting.
            # For simplicity in this example, we'll proceed with sudo in shell command.
            subprocess.run(['sudo', 'ip', 'link', 'set', can_interface_name, 'down'], check=False)
            subprocess.run(['sudo', 'ip', 'link', 'set', can_interface_name, 'up', 'type', 'can', 'bitrate', str(args.can_bitrate)], check=True)

[gemini-code-assist]

It's good to clean up audio resources before exiting. However, the code directly accesses local variables audio_stream and p_audio_instance using locals(). This can be unreliable, especially if the variables are not guaranteed to be defined in the current scope. Since these are global variables, it's safer to directly reference them.

        # Cleanup audio before exiting (use new global names)
        if audio_stream and audio_stream.is_active(): 
            audio_stream.stop_stream()
            audio_stream.close()
        if p_audio_instance: 
            p_audio_instance.terminate()

[gemini-code-assist]

Setting audio_playback_thread = None signals the thread to stop, but there's no guarantee that the thread will stop immediately. It's possible that the thread could still be running and writing to the audio stream after this line. Consider using a threading.Event to signal the thread to stop and waiting for the thread to join with a timeout. This provides a more robust way to ensure that the thread has stopped before proceeding with cleanup.

            active_thread_to_join = audio_playback_thread
            # Signal for the thread's loop condition. The thread checks:
            # `if audio_playback_thread != current_thread_obj:`
            # So, setting it to None makes that condition true for the running thread if it was `current_thread_obj`.
            # audio_playback_thread = None
            # Instead, signal the thread to stop using an event
            audio_playback_thread_event.set() # Assuming you create an event object