server.py

#!/usr/bin/python3

import sys, gi, os, socket, time, subprocess, psutil

from cam import Camera
from audio import AudioStream

gi.require_version('Gst', '1.0')
gi.require_version('GLib', '2.0')
from gi.repository import GObject, Gst, GstRtspServer, GLib, GObject


class Controller:
    def __init__(self):
        Gst.init(sys.argv)
        self.mainloop = GLib.MainLoop()
        self.socket = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socket.bind(('', 6780))
        self.socket.listen(8)
        self.socket.setblocking(0)
        self.conns = {}
        self.savemsgs = {}
        self.rtsp_server = GstRtspServer.RTSPServer.new()
        self.rtsp_server.set_service('8080')
        self.rtsp_server.set_address('::')
        self.rtsp_server.attach(None)

        GLib.io_add_watch(self.socket.fileno(),
                          GLib.IO_IN | GLib.IO_ERR | GLib.IO_HUP,
                          self.server_msg)

        self.cams = []
        self.audios = []

    def add_camera(self, name, type, device = None, width = 1280, height = 720, framerate = 30, audio = None):
        cam = Camera(name, {
            'rtsp': self.rtsp_server,
            'audio': audio if audio or not self.audios else self.audios[0],
            'video_source': type, 'device': device, 'width': width, 'height': height, 'framerate': framerate})
        self.cams.append(cam)

        return cam

    def server_msg(self, source, cond):
        if (cond & GLib.IO_IN):
            client, addr = self.socket.accept()
            client.setblocking(0)
            self.conns[client.fileno()] = client
            self.savemsgs[client.fileno()] = None
            GLib.io_add_watch(client.fileno(),
                              GLib.IO_IN | GLib.IO_ERR | GLib.IO_HUP,
                              self.client_msg)
            print("New client from ", addr)
        else:
            print("cond: %x\n" % cond)

        return True

    def client_msg(self, source, cond):
        socket = self.conns[source]

        if cond & GLib.IO_IN:
            ba = socket.recv(4096)
            if not ba:
                cond |= GLib.IO_HUP

            if self.savemsgs[source]:
                ba = self.savemsgs[source] + ba

            nl = ba.find(b'\n')
            while nl >= 0:
                str = ba[0:nl].decode(errors='replace')
                self.run_command(socket, str)
                ba = ba[nl+1:]
                nl = ba.find(b'\n')

            self.savemsgs[source] = ba

        if cond & (GLib.IO_ERR | GLib.IO_HUP):
            socket.close()
            del self.conns[source]
            del self.savemsgs[source]
            return False

        return True

    def run_command(self, socket, str):
        if str.startswith(':'):
            name, cmd = str.split(' ', 1)
            name = name[1:]

            for cam in self.cams:
                if cam.name == name:
                    cam.run_command(socket, cmd)
                    return
            socket.send(b'Invalid camera ' + name.encode() + b'\n')
        elif str:
            for cam in self.cams:
                cam.run_command(socket, str)
        else:
            socket.send(b'Usage: (:camera) <command>\n')
        return

    def shutdown(self):
        for cam in self.cams:
            cam.shutdown()

        self.snd.set_state(Gst.State.NULL)

        self.mainloop.quit()

        self.socket.close()

        for socket in self.conns:
            socket.close()

        self.conns = {}
        self.savemsgs = {}

    def add_audio_stream(self, audio, shared):
        sound_stream = GstRtspServer.RTSPMediaFactory()
        sound_stream.set_shared(shared)
        sound_stream.set_latency(100)
        sound_stream.set_launch(audio.shmsrc() + ' ! queue ! ' + audio.payloader() + ' name=pay0')
        self.rtsp_server.get_mount_points().add_factory('/snd.' + audio.name, sound_stream)

        if not self.audios:
            self.rtsp_server.get_mount_points().add_factory('/snd', sound_stream)

        self.audios.append(audio)

    def run(self):
        for audio in self.audios:
            audio.run()

        for cam in self.cams:
            cam.run()

        self.mainloop.run()

if __name__ == '__main__':
    main = Controller()
    for x in os.listdir('/sys/class/sound/'):
        if x.startswith('pcmC') and x.endswith('D0c') and len(x) == 8:
            hwname = 'hw:' + x[4] + ',0'
            if psutil.cpu_count() > 1:
                audio_pipe = (
                    f'alsasrc latency-time=32000 buffer-time=256000 do-timestamp=1 device={hwname} ! '
                    f'audio/x-raw, format=S16LE, channels=2 ! '
                    f'voaacenc bitrate=64000 ! audio/mpeg, mpegversion=4, stream-format=raw ! aacparse ')

                audio = AudioStream(hwname, audio_pipe, 'rtpmp4apay', 'aacparse')
                main.add_audio_stream(audio, False)
            else:
                audio_pipe = (
                    f'alsasrc latency-time=32000 buffer-time=180000 do-timestamp=1 device={hwname} ! '
                    f'audio/x-raw, format=S16LE, channels=2 ! '
                    f'lamemp3enc ! audio/mpeg ! mpegaudioparse ')

                audio = AudioStream(hwname, audio_pipe, 'rtpmpapay', 'mpegaudioparse')
                main.add_audio_stream(audio, True)

    if socket.gethostname() == 'raspberry':
        main.add_camera('cam', 'rpicamsrc')
    elif socket.gethostname() == 'cam1':
        main.add_camera('cam2', 'libcamerasrc')
    elif socket.gethostname() == 'cam0':
#        main.add_camera('cam0', 'libcamerasrc')
        main.add_camera('cam1', 'uvch264src', '/dev/video2', audio = main.audios[1])
        main.add_camera('cam0', 'v4l2src', '/dev/video0', 2560, 1440, 30)
    elif socket.gethostname() == 'cam4':
        main.add_camera('cam4', 'uvch264src')
    elif socket.gethostname() == 'cam5':
        main.add_camera('cam5', 'rpicamsrc')

    main.run()