Multiple cameras websocket view - stream with raspivid, nodejs, and angular

SO, I have this code on multiple raspi cameras:

const express = require('express');
const WebSocket = require('ws');
const raspividStream = require('raspivid-stream');
const fs = require('fs');
const readline = require('readline');
const app = express();
const wss = require('express-ws')(app);
const { exec } = require('child_process');
const path = require('path');

var vstreamCounter = 0;
var videoStream = null;
var recording = false;
var websoc = null;

var currentDir = __dirname;
var camfn = 'camera.conf';
var p = path.join(currentDir, camfn);


if (fs.existsSync(p)) {

const fileStream = fs.createReadStream(p);

fileStream.on('error', (error) => {
console.error(`Error reading file: ${error.message}`);
});

const rl = readline.createInterface({
input: fileStream,
crlfDelay: Infinity
});

rl.on('line', (line) => {
console.log(`Got ip from file: ${line}`);

var serverUrl = 'ws://'+line+':1337';
console.log(serverUrl);
websoc = new WebSocket(serverUrl);

websoc.on('open', function open() {
console.log('Connected to the server');
});
});

rl.on('close', () => {
console.log('File reading finished');
});

} else {
console.log("missing configuration file camera.conf");
process.exit();
}

app.get('/camera-status', (req, res) => {

exec('vcgencmd get_camera', (error, stdout, stderr) => {
if (error) {
console.error(`Error: ${error.message}`);
return;
}
if (stderr) {
console.error(`stderr: ${stderr}`);
return;
}
console.log(`stdout: ${stdout}`);

var position = stdout.indexOf(",");

if (position != -1) {
var sbstr = stdout.substring(0, position);
var rgxp = new RegExp("1", 'g');
var occurences = sbstr.match(rgxp);

if (occurences.length == 2){
res.send(200);
} else {
res.send(500);
}
}

});
});

app.ws('/vstream', async (ws, req) => {
console.log('Client connected');

vstreamCounter = vstreamCounter + 1;

ws.send(JSON.stringify({
action: "init",
width: 640,
height: 480
}));

videoStream = await raspividStream();

videoStream.on('data', (data) => {
if (ws.readyState === WebSocket.OPEN) {
ws.send(data, { binary: true }, (error) => { if (error) {console.error(error)/* process.exit()*/;} });


}
});

ws.on("message", async (msg) => {
await messageHandling(msg, null);
});

if (ws.readyState === WebSocket.OPEN) {
ws.on('close', async () => {
await closing(videoStream);
});
}
});

app.ws('/vstream-90', async (ws, req) => {
console.log('Client connected');

vstreamCounter = vstreamCounter + 1;

ws.send(JSON.stringify({
action: "init",
width: 640,
height: 480,
rotation: 90
}));

videoStream = await raspividStream({ rotation: 90 });

videoStream.on('data', (data) => {
if (ws.readyState === WebSocket.OPEN) {
ws.send(data, { binary: true }, (error) => { if (error) {console.error(error)/* process.exit()*/;} });
}
});

ws.on("message", async (msg) => {
await messageHandling(msg, 90);
});

if (ws.readyState === WebSocket.OPEN) {
ws.on('close', async () => {
await closing(videoStream);
});
}
});

app.ws('/vstream-180', async (ws, req) => {

console.log('Client connected');

vstreamCounter = vstreamCounter + 1;

ws.send(JSON.stringify({
action: "init",
width: 640,
height: 480,
rotation: 180
}));

videoStream = await raspividStream({ rotation: 180 });

videoStream.on('data', async (data) => {

if (ws.readyState === WebSocket.OPEN) {
ws.send(data, { binary: true }, (error) => { if (error) {console.error(error)/* process.exit()*/;} });
}
});

ws.on("message", async (msg) => {
await messageHandling(msg, 180);
});

if (ws.readyState === WebSocket.OPEN) {
ws.on('close', async () => {
await closing(videoStream);
});}
});


var recording = true;
var ss = null;
var recordLock = 0;

async function messageHandling(msg, rot){

if (msg.toString().includes("loc-")){
websoc.send(msg);
}

if (msg.toString().includes("desc-")){
websoc.send(msg);
}

if (msg.toString() === "record"){

//if (recordLock == 0) {

recordLock = recordLock + 1;

if (rot){
ss = await raspividStream({ rotation: rot });
} else {
ss = await raspividStream();
}

// if (recording){

ss.on('data', async (d) => {
if (websoc.readyState === WebSocket.OPEN) {
websoc.send(d, { binary: true }, (error) => { if (error) {console.error(error)/* process.exit()*/;} });
}
});
//}
//}
}

if (msg.toString() === "stoprecord"){
// if (recording) {
recording = false;
websoc.send("stoprecord");
console.log("stopped recording");
//process.exit();
// }
}
}

async function closing(){
console.log('Client left');

videoStream.removeAllListeners('data');

vstreamCounter = vstreamCounter - 1;

//await burnThemInClose(filename, location);

console.log(vstreamCounter);

if (vstreamCounter === 0){
process.exit();
}
}

app.listen(8080, function(err){
if (err) console.log("Error in server setup")
console.log("Server listening on Port");
});


This is server side code that cameras connect to for processing the stream:

const WebSocket = require('ws')
const fs = require('fs');
const ffmpeg = require('fluent-ffmpeg');
const { finished } = require('stream');
const { promisify } = require('util');

const finishedAsync = promisify(finished);
const writeFileAsync1 = promisify((stream, data, callback) => {
stream.write(data, callback);
});

const server = new WebSocket.Server({ port: 1337 });
let location = null;

server.on('connection', (ws) => {
console.log('Client connected');

let writeStream = null;
let filename = null;

let view = null;
let camcnt = 0;

ws.on('message', async (message) => {

// console.log("in message handler");
const msgStr = message.toString();

if (msgStr.includes("desc-")) {
view = msgStr.slice(5);
console.log(`View set to: ${view}`);
}

if (msgStr.includes("loc-")) {
location = msgStr.slice(4);
console.log(`Location set to: ${location}`);
}

if (msgStr.includes("cam") && camcnt === 0) {
console.log(`Camera message received: ${msgStr}`);

if (!location || !view) {
console.error('Location or view not set. Cannot create filename.');
return;
}

filename = generateFileName(location, view);
writeStream = fs.createWriteStream(filename);
console.log(`Write stream created: ${filename}`);
camcnt++;
}

if (writeStream && !msgStr.startsWith("desc-") && !msgStr.startsWith("loc-") && !msgStr.includes("cam1") && !msgStr.includes("cam2")) {
if (!writeStream.writableEnded) {
try {
await writeFileAsync1(writeStream, message);
console.log('Data written to file');
} catch (error) {
console.error('Error writing to stream:', error);
}
}
}

if (msgStr === "stoprecord") {
if (writeStream) {
if (!writeStream.writableEnded) {
writeStream.end();
try {
await finishedAsync(writeStream);
console.log('Write stream ended');
} catch (error) {
console.error('Error ending the stream:', error);
}
}
}
// ws.close();
console.log('WebSocket connection closed');
}
});

ws.on('close', async () => {
console.log('Client disconnected');
if (writeStream && !writeStream.writableEnded) {
writeStream.end();
try {
await finishedAsync(writeStream);
console.log('Write stream ended');
} catch (error) {
console.error('Error ending the stream:', error);
}
}
});
});






function generateFileName(location, view) {
var d = new Date();
var filename = location +"_"+view+"_"+
d.getFullYear() + "-" +
("00" + (d.getMonth() + 1)).slice(-2) + "-" +
("00" + d.getDate()).slice(-2) + "_" +
("00" + d.getHours()).slice(-2) + ":" +
("00" + d.getMinutes()).slice(-2) + ":" +
("00" + d.getSeconds()).slice(-2) + ".h264";

filename = filename.replace(/["']/g, "");
filename = filename.replace(/:/g, "_");

return filename;
}

async function reencodeVideo(inputPath, outputPath) {
return new Promise((resolve, reject) => {
ffmpeg(inputPath)
.videoCodec('libx264')
.audioCodec('aac')
.outputOptions('-movflags', 'faststart') // Ensures compatibility with streaming
.on('end', () => {
console.log('Video re-encoding complete.');
resolve();
})
.on('error', (err) => {
console.error('Error:', err);
reject(err);
})
.save(outputPath);
});
}

function getVideoMetadata(videofilename) {
return new Promise((resolve, reject) => {
ffmpeg.ffprobe(videofilename, (err, metadata) => {
if (err) {
reject(err);
} else {
resolve(metadata);
}
});
});
}

function generateSrtContent(duration, staticText, startTime) {

// Convert start time to seconds
const [hours, minutes, seconds] = startTime.split(':').map(Number);
const startTimeSeconds = hours * 3600 + minutes * 60 + seconds;

let srtContent = '';

for (let i = 0; i < duration; i++) {
const currentTime = startTimeSeconds + i;
const start = new Date(i * 1000).toISOString().substr(11, 8) + ',000';
const end = new Date((i + 1) * 1000).toISOString().substr(11, 8) + ',000';
const timeText = new Date(currentTime * 1000).toISOString().substr(11, 8);

srtContent += `${i + 1}\n`;
srtContent += `${start} --> ${end}\n`;
srtContent += `${staticText} ${timeText}\n\n`;
}

return srtContent;
}

function writeFileAsync(filePath, content) {
return new Promise((resolve, reject) => {
fs.writeFile(filePath, content, (err) => {
if (err) {
reject(err);
} else {
resolve();
}
});
});
}

function encodeVideoWithSubtitles(inputVideo, subtitleFile, outputVideo) {
return new Promise((resolve, reject) => {
ffmpeg(inputVideo)
.outputOptions('-vf', `subtitles=${subtitleFile}`)
.output(outputVideo)
.on('end', () => {
console.log('Video processing complete.');
resolve();
})
.on('error', (err) => {
console.error('Error:', err);
reject(err);
})
.run();
});
}

var staticcnt = 0;

async function burnThemInClose(filename, location){
try {
var d = new Date();
var hhmmss = ("00" + d.getHours()).slice(-2) + ":" +
("00" + d.getMinutes()).slice(-2) + ":" +
("00" + d.getSeconds()).slice(-2);

var staticTime = d.getFullYear() + "-" +
("00" + (d.getMonth() + 1)).slice(-2) + "-" +
("00" + d.getDate()).slice(-2);

var newMP4 = filename.slice(0, -5)+".mp4"
await reencodeVideo(filename, "static"+staticcnt+".mp4");

const metadata = await getVideoMetadata("static"+staticcnt+".mp4");
staticnt = staticcnt + 1;
console.log(metadata);
const duration = Math.floor(metadata.format.duration);
const outputSrt = "titulky.srt";
const srtContent = generateSrtContent(duration, location+" "+staticTime, hhmmss);

if (fs.existsSync(outputSrt)) {
fs.unlinkSync(outputSrt);
}

await writeFileAsync(outputSrt, srtContent);

// await encodeVideoWithSubtitles("static.mp4", outputSrt, newMP4);

if (fs.existsSync("static.mp4")) {
fs.unlinkSync("static.mp4");
}

if (fs.existsSync(filename)) {
// fs.unlinkSync(filename);
}

if (fs.existsSync(outputSrt)) {
fs.unlinkSync(outputSrt);
}

console.log('encoding finished');
} catch (err) {
console.error('Error:', err);
}
}


adn this is angluar function for changing the streams:

async changeStream(camera: RobotComponent) {

this.selectedCamera = camera;

console.log(this.selectedCamera);


if (camera.split_screen){

this.playerOne.send("stoprecord");
this.playerTwo.send("stoprecord");
this.clearCanvas();
this.canvasTwoElement.hidden = false;


this.playerOne.disconnect();
this.playerTwo.disconnect();

this.playerOne.removeAllListeners();
this.playerTwo.removeAllListeners();


let ws = new WebSocket(this.selectedCamera.camera_stream);
let ws2 = new WebSocket(this.selectedCamera.camera_stream_two);

this.playerOne.connectWithCustomClient(ws);
this.playerTwo.connectWithCustomClient(ws2);


await this.delay(500);

this.playerOne.send("desc-"+camera.description+"_cam1");
await this.delay(1500);
this.playerOne.send("record");

await this.delay(500);

this.playerTwo.send("desc-"+camera.description+"_cam1");
await this.delay(1500);
this.playerTwo.send("record");

//this.playerOne.connectByUrl(camera.camera_stream);
//this.playerTwo.connectByUrl(camera.camera_stream_two);

console.log(this.playerOne);
console.log(this.playerTwo);


this.useSplitScreen = true;
} else {
// this.stopStream();
this.clearCanvas()

this.playerOne.disconnect;
this.playerOne.removeAllListeners;

this.playerOne.connectByUrl(camera.camera_stream);
await this.delay(500);
//this.playerOne.send('REQUESTSTREAM');

this.playerOne.send("desc-"+camera.description+"_cam1");
await this.delay(1500);
this.playerOne.send("record");
this.useSplitScreen = false;
}
}


I would like to start new write stream on processing serverside, when the angular changeStream function is called, with new view - with two cameras. Problem is that first loaded view is recorded to two separate files which is correct, but when i call change stream in angular, first two streams are ended which is correct but second view write streams are not created on serverside despite that "record" message is sent to the new newly created websocket stream.

Can any one help out please.

Continue reading...