Merge develop branch into main (#1)

* refactor: functional

* refactor: ♻️ move alsa_hw_mapping to config param

---------

Co-authored-by: Nick Pourazima <nick.pourazima@gmail.com>
This commit is contained in:
Craig Hesling 2023-02-07 20:59:02 -05:00 committed by GitHub
parent b6202bbaad
commit 0c43bba31c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 445 additions and 224 deletions

3
.gitignore vendored
View File

@ -1,2 +1 @@
*.wav *.code*
*.mp3

112
README.md Normal file
View File

@ -0,0 +1,112 @@
# Rotary Phone Audio Guestbook
This project transforms a rotary phone into a voice recorder for use at special events (i.e. wedding audio guestbook, etc.).
- [Rotary Phone Audio Guestbook](#rotary-phone-audio-guestbook)
- [Background](#background)
- [Post-Event](#post-event)
- [Future Potential](#future-potential)
- [Materials](#materials)
- [Setup](#setup)
- [Hardware](#hardware)
- [Wiring](#wiring)
- [Microphone Replacement (Optional)](#microphone-replacement-optional)
- [Software](#software)
- [Dev Environment](#dev-environment)
- [Dependencies](#dependencies)
- [AudioInterface Class](#audiointerface-class)
- [rotaryGuestBook.service](#rotaryguestbookservice)
- [Operation Mode 1: rotaryGuestBook](#operation-mode-1-rotaryguestbook)
- [Operation Mode 2: rotaryGuestBookwithRotaryDialer](#operation-mode-2-rotaryguestbookwithrotarydialer)
## Background
I was inspired by my own upcoming wedding to put together a DIY solution for an audio guestbook using a rotary phone. Most online rentals were charging $600 for an experience that didn't even offer the ability to add a custom voice mail and took about 4-6 weeks of turn around time to process the audio after the event. I tried to use as many parts that I had laying around to keep costs down. It worked out quite well and we were able to gather some very special voice messages.
Below you will find a parts list and detailed setup guide. Please feel free to reach out to me with any questions.
## Post-Event
Since this was a trial by fire type of scenario there ended up being a few gotchas at the real event which I've since accounted for. Namely setting a time limit on the recording length as we had some youngsters leaving 5+ minute messages repeatedly and this ended up draining the battery.
### Future Potential
A few weeks before the wedding I had the code registering dialed numbers from the rotary encoder with the goal of playing back special messages for certain guests who dialed a certain combination (i.e. dial an area code to hear a special message to my old roomates). The details of this operation mode are described in Mode 2 below. In order to activate this mode I had to wait for input when the phone was off the hook. This required an extra step of dialing zero before leaving a normal voice message. In the end we decided to keep it simple and I've thus migrated this code to the dev branch along with the code to run through post-porcessing the audio in a separate process.
If any one is interested in expanding this please feel free.
## Materials
| Part|Notes|Quantity|Cost|
| - | - | - | - |
| [rotary phone](https://www.ebay.com/b/Rotary-Dial-Telephone/38038/bn_55192308) | Estate/garage/yard sales are probably the best places to find once of these. Ideally one with a phone jack since we will be using these four wires extensively. | 1 | $0.00-$60.00 |
| [raspberry pi zero](https://www.raspberrypi.com/products/raspberry-pi-zero/) | I didn't realize how hard these are to find these days. You can use any rpi or arduino style single-board computer but be aware of size constraints (i.e. must fit inside the rotary phone enclosure) | 1 | $9.99 |
| [raspberry pi zero case](https://www.adafruit.com/product/3252) | Optional: added for protection. One of the cases on Amazon has a heat-sink cutout which might be nice for better heat dissapation since it will all be enclosed in the end. | 1 | $4.95 |
| [micro SD card](https://a.co/d/1gb2zhC) | Any high capacity/throughput micro SD card that is rpi compatible | 1 | $8.99 |
| [USB Audio Adapter](https://www.adafruit.com/product/1475) | Note: I removed the external plastic shell and directly soldered the wires instead of using the female 3.5mm receptacle. | 1 | $4.95 |
| [USB OTG Host Cable - MicroB OTG male to A female](https://www.adafruit.com/product/1099) | | 1 | $2.50 |
| --- | **--- If running off a battery ---** | --- | --- |
| [LiPo Battery](https://www.adafruit.com/product/2011)| Optional: maximize capacity based on what will fit within your rotary enclosure. |1| $12.50 |
| [LiPo Shim](https://www.adafruit.com/product/3196)| Optional: if you plan to run this off a LiPo I would recommend something like this to interface with the rpi zero. |1| $9.95 |
| [LiPo Charger](https://www.adafruit.com/product/1904) | Optional: for re-charging the LiPo. |1| $6.95 |
| --- | **--- If replacing the built-it microphone ---** | --- | --- |
| [LavMic](https://www.amazon.com/dp/B01N6P80OQ?ref=nb_sb_ss_w_as-reorder-t1_ypp_rep_k3_1_9&amp=&crid=15WZEWMZ17EM9&amp=&sprefix=saramonic) | Optional: if you'd like to replace the carbon microphone. This is an omnidirectional lavalier mic and outputs via a 3.5mm TRS | 1 | $24.95 |
## Setup
### Hardware
#### Wiring
#### Microphone Replacement (Optional)
I found the sound quality of the built-in [carbon microphone](https://en.wikipedia.org/wiki/Carbon_microphone) on the rotary phone to be quite lacking in terms of amplitude, dynamic range and overall vocal quality. I tried boosting the gain from the digital (ALSA driver) side but this introduced an incredible amount of noise as expected. I then approached this from the analog domain and tried alternative circuitry to boost the sound quality based off this [carbon-to-dynamic converter](https://www.circuits-diy.com/mic-converter-circuit/).
Might be worth a further investigation in the future since it retains the integrity of the original rotary phone.
My final attempt involved the introduction of some post-proceesing (see dev branch) to bandpass some of the freqs outside the speech domain and add some normalization. The processing was costly in terms of processing and power consumption/rendering time and I ultimately decided it was worth acquiring something that yielded a better capture right out the gate. Crap in, crap out - as they say in the sound recording industry.
To replace:
- Unscrew mouthpiece and remove the carbon mic
- Pop out the plastic terminal housing with the two metal leads
- Unscrew red and black wires from terminal
![image](images/phone2.jpg)
### Software
#### Dev Environment
- rpi image: [Rasbian](https://www.raspberrypi.com/documentation/computers/getting-started.html) w/ SSH enabled
- rpi on same network as development machine
- Desktop IDE: vscode w/ [SSH FS extension](https://marketplace.visualstudio.com/items?itemName=Kelvin.vscode-sshfs)
[Here's](https://jayproulx.medium.com/headless-raspberry-pi-zero-w-setup-with-ssh-and-wi-fi-8ddd8c4d2742) a great guide to get the rpi setup headless w/ SSH & WiFi dialed in.
#### Dependencies
- `pip3 install -r requirements.txt` or pip install each manually:
- [GPIOZero](https://gpiozero.readthedocs.io)
- [Pydub](http://pydub.com/)
- [PyAudio](https://people.csail.mit.edu/hubert/pyaudio/)
- [PyYAML](https://pyyaml.org/)
#### [AudioInterface Class](audioInterface.py)
#### [rotaryGuestBook.service](rotaryGuestBook.service)
This service starts the python script on boot. Place it in the `/etc/systemd/system` directory.
`systemctl enable audioGuestBook.service`
`systemctl start audioGuestBook.service`
#### Operation Mode 1: [rotaryGuestBook](rotaryGuestBook.py)
#### Operation Mode 2: [rotaryGuestBookwithRotaryDialer](rotaryGuestBookwithRotaryDialer.py)
***Note*:** Untested - decided not to go this route for my own wedding
- This mode is a special modification of the normal operation and requires a slightly different wiring connection since it accepts input from the rotary dialer.
- The idea was to playback special messages when particular users dial a certain number combination (i.e. 909 would play back a message for certain guests who lived with the groom in that area code).
- In this mode of operation the users will need to dial 0 on the rotary dialer in order to initiate the voicemail.

73
audioGuestBook.py Normal file
View File

@ -0,0 +1,73 @@
#! /usr/bin/env python3
import audioInterface
import os
import yaml
import sys
from datetime import datetime
from gpiozero import Button
from signal import pause
from pydub import AudioSegment
from pydub.playback import play
try:
with open("config.yaml") as f:
config = yaml.load(f, Loader=yaml.FullLoader)
except FileNotFoundError as e:
print(
f"Could not find the config.yaml file. FileNotFoundError: {e}. Check config location and retry."
)
sys.exit(1)
hook = Button(config["hook_gpio"])
def off_hook() -> None:
print("Phone off hook, ready to begin!")
audio_interface = audioInterface.AudioInterface(config, hook)
# playback voice message through speaker
print("Playing voicemail message...")
play(
AudioSegment.from_wav(
os.path.dirname(os.path.abspath(config["source_file"]))
+ "/sounds/voicemail.wav"
)
- config["playback_reduction"]
)
# start recording beep
print("Playing beep...")
play(
AudioSegment.from_wav(
os.path.dirname(os.path.abspath(config["source_file"])) + "/sounds/beep.wav"
)
- config["beep_reduction"]
)
# now, while phone is off the hook, record audio from the microphone
print("recording")
audio_interface.record()
audio_interface.stop()
output_file = (
os.path.dirname(os.path.abspath(config["source_file"]))
+ "/recordings/"
+ f"{datetime.now().isoformat()}"
)
audio_interface.close(output_file + ".wav")
print("Finished recording!")
def on_hook() -> None:
print("Phone on hook.\nSleeping...")
def main():
hook.when_pressed = off_hook
hook.when_released = on_hook
pause()
if __name__ == "__main__":
main()

11
audioGuestBook.service Normal file
View File

@ -0,0 +1,11 @@
[Unit]
Description=Rotary Phone Guest Book Project
After=multi-user.target
[Service]
WorkingDirectory=/home/<username>/Desktop/rotaryGuestBook
User=<username>
Type=simple
Restart=always
ExecStart=/usr/bin/env python3 /home/<username>/Desktop/rotaryGuestBook/rotaryGuestBook.py
[Install]
WantedBy=multi-user.target

98
audioInterface.py Normal file
View File

@ -0,0 +1,98 @@
#! /usr/bin/env python3
import pyaudio
import time
import wave
from pydub import AudioSegment
from pydub.effects import normalize, compress_dynamic_range
from pydub.scipy_effects import band_pass_filter
class AudioInterface:
def __init__(self, config, hook) -> None:
self.audio = pyaudio.PyAudio()
self.chunk = config["buffer_size"]
self.chans = config["channels"]
self.format = pyaudio.paInt16 # 16-bit resolution
self.frames = [] # raw data frames recorded from mic
self.hook = hook
self.samp_rate = config["sample_rate"]
self.recording_limit = config["recording_limit"]
self.dev_index = config["alsa_hw_mapping"] # device index found by p.get_device_info_by_index(ii)
def record(self):
# create pyaudio stream
self.stream = self.audio.open(
format=self.format,
rate=self.samp_rate,
channels=self.chans,
input_device_index=self.dev_index,
input=True,
frames_per_buffer=self.chunk,
)
# loop through stream and append audio chunks to frame array
try:
start = time.time()
while self.hook.is_pressed:
if time.time() - start < self.recording_limit:
data = self.stream.read(self.chunk, exception_on_overflow=True)
self.frames.append(data)
else:
break
except KeyboardInterrupt:
print("Done recording")
except Exception as e:
print(str(e))
def play(self, file):
self.wf = wave.open(file, "rb")
self.stream = self.audio.open(
format=self.audio.get_format_from_width(self.wf.getsampwidth()),
channels=self.wf.getnchannels(),
rate=self.wf.getframerate(),
output=True,
)
""" Play entire file """
data = self.wf.readframes(self.chunk)
while len(data):
self.stream.write(data)
data = self.wf.readframes(self.chunk)
def stop(self):
# stop the stream
self.stream.stop_stream()
# close it
self.stream.close()
# terminate the pyaudio instantiation
self.audio.terminate()
def close(self, output_file):
# save the audio frames as .wav file
with wave.open(output_file, "wb") as wavefile:
wavefile.setnchannels(self.chans)
wavefile.setsampwidth(self.audio.get_sample_size(self.format))
wavefile.setframerate(self.samp_rate)
wavefile.writeframes(b"".join(self.frames))
def postProcess(self, outputFile):
"""
TODO: Evaluate whether this is worthwhile...
"""
source = AudioSegment.from_wav(outputFile + ".wav")
print("Filtering...")
filtered = band_pass_filter(source, 300, 10000)
print("Normalizing...")
normalized = normalize(filtered)
print("Compress Dynamic Range")
compressed = compress_dynamic_range(normalized)
print("Exporting normalized")
normalized.export(outputFile + "normalized.wav", format="wav")
print("Exporting compressed")
compressed.export(outputFile + "compressed.mp3", format="mp3")
print("Finished...")

View File

@ -1,9 +1,12 @@
hook_gpio: 22 alsa_hw_mapping: 1
rotary_gpio: 3
rotary_hold_time: 0.25
rotary_hold_repeat: true
playback_reduction: 16
beep_reduction: 24 beep_reduction: 24
sample_rate: 44100
buffer_size: 4096 buffer_size: 4096
channels: 2 channels: 2
hook_gpio: 22
playback_reduction: 16
recording_limit: 60
rotary_gpio: 23
rotary_hold_repeat: true
rotary_hold_time: 0.25
sample_rate: 44100
source_file: "audioGuestBook.py"

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.9 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.7 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.7 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.0 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.4 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.7 MiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 667 KiB

BIN
images/phone2.jpg Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 11 KiB

4
requirements.txt Normal file
View File

@ -0,0 +1,4 @@
gpiozero==1.6.2
pyaudio==0.2.12
pydub==0.25.1
PyYAML==6.0

View File

@ -1,215 +0,0 @@
#! /usr/bin/env python3
import os
import yaml
import pyaudio
import wave
from datetime import datetime
from gpiozero import Button
from multiprocessing import Process
from signal import pause
from pydub.scipy_effects import band_pass_filter
from pydub.effects import normalize
from pydub import AudioSegment
from pydub.playback import play
with open("config.yaml") as f:
config = yaml.load(f, Loader=yaml.FullLoader)
hook = Button(config["hook_gpio"])
# rotaryDial = Button(pin=config['rotary_gpio'], hold_time=config['rotary_hold_time'], hold_repeat=config['rotary_hold_repeat'])
"""
TODO: These globals are a temp solution for the rotary dialer, would love to not
depend on globals for this logic.
"""
# count = 0
# dialed = []
# reset_flag = False
class AudioInterface:
def __init__(self) -> None:
self.audio = pyaudio.PyAudio()
self.samp_rate = config["sample_rate"]
self.chunk = config["buffer_size"]
self.chans = config["channels"]
self.format = pyaudio.paInt16 # 16-bit resolution
self.frames = [] # raw data frames recorded from mic
def record(self):
# create pyaudio stream
dev_index = 1 # device index found by p.get_device_info_by_index(ii)
self.stream = self.audio.open(
format=self.format,
rate=self.samp_rate,
channels=self.chans,
input_device_index=dev_index,
input=True,
frames_per_buffer=self.chunk,
)
# loop through stream and append audio chunks to frame array
try:
# TODO: either pass hook as object into class, or figure out another cleaner solution
while hook.is_pressed:
data = self.stream.read(self.chunk, exception_on_overflow=True)
self.frames.append(data)
except KeyboardInterrupt:
print("Done recording")
except Exception as e:
print(str(e))
def play(self, file):
self.wf = wave.open(file, "rb")
self.stream = self.audio.open(
format=self.audio.get_format_from_width(self.wf.getsampwidth()),
channels=self.wf.getnchannels(),
rate=self.wf.getframerate(),
output=True,
)
""" Play entire file """
data = self.wf.readframes(self.chunk)
while len(data):
self.stream.write(data)
data = self.wf.readframes(self.chunk)
def stop(self):
# stop the stream
self.stream.stop_stream()
# close it
self.stream.close()
# terminate the pyaudio instantiation
self.audio.terminate()
def close(self, outputFile):
# save the audio frames as .wav file
with wave.open(outputFile, "wb") as wavefile:
wavefile.setnchannels(self.chans)
wavefile.setsampwidth(self.audio.get_sample_size(self.format))
wavefile.setframerate(self.samp_rate)
wavefile.writeframes(b"".join(self.frames))
def postProcess(self, outputFile):
"""
TODO: Evaluate whether this is worthwhile...
"""
source = AudioSegment.from_wav(outputFile + ".wav")
print("Filtering...")
filtered = band_pass_filter(source, 300, 10000)
print("Normalizing...")
normalized = normalize(filtered)
# print("Compress Dynamic Range")
# compressed = compress_dynamic_range(normalized)
print("Exporting normalized")
normalized.export(outputFile + "normalized.wav", format="wav")
# print("Exporting compressed")
# compressed.export(outputFile + "compressed.mp3", format="mp3")
print("Finished...")
def offHook():
print("Phone off hook, ready to begin!")
# if dialed and dialed[0] == 0:
audioInterface = AudioInterface()
# playback voice message through speaker
print("Playing voicemail message...")
play(
AudioSegment.from_wav(
os.path.dirname(os.path.abspath("rotaryGuestBook.py"))
+ "/sounds/voicemail.wav"
)
- config["playback_reduction"]
)
# start recording beep
print("Playing beep...")
play(
AudioSegment.from_wav(
os.path.dirname(os.path.abspath("rotaryGuestBook.py")) + "/sounds/beep.wav"
)
- config["beep_reduction"]
)
# now, while phone is not off the hook, record audio from the microphone
print("recording")
audioInterface.record()
audioInterface.stop()
outputFile = (
os.path.dirname(os.path.abspath("rotaryGuestBook.py"))
+ "/recordings/"
+ f"{datetime.now().isoformat()}"
)
audioInterface.close(outputFile + ".wav")
print("finished recording")
"""
post processing
"""
# print("spawn postProcessing thread")
# Process(target=audioInterface.postProcess, args=(outputFil e,)).start()
"""
rotary dialer special messages
"""
# if dialed[0:3] == [9,2,7]:
# # play special vm
# play(AudioSegment.from_wav(os.path.dirname(os.path.abspath("rotaryGuestBook.py")) + "/sounds/927.wav") - config['playback_reduction'])
# elif dialed[0:4] == [5,4,5,3]:
# # play special vm
# play(AudioSegment.from_wav(os.path.dirname(os.path.abspath("rotaryGuestBook.py")) + "/sounds/beep.wav") - config['beep_reduction'])
def onHook():
print("Phone on hook. Sleeping...")
# print("Resetting dial list")
# global dialed
# dialed = []
# reset_pulse_counter()
# def dialing():
# if hook.is_pressed:
# global count, reset_flag
# count+=1
# print(f"dialing, increment count: {count}")
# reset_flag = False
# def reset_pulse_counter():
# global count, reset_flag
# count = 0
# print(f"reset count: {count}")
# reset_flag = True
# def held():
# if not reset_flag:
# print("holding")
# print(count)
# global dialed
# if (count == 10):
# dialed.append(0)
# else:
# dialed.append(count)
# print(f"number dialed: {dialed}")
# offHook()
# reset_pulse_counter()
def main():
# rotaryDial.when_pressed = dialing
# rotaryDial.when_held = held
hook.when_pressed = offHook
hook.when_released = onHook
pause()
if __name__ == "__main__":
main()

BIN
sounds/beep.wav Normal file

Binary file not shown.

View File

@ -0,0 +1,136 @@
#! /usr/bin/env python3
import audioInterface
import os
import yaml
import sys
from datetime import datetime
from gpiozero import Button
from multiprocessing import Process
from signal import pause
from pydub import AudioSegment
from pydub.playback import play
from pydub.scipy_effects import band_pass_filter
from pydub.effects import normalize, compress_dynamic_range
try:
with open("config.yaml") as f:
config = yaml.load(f, Loader=yaml.FullLoader)
except FileNotFoundError as e:
print(
f"Could not find the config.yaml file. FileNotFoundError: {e}. Check config location and retry."
)
sys.exit(1)
hook = Button(config["hook_gpio"])
rotaryDial = Button(pin=config['rotary_gpio'], hold_time=config['rotary_hold_time'], hold_repeat=config['rotary_hold_repeat'])
"""
TODO: These globals are a temp solution for the rotary dialer, would love to not
depend on globals for this logic.
"""
count = 0
dialed = []
reset_flag = False
def off_hook() -> None:
print("Phone off hook, ready to begin!")
# if dialed and dialed[0] == 0:
audio_interface = audioInterface.AudioInterface(config, hook)
# playback voice message through speaker
print("Playing voicemail message...")
play(
AudioSegment.from_wav(
os.path.dirname(os.path.abspath(config["source_file"]))
+ "/sounds/voicemail.wav"
)
- config["playback_reduction"]
)
# start recording beep
print("Playing beep...")
play(
AudioSegment.from_wav(
os.path.dirname(os.path.abspath(config["source_file"])) + "/sounds/beep.wav"
)
- config["beep_reduction"]
)
# now, while phone is not off the hook, record audio from the microphone
print("recording")
audio_interface.record()
audio_interface.stop()
output_file = (
os.path.dirname(os.path.abspath(config["source_file"]))
+ "/recordings/"
+ f"{datetime.now().isoformat()}"
)
audio_interface.close(output_file + ".wav")
print("Finished recording!")
"""
post processing
"""
print("spawn postProcessing thread")
Process(target=audio_interface.postProcess, args=(output_file,)).start()
"""
rotary dialer special messages
"""
if dialed[0:3] == [9,2,7]:
# play special vm
play(AudioSegment.from_wav(os.path.dirname(os.path.abspath(config["source_file"])) + "/sounds/927.wav") - config['playback_reduction'])
elif dialed[0:4] == [5,4,5,3]:
# play special vm
play(AudioSegment.from_wav(os.path.dirname(os.path.abspath(config["source_file"])) + "/sounds/beep.wav") - config['beep_reduction'])
def on_hook() -> None:
print("Phone on hook. Sleeping...")
print("Resetting dial list")
global dialed
dialed = []
reset_pulse_counter()
def dialing() -> None:
if hook.is_pressed:
global count, reset_flag
count+=1
print(f"dialing, increment count: {count}")
reset_flag = False
def reset_pulse_counter() -> None:
global count, reset_flag
count = 0
print(f"reset count: {count}")
reset_flag = True
def held() -> None:
if not reset_flag:
print("holding")
print(count)
global dialed
if (count == 10):
dialed.append(0)
else:
dialed.append(count)
print(f"number dialed: {dialed}")
off_hook()
reset_pulse_counter()
def main():
rotaryDial.when_pressed = dialing
rotaryDial.when_held = held
hook.when_pressed = off_hook
hook.when_released = on_hook
pause()
if __name__ == "__main__":
main()