For use with trunk-recorder & liquidsoap

common.liquidsoap

set("tag.encodings",["UTF-8","ISO-8859-1"])

# Configure Logging
set("log.file",false)
set("log.level",3)
set("log.stdout",true)
set("log.syslog",false)
set("log.syslog.facility","DAEMON")
set("log.syslog.program","liquidsoap-#{STREAMID}")

# create a socket to send commands to this instance of liquidsoap
set("server.socket",true)
set("server.socket.path","/tmp/sockets/#{STREAMID}.sock")
set("server.socket.permissions",511)
# This creates a 1 second silence period generated programmatically (no disk reads)
# silence = amplify(0.1, noise(duration=1.))
silence = blank(duration=1.)

# This pulls the alpha tag out of the wav file
def append_title(m) =
    [("title",">> Scanning <<")]
end

silence = map_metadata(append_title, silence)

recorder_queue = request.queue()
recorder_queue = server.insert_metadata(id="S4",recorder_queue)

# If there is anything in the queue, play it.  If not, play the silence defined above repeatedly:
stream = fallback(track_sensitive=false, [recorder_queue, silence])

title = '$(if $(title),"$(title)","...Scanning...")'
stream = rewrite_metadata([("title", title)], stream)

output.icecast( %mp3(stereo=false, bitrate=96, samplerate=22050, internal_quality=9, msg="testing"),
  host=HOST, port=PORT, password=PASSWORD, genre="Scanner",
  description="Scanner audio", mount=MOUNT,  name=NAME, user="source", stream)

queue-to-liquidsoap.py

#!/usr/bin/env python3
import socket
import sys
import os
import csv
import json
import logging
import sys
import subprocess

# TODO: Read various file metadata from the associated JSON file rather than derpy parsing of the filename

logging.basicConfig(
    stream=sys.stdout,
    format='[%(asctime)s] %(levelname)s - %(message)s',
    level=logging.DEBUG,
)


def ffmpeg_filter_string(options):
    return ':'.join([f"{k}={v}" for k, v in options.items()])


class Configuration(object):
    def __init__(self, file="config.json"):
        super(Configuration, self).__init__()
        with open(file) as json_file:
            self.data = json.load(json_file)

    def read(self, key, default):
        try:
            value = self.data
            for key in key.split('.'):
                value = value[key]
            return value
        except KeyError as e:
            return default


class Talkgroups(object):

    def __init__(self, talkgroups_file):
        super(Talkgroups, self).__init__()
        self.talkgroups = []
        with open(os.path.abspath(talkgroups_file), newline='') as csvfile:
            rowreader = csv.reader(csvfile)
            for row in rowreader:
                self.talkgroups.append(row)

    def name(self, talkgroup: int):
        for row in self.talkgroups:
            if int(row[0]) == talkgroup:
                return str(row[3])
        return 'Unknown'

    def streams(self, talkgroup: int):
        for row in self.talkgroups:
            if int(row[0]) == talkgroup:
                return row[8].split('|')
        return []


class Talkgroup(object):

    def __init__(self, name, number: int, streams: [str]):
        super(Talkgroup, self).__init__()
        self.name = name
        self.number = number
        self.streams = streams

    def __str__(self):
        return f"Talkgroup : name={self.name} number={self.number} streams={self.streams}"


class TrunkRecorder(object):
    def __init__(self, config: Configuration):
        super(TrunkRecorder, self).__init__()
        self.config = config

    def system_name(self, recording) -> str:
        name = recording.replace(self.config.read('captureDir', os.getcwd()), '')
        logging.debug(f'system_name {name}')
        logging.debug(self.config.read('captureDir', 'no'))
        return str(os.path.normpath(name).lstrip(os.path.sep).split(os.path.sep)[0])

    def talkgroup_number(self, file) -> int:
        return int(os.path.basename(file).split('-')[0])

    def talkgroup(self, file):
        system_name = self.system_name(file)
        logging.debug(f'system_name {system_name}')
        talkgroup = self.talkgroup_number(file)
        for system in self.config.read('systems', []):
            if system['shortName'] == system_name:
                talkgroups = Talkgroups(system['talkgroupsFile'])
                return Talkgroup(talkgroups.name(talkgroup), talkgroup, talkgroups.streams(talkgroup))
        return Talkgroup('Unknown', talkgroup, [])

    def recording(self, file):
        file = os.path.abspath(file)
        return TrunkRecording(file, self.system_name(file), self.talkgroup(file))


class TrunkRecording(object):
    def __init__(self, file: str, system_name: str, talkgroup: Talkgroup):
        super(TrunkRecording, self).__init__()
        self.file = file
        self.system_name = system_name
        self.talkgroup = talkgroup

    def filename_without_extension(self):
        return os.path.splitext(self.file)[0]

    def __str__(self):
        return f"TrunkRecording : file={self.file} system_name={self.system_name} talkgroup={self.talkgroup}"

    # ref: https://k.ylo.ph/2016/04/04/loudnorm.html
    # ref: https://www.auphonic.com/blog/2013/01/07/loudness-targets-mobile-audio-podcasts-radio-tv/
    # ref: https://auphonic.com/blog/2012/08/02/loudness-measurement-and-normalization-ebu-r128-calm-act/
    def normalized(self, target_i=-16, target_lra=6.0, target_tp=-1.0):
        options = {
            "i": target_i,
            "tp": target_tp,
            "lra": target_lra,
            "dual_mono": "true",
            "print_format": "json"
        }
        command = ["ffmpeg", "-i", self.file, "-af", f"loudnorm={ffmpeg_filter_string(options)}", "-f", "null", "-"]
        logging.debug(' '.join(command))

        result = subprocess.run(command, capture_output=True, universal_newlines=True)
        lines = result.stderr.splitlines()
        for x in range(0, len(lines)):
            try:
                subset = ''.join(lines[x:])
                params = json.loads(subset)
                break
            except json.decoder.JSONDecodeError as e:
                pass

        options = {
            "i": target_i,
            "tp": target_tp,
            "lra": target_lra,
            "measured_i": params['output_i'],
            "measured_tp": params['output_tp'],
            "measured_lra": params['output_lra'],
            "measured_thresh": params['input_thresh'],
            "offset": 0,
            "linear": "true",
            "dual_mono": "true",
            "print_format": "summary",
        }
        output_file = f"{self.filename_without_extension()}-norm.wav"
        command = ["ffmpeg", "-i", self.file, "-metadata", f'title="{self.talkgroup.name}"', "-af", f"loudnorm={ffmpeg_filter_string(options)}", "-ar", "8k", "-y", output_file]
        logging.debug(' '.join(command))
        result = subprocess.run(command, capture_output=True, universal_newlines=True)

        return TrunkRecording(output_file, self.system_name, self.talkgroup)

    def encoded(self):
        output_file = f"{self.filename_without_extension()}.mp3"
        command = ["ffmpeg", "-i", self.file, "-codec:a", "libmp3lame", "-b:a", "96k", "-y", output_file]
        logging.debug(' '.join(command))
        result = subprocess.run(command, capture_output=True, universal_newlines=True)

        return TrunkRecording(output_file, self.system_name, self.talkgroup)


class Systems(object):

    def __init__(self, config: Configuration):
        super(System, self).__init__()
        self.config = config

    def talkgroups_for(self, short_name):
        for system in self.config.read('systems', []):
            if system['shortName'] == short_name:
                return Talkgroups(self.talkgroups_file)
        raise ValueError(f"No system found with the short_name {short_name}")


class Liquidsoap(object):

    def __init__(self, config: Configuration):
        super(Liquidsoap, self).__init__()
        self.socketDir = config.read('liquidsoap.socketDir', '/var/run/liquidsoap')

    def queue(self, file: TrunkRecording):
        for stream in file.talkgroup.streams:
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            address = os.path.join(self.socketDir, f"{stream}.sock")

            try:
                sock.connect(address)
            except socket.error as msg:
                continue

            messages = [
                f'queue.push annotate:title="{file.talkgroup.name}":{file.file}',
                'quit',
            ]
            for message in messages:
                sock.send((message + "\n").encode('utf-8'))
            logging.info(f"queued {file} to socket {address}")


config = Configuration()
trunk_recorder = TrunkRecorder(config)
liquidsoap = Liquidsoap(config)

# We don't encode here because liquidsoap will re-encode and that murders the quality
recording = trunk_recorder.recording(sys.argv[1]).normalized()

liquidsoap.queue(recording)

stream-name.liquidsoap

HOST="icecast-server.example.com"
PORT=8000
MOUNT="/stream-name"
PASSWORD="source-password-defined-in-icecast"
NAME="stream title"
STREAMID="stream-name matching what is used in the talkgroupsFile CSV file"
%include "common.liquidsoap"

what is the filename and format of the CSV?

@kb2ear it's been a while and I don't have this running at the moment (although it remains on the todo list) but a single line from the talk groups CSV looks like (there can be many of course)

1002,3ab,D,County Sheriff Tactical 1,Tactical 1,CSOT1,Police,9,cso|azleo

All of the CSV fields up to cso|azleo are standard. Or at least were. idk if they have changed their CSV format in recent versions of trunk-recorder.

The last field, cso|azleo is a | delimited list of liquidsoap streams that should receive the recordings. I refer to this as "stream id" in multiple places.

So if I had three liquidsoap streams cso, azleo, fire, then radio traffic from that channel would be sent to both cso and azleo inputs.

A very slimmed down version of my trunk-recorder config looks like,

{
    "sources": [ ... ],
    "systems": {
        {
            "control_channels": [
                123456789,
                123450987
            ],
            "type": "p25",
            "talkgroupsFile": "AZp25.csv",
            "shortName": "AZp25",
            "uploadScript": "queue-to-liquidsoap.py"
        },
    },
    <other settings>
    "liquidsoap": {
        "socketDir": "/tmp/sockets"
    }
}