Merge pull request '#33 Implement Recording' (#34) from #33-Implement-Recording into master

Reviewed-on: #34

.gitignore

@@ -6,5 +6,6 @@ node_modules/
 *.log
 *.txt
 *.env
+*.wav
 !requirements.txt
 *testOP25Dir/

Client/controllers/botController.js

@@ -10,6 +10,7 @@ const { closeProcessWrapper } = require("../utilities/utilities");
 
 // Global vars
 let pythonProcess;
+let recordingProcess;
 
 /**
  * Get Status of the discord process
@@ -81,4 +82,78 @@ exports.leaveServer = async (req, res) => {
     pythonProcess = await closeProcessWrapper(pythonProcess);
     
     return res.sendStatus(202);
+}
+
+/**
+ * Start a recording of what the bot is listening to, if it's currently connected
+ * 
+ * @param {*} req 
+ * @param {*} res 
+ */
+exports.startRecording = async (req, res) => {
+    log.INFO("Starting recording")
+    //if (pythonProcess === undefined) return res.sendStatus(204);
+    if (!recordingProcess === undefined) return res.sendStatus(202);
+    const deviceId = process.env.AUDIO_DEVICE_ID;
+    const filename = "./recordings/" + new Date().toJSON().slice(0,10) + ".wav";
+
+    // Joining the server to record
+    log.INFO("Start recording: ", deviceId, filename);
+    if (process.platform === "win32") {
+        log.DEBUG("Starting Windows Python");        
+        recordingProcess = await spawn('python', [resolve(__dirname, "../pdab/recorder.py"), deviceId, filename ], { cwd: resolve(__dirname, "../pdab/").toString() });        
+    }
+    else {
+        log.DEBUG("Starting Linux Python");
+        recordingProcess = await spawn('python3', [resolve(__dirname, "../pdab/recorder.py"), deviceId, filename ], { cwd: resolve(__dirname, "../pdab/") });
+    }
+
+    await this.getProcessOutput(recordingProcess);
+    
+    return res.sendStatus(200);
+}
+
+/**
+ * Stop the recording if the bot is currently recording
+ * 
+ * @param {*} req 
+ * @param {*} res 
+ */
+exports.stopRecording = async (req, res) => {
+    log.INFO("Stopping recording the server");
+    if (!recordingProcess) return res.sendStatus(202)
+
+    recordingProcess = await closeProcessWrapper(recordingProcess);
+    
+    return res.sendStatus(200);
+}
+
+
+/**
+ * Get the output of a running process
+ * 
+ * @param {*} runningProcess 
+ * @returns 
+ */
+exports.getProcessOutput = async (runningProcess) => {
+    let fullOutput;
+    runningProcess.stdout.setEncoding('utf8');    
+    runningProcess.stdout.on("data", (data) => {
+        botLog.VERBOSE("From Process: ", data);
+        fullOutput += data.toString();
+    });
+
+    runningProcess.stderr.on('data', (data) => {
+        botLog.VERBOSE(`stderr: ${data}`);
+        fullOutput += data.toString();
+      });
+      
+      runningProcess.on('close', (code) => {
+    log.DEBUG(`child process exited with code ${code}`);
+    log.VERBOSE("Full output from bot: ", fullOutput);
+    }); 
+
+    runningProcess.on("error", (code, signal) => {
+        log.ERROR("Error from the process: ", code, signal);
+    });
 }

Client/pdab/recorder.py

@@ -0,0 +1,150 @@
+import pyaudio
+import wave, logging, threading, time, queue, signal, argparse
+from os import path, makedirs
+
+logging.basicConfig(format="%(asctime)s: %(message)s", level=logging.INFO,datefmt="%H:%M:%S")
+
+class DiscordRecorder:
+    def __init__(self, DEVICE_ID, CHUNK = 1024, FORMAT = pyaudio.paInt16, CHANNELS = 2, RATE = 48000, FILENAME = "./recs/radio.wav"):
+        self.pa_instance = pyaudio.PyAudio()
+
+        self.DEVICE_ID = DEVICE_ID
+        self.CHUNK = CHUNK
+        self.FORMAT = FORMAT
+        self.CHANNELS = CHANNELS
+        self.RATE = RATE
+
+        self.FILENAME = FILENAME
+        self._check_file_path_exists()
+
+        self.queued_frames = queue.Queue()
+
+        self.stop_threads = threading.Event()
+
+        self.recording_thread = None
+        self.saving_thread = None
+
+        self.running_stream = None
+
+    # Wrapper to check if the given filepath (not file itself) exists
+    def _check_file_path_exists(self):
+        if not path.exists(path.dirname(self.FILENAME)):            
+            makedirs(path.dirname(self.FILENAME), exist_ok=True)
+
+    # Wrapper for the recorder thread; Adds new data to the queue
+    def _recorder(self):
+        logging.info("* Recording Thread Starting")
+        while True:
+            data = self.running_stream.read(self.CHUNK)            
+            self.queued_frames.put(data)
+
+            # check for stop
+            if self.stop_threads.is_set():
+                break
+
+    # Wrapper for saver thread; Saves the queue to the file
+    def _saver(self):
+        logging.info("* Saving Thread Starting")
+        while True:
+            if not self.queued_frames.empty():
+                dequeued_frames = []
+                for i in range(self.queued_frames.qsize()):
+                    dequeued_frames.append(self.queued_frames.get())
+
+                if not path.isfile(self.FILENAME):
+                    wf = wave.open(self.FILENAME, 'wb')
+                    wf.setnchannels(self.CHANNELS)
+                    wf.setsampwidth(self.pa_instance.get_sample_size(self.FORMAT))
+                    wf.setframerate(self.RATE)
+                    wf.writeframes(b''.join(dequeued_frames))
+                    wf.close()
+                else:                                
+                    read_file = wave.open(self.FILENAME, 'rb')
+                    read_file_data = read_file.readframes(read_file.getnframes())
+                    read_file.close()
+                        
+                    wf = wave.open(self.FILENAME, 'wb')
+                    wf.setnchannels(self.CHANNELS)
+                    wf.setsampwidth(self.pa_instance.get_sample_size(self.FORMAT))
+                    wf.setframerate(self.RATE)
+
+                    wf.writeframes(read_file_data)
+                    wf.writeframes(b''.join(dequeued_frames))
+                    wf.close()
+
+            # check for stop
+            if self.stop_threads.is_set():
+                break
+
+            time.sleep(5)
+
+    # Start the recording function
+    def start_recording(self):
+        logging.info("* Recording")
+
+        self.running_stream = self.pa_instance.open(
+            input_device_index=self.DEVICE_ID,
+            format=self.FORMAT,
+            channels=self.CHANNELS,
+            rate=self.RATE,
+            input=True,
+            frames_per_buffer=self.CHUNK
+        )
+        
+        self.recording_thread = threading.Thread(target=self._recorder)
+        self.recording_thread.start()
+
+        self.saving_thread = threading.Thread(target=self._saver)
+        self.saving_thread.start()
+    
+    # Stop the recording function
+    def stop_recording(self):
+        self.stop_threads.set()
+        self.recording_thread.join()
+        self.saving_thread.join()
+        self.running_stream.stop_stream()
+        self.running_stream.close()
+        self.pa_instance.terminate()
+        
+        logging.info("* Done recording")
+
+
+class GracefulExitCatcher:
+  def __init__(self, stop_callback):
+    self.stop = False
+
+    # The function to run when the exit signal is caught
+    self.stop_callback = stop_callback
+
+    # Update what happens when these signals are caught
+    signal.signal(signal.SIGINT, self.exit_gracefully)
+    signal.signal(signal.SIGTERM, self.exit_gracefully)
+
+  def exit_gracefully(self, *args):
+    logging.info("* Stop signal caught...")
+    
+    # Stop the main loop
+    self.stop = True
+
+    # Run the given callback function
+    self.stop_callback()
+
+
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("deviceId", type=int, help="The ID of the audio device to use")
+    parser.add_argument("filename", type=str, help="The filepath/filename of the output file")    
+    args = parser.parse_args()
+
+    logging.debug("Arguments:", args)
+    
+    recorder = DiscordRecorder(args.deviceId, FILENAME=args.filename)
+
+    exit_catcher = GracefulExitCatcher(recorder.stop_recording)
+
+    recorder.start_recording()
+
+    while not exit_catcher.stop:
+        time.sleep(1)

Client/routes/bot.js

@@ -27,4 +27,20 @@ router.post('/join', botController.joinServer);
  */
 router.post('/leave', botController.leaveServer);
 
+/** POST bot start recording
+ * If the bot is in a channel, it will start to record what it hears
+ *
+ * The status of the bot: 200 = starting to record, 202 = already recording, 204 = not in a server, 500 + JSON = encountered error
+ * @returns status
+ */
+router.post('/startRecording', botController.startRecording);
+
+/** POST bot stop recording
+ * If the bot is recording, it will stop recording
+ *
+ * The status of the bot: 200 = will stop the recording, 204 = not currently recording, 500 + JSON = encountered error
+ * @returns status
+ */
+router.post('/stopRecording', botController.stopRecording);
+
 module.exports = router;

Server/commands/start-record.js

@@ -0,0 +1,37 @@
+const { SlashCommandBuilder } = require('discord.js');
+const { DebugBuilder } = require("../utilities/debugBuilder");
+const log = new DebugBuilder("server", "start-record");
+const { getAllConnections } = require("../utilities/mysqlHandler");
+const { requestOptions, sendHttpRequest } = require("../utilities/httpRequests");
+
+module.exports = {
+	data: new SlashCommandBuilder()
+	.setName('start-record')
+	.setDescription('Starts recording all bots online'),
+	example: "start-record",
+	isPrivileged: false,
+	requiresTokens: false,
+	defaultTokenUsage: 0,
+	deferInitialReply: false,
+	async execute(interaction) {
+		try{
+			await interaction.reply(`Ok, ${interaction.member}. **Recording** will begin shorting.`);
+			// Get nodes online			
+			getAllConnections((connections) => {
+				for (const connection of connections){
+					const reqOptions = new requestOptions("/bot/startRecording", "POST", connection.node.ip, connection.node.port);
+					sendHttpRequest(reqOptions, JSON.stringify({}), async (responseObj) => {
+						log.VERBOSE("Response Object from node: ", connection, responseObj);
+						if (!responseObj || !responseObj.statusCode == 202 || !responseObj.statusCode == 204) return false;
+						if (responseObj.statusCode >= 300) return log.ERROR(responseObj.body);
+						// Bot is recording
+						await interaction.channel.send(`**${connection.clientObject.name} is now recording**`); 
+					});
+				}					
+			});						
+		}catch(err){
+			log.ERROR(err)        
+            //await interaction.reply(err.toString());
+		}
+	}
+};

Server/commands/stop-record.js

@@ -0,0 +1,37 @@
+const { SlashCommandBuilder } = require('discord.js');
+const { DebugBuilder } = require("../utilities/debugBuilder");
+const log = new DebugBuilder("server", "stop-record");
+const { getAllConnections } = require("../utilities/mysqlHandler");
+const { requestOptions, sendHttpRequest } = require("../utilities/httpRequests");
+
+module.exports = {
+	data: new SlashCommandBuilder()
+	.setName('stop-record')
+	.setDescription('Starts recording all bots online'),
+	example: "stop-record",
+	isPrivileged: false,
+	requiresTokens: false,
+	defaultTokenUsage: 0,
+	deferInitialReply: false,
+	async execute(interaction) {
+		try{
+			await interaction.reply(`Ok, ${interaction.member}. **Recording** will stop shorting.`);
+			// Get nodes online			
+			getAllConnections((connections) => {
+				for (const connection of connections){
+					const reqOptions = new requestOptions("/bot/stopRecording", "POST", connection.node.ip, connection.node.port);
+					sendHttpRequest(reqOptions, JSON.stringify({}), async (responseObj) => {
+						log.VERBOSE("Response Object from node: ", connection, responseObj);
+						if (!responseObj || !responseObj.statusCode == 204) return false;
+						if (responseObj.statusCode >= 300) return log.ERROR(responseObj.body);
+						// Bot is recording
+						await interaction.channel.send(`**${connection.clientObject.name} has stopped recording**`); 
+					});
+				}					
+			});						
+		}catch(err){
+			log.ERROR(err)        
+            //await interaction.reply(err.toString());
+		}
+	}
+};

Server/events/interactionCreate.js

@@ -9,7 +9,7 @@ module.exports = {
 	name: Events.InteractionCreate,
 	async execute(interaction) {
 		const command = interaction.client.commands.get(interaction.commandName);
-		log.VERBOSE("Interaction for command: ", command);
+		log.VERBOSE("Interaction created for command: ", command);
 
 		// Execute autocomplete if the user is checking autocomplete
 		if (interaction.isAutocomplete()) {