Python

Python bits.

alsa.py

class sndid.alsa.ArgsModel(*, ip: str = '127.0.0.1', port: int = 9988, rate: int = 48000, time: int = 10)

Bases: BaseModel

Model for command-line arguments

ip

IP address to stream to. Default is 127.0.0.1.

Type:

str

port

Port to stream to. Default is 9988.

Type:

int

rate

Sample rate. Default is 48000.

Type:

int

time

Duration to stream for. Default is 10.

Type:

int

Validators:

validate_int_values: Validates that integer values are positive.

ip: str

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'ip': FieldInfo(annotation=str, required=False, default='127.0.0.1'), 'port': FieldInfo(annotation=int, required=False, default=9988), 'rate': FieldInfo(annotation=int, required=False, default=48000), 'time': FieldInfo(annotation=int, required=False, default=10)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

port: int

rate: int

time: int

classmethod validate_int_values(v)

sndid.alsa.main()

Main function to parse arguments and run ALSA stream

sndid.alsa.parse_args() → ArgsModel

Parse command-line arguments and return ArgsModel instance

sndid.alsa.run_alsa_stream(ip: str, port: int, rate: int, time: int)

Run ALSA stream to specified IP address and port with given sample rate and duration

Parameters:

• ip (str) – IP address to stream to.

• port (int) – Port to stream to.

• rate (int) – Sample rate.

• time (int) – Duration to stream for.

The run_alsa_stream function takes in the IP address, port, sample rate, and duration as arguments and runs an ALSA stream to the specified IP address and port with the given sample rate and duration.

The function constructs a command that uses arecord to record audio input and nc (netcat) to send the audio data to the specified IP address and port. The command is run using the subprocess.run function.

analyzerwrapper.py

class sndid.analyzerwrapper.AnalyzerWrapper(config: Dict[str, int | float])

Bases: object

AnalyzerWrapper class to interface with the BirdNet library.

analyze(input_index: int, segment: Any) → None

Analyze an audio segment.

Parameters:

• input_index – Index of the input audio segment

• segment – Audio segment to analyze

Returns:

None

class sndid.analyzerwrapper.Config(*, sample_rate: int, lat: float, lon: float, min_confidence: float)

Bases: BaseModel

Configuration for the AnalyzerWrapper.

lat: float

lon: float

min_confidence: float

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'lat': FieldInfo(annotation=float, required=True, metadata=[Ge(ge=-90), Le(le=90)]), 'lon': FieldInfo(annotation=float, required=True, metadata=[Ge(ge=-180), Le(le=180)]), 'min_confidence': FieldInfo(annotation=float, required=True, metadata=[Ge(ge=0), Le(le=1)]), 'sample_rate': FieldInfo(annotation=int, required=True, metadata=[Gt(gt=0)])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

classmethod not_negative(v)

Validator to ensure the value is not negative.

Parameters:

v – The value to validate

Returns:

The validated value

Raises:

ValueError – If the value is negative

sample_rate: int

audiodetectionwriter.py

class sndid.audiodetectionwriter.AudioDetectionWriter(detections_file: str, detections_dir: str, max_queue_size: int = 1000)

Bases: object

Class for writing audio detections to file.

This class writes audio detections to file using a separate writer thread. It uses a queue to buffer the detection data and writes the data to file in the background. This allows for efficient handling of detection data while avoiding blocking the main thread.

detections_writer

The detections writer object that handles writing the detection data to file.

Type:

DetectionsWriter

detection_queue

The queue used to buffer the detection data.

Type:

queue.Queue

_stop_event

The stop event used to signal the writer thread to stop.

Type:

threading.Event

flush()

Flush the detection data to file.

This method flushes any remaining detection data in the queue to file.

stop()

Stop the AudioDetectionWriter object.

This method stops the writer thread and cleans up the object.

write_detections(recording_data: Any, detections: Any, detection_timestamp: float, input_index: int)

Write detection data to the detection queue.

This method writes the detection data to the detection queue, which is then processed by the writer thread.

Parameters:

• recording_data (Any) – The recording data associated with the detection.

• detections (Any) – The detection data.

• detection_timestamp (float) – The timestamp of the detection.

• input_index (int) – The index of the input associated with the detection.

audiowriter.py

class sndid.audiowriter.AudioWriter(detections_dir)

Bases: object

Class responsible for writing audio detections to disk.

Parameters:

detections_dir (str) – Directory to save audio detections.

detections_dir

Directory to save audio detections.

Type:

str

write_audio(recording_data, detections, detection_timestamp, input_index)

Write audio detections to disk.

Parameters:

• recording_data (np.ndarray) – The audio data to write.

• detections (List[Dict[str, Union[str, float]]]) – List of detections, where each detection is a dictionary containing start_time, end_time, and scientific_name keys.

• detection_timestamp (datetime) – The timestamp of the detection.

• input_index (int) – Index of the input in the multiplexed recording.

client.py

class sndid.client.ClientArguments(*, ip: str = '127.0.0.1', port: int = 9988, file: str = 'samples/mono.wav')

Bases: BaseModel

A class representing the command-line arguments for the sndid-client.

ip

The IP address of the server to connect to. Default is 127.0.0.1.

Type:

str

port

The port number to connect to on the server. Default is 9988.

Type:

int

file

The file to send to the server. Default is samples/mono.wav.

Type:

str

file: str

ip: str

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'file': FieldInfo(annotation=str, required=False, default='samples/mono.wav'), 'ip': FieldInfo(annotation=str, required=False, default='127.0.0.1'), 'port': FieldInfo(annotation=int, required=False, default=9988)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

port: int

sndid.client.main()

Run the sndid-client.

This function parses the command-line arguments, constructs a command to send the specified file to the server using netcat, and then executes that command.

sndid.client.parse_arguments()

Parse the command-line arguments for the sndid-client.

Returns:

An instance of ClientArguments with the parsed arguments.

Return type:

ClientArguments

config.py

class sndid.config.Config

Bases: object

static load_config(file_path=None)

Load the configuration file.

Parameters:

file (str) – Configuration file name. Defaults to “sndid.conf”.

Returns:

Configuration data.

Return type:

dict

static set_defaults(config)

Set default values for configuration.

Parameters:

config (dict) – Configuration data.

Returns:

Configuration data with default values.

Return type:

dict

detectionswriter.py

class sndid.detectionswriter.DetectionsWriter(detections_file, detections_dir)

Bases: object

A class for writing detection data to file.

This class provides a way to write detection data to both a text file and an audio file. It uses a queue to handle the detection data and writes the data in a separate thread.

stop()

Stop the DetectionsWriter thread.

write_audio(recording_data, detections, detection_timestamp, input_index)

Write audio data to file.

Parameters:

• recording_data – The audio data to write.

• detections – The detection data associated with the audio.

• detection_timestamp – The timestamp of the detection.

• input_index – The index of the input.

write_detections(detections, detection_timestamp, input_index)

Add detection data to the queue.

Parameters:

• detections – A list of detections.

• detection_timestamp – The timestamp of the detection.

• input_index – The index of the input.

filecli.py

class sndid.filecli.Args(*, c: float = 0.3, i: str = 'samples/sample.wav', a: str = '40.57', o: str = '-105.23', y: int = 2024, m: int = 6, d: int = 14, f: str = 'CSV')

Bases: BaseModel

Args class representing the command line arguments.

confidence

Minimum Confidence (default: 0.3).

Type:

float

input

Input filename to process (default: samples/sample.wav).

Type:

str

latitude

Latitude (default: 40.57).

Type:

str

longitude

Longitude (default: -105.23).

Type:

str

year

Year (default: today’s year).

Type:

int

month

Month (default: today’s month).

Type:

int

day

Day (default: today’s day).

Type:

int

format

Print output in CSV or JSON (default CSV).

Type:

str

confidence: float

day: int

format: str

input: str

latitude: str

longitude: str

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'confidence': FieldInfo(annotation=float, required=False, default=0.3, alias='c', alias_priority=2, title='confidence', description='Minimum Confidence (default: 0.3)'), 'day': FieldInfo(annotation=int, required=False, default=14, alias='d', alias_priority=2, title='day', description="Day (default: today's day)"), 'format': FieldInfo(annotation=str, required=False, default='CSV', alias='f', alias_priority=2, title='format', description='Print output in CSV or JSON (default CSV)'), 'input': FieldInfo(annotation=str, required=False, default='samples/sample.wav', alias='i', alias_priority=2, title='input', description='Input filename to process (default: samples/sample.wav)'), 'latitude': FieldInfo(annotation=str, required=False, default='40.57', alias='a', alias_priority=2, title='latitude', description='Latitude (default: 40.57)'), 'longitude': FieldInfo(annotation=str, required=False, default='-105.23', alias='o', alias_priority=2, title='longitude', description='Longitude (default: -105.23)'), 'month': FieldInfo(annotation=int, required=False, default=6, alias='m', alias_priority=2, title='month', description="Month (default: today's month)"), 'year': FieldInfo(annotation=int, required=False, default=2024, alias='y', alias_priority=2, title='year', description="Year (default: today's year)")}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

month: int

year: int

sndid.filecli.main()

main function to process the bird sound file.

sndid.filecli.parse_args() → Args

parse_args function to parse the command line arguments.

Returns:

The parsed command line arguments.

Return type:

Args

jackcli.py

sndid.jackcli.main()

Main function for sndid-jack.

Parses command line arguments, sets up configuration, initializes logging, and starts the main SndIdJack process.

sndid.jackcli.parse_arguments()

Parse command line arguments for sndid-jack.

Returns:

Parsed arguments.

Return type:

argparse.Namespace

jack_lazy_loader.py

sndid.jack_lazy_loader.jack_lazy_load(module_name)

Imports a module lazily, caching the result for future use.

Parameters:

module_name (str) – The name of the module to import.

Returns:

The imported module.

Return type:

any

jackmain.py

list.py

sndid.list.format_species_list(species_list)

Format the list of species for output.

Parameters:

species_list (list) – A list of species.

Returns:

A formatted list of species.

Return type:

list

sndid.list.get_species_list(args)

Get the list of species based on the provided arguments.

Parameters:

args (argparse.Namespace) – Parsed command line arguments.

Returns:

A list of species.

Return type:

list

sndid.list.main()

Main function to run sndid-list.

sndid.list.parse_arguments()

Parse command line arguments for sndid-list.

Returns:

Parsed arguments.

Return type:

argparse.Namespace

segment_analyzer.py

class sndid.segment_analyzer.SegmentAnalyzer(analyzer_wrapper)

Bases: Thread

SegmentAnalyzer is a class that analyzes a given segment of data and returns any detections found within that segment. It does this by using an analyzer wrapper to analyze the segment and return any detections.

analyze(input_index, segment)

Analyzes the given segment of data and returns any detections found within that segment.

Parameters:

• input_index (int) – The index of the input data being analyzed.

• segment (object) – The segment of data to analyze.

Returns:

A tuple containing the analyzed segment, any detections found, and the timestamp of the detection. If no detections were found, this method returns None.

Return type:

tuple

server.py

class sndid.server.TCPHandler(*args, params: TCPHandlerParams, **kwargs)

Bases: StreamRequestHandler

Handle TCP requests.

Parameters:

params (TCPHandlerParams) – Parameters for handling request.

handle()

Handle the client request.

class sndid.server.TCPHandlerParams(*, lat: float, lon: float, year: int, month: int, day: int, confidence: float)

Bases: BaseModel

Params for TCPHandler.

lat

Latitude.

Type:

float

lon

Longitude.

Type:

float

year

Year.

Type:

int

month

Month.

Type:

int

day

Day.

Type:

int

confidence

Confidence.

Type:

float

Validation: - Day must be between 1 and 31. - Month must be between 1 and 12. - Confidence must be between 0 and 1.

confidence: float

day: int

lat: float

lon: float

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'confidence': FieldInfo(annotation=float, required=True), 'day': FieldInfo(annotation=int, required=True), 'lat': FieldInfo(annotation=float, required=True), 'lon': FieldInfo(annotation=float, required=True), 'month': FieldInfo(annotation=int, required=True), 'year': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

month: int

classmethod validate_confidence(v)

Validate confidence is between 0 and 1.

classmethod validate_day(v)

Validate day is between 1 and 31.

classmethod validate_month(v)

Validate month is between 1 and 12.

year: int

sndid.server.initialize_logging()

Initialize logging.

sndid.server.main()

sndid.server.start_server()

Start the server.

sndidjack.py

class sndid.sndidjack.SndIdJack(config)

Bases: object

Process audio and identify species using SndIdJack.

config

Configuration data.

Type:

dict

detections_writer

Object for writing detections.

Type:

TextWriter

audio_segmenter

Object for segmenting audio.

Type:

Segmenter

audio_processor

Object for processing audio.

Type:

AudioProcessor

species_identifier

Object for identifying species.

Type:

SpeciesIdentifier

analyze_segment(input_index, segment)

Analyze a segment of audio data for input_index.

Parameters:

• input_index (int) – Index of the input.

• segment (numpy.ndarray) – The audio segment to analyze.

Returns:

If detections are made, return a tuple of the segment, detections, and detection_timestamp. Otherwise, return None.

Return type:

tuple or None

connect_client()

Connect client inports to their respective sources.

main()

Main process for SndIdJack.

process(frames)

Process audio frames and segment data.

Parameters:

frames (numpy.ndarray) – Audio frames to process.

register_client()

Register client inports.

segment_data(input_index)

Segment data for further processing.

Parameters:

input_index (int) – Index of the input.

shutdown(status, reason)

Shutdown callback function.

Parameters:

• status (int) – JACK status code.

• reason (str) – JACK shutdown reason.

stream.py

sndid.stream.main()

Runs the sndid-stream command line interface.

sndid.stream.run_ffmpeg_command(ip, port, time, url)

Runs the ffmpeg command with subprocess, handling piping correctly.

Parameters:

• ip (str) – Server IP address. Default is “127.0.0.1”.

• port (int) – Server network port. Default is 9988.

• time (int) – Length of segments in seconds. Default is 60.

• url (str) – Input url. Required.

Raises:

CalledProcessError – If the ffmpeg process returns a non-zero exit code.

sndid.stream.signal_handler(signum, frame)

Handles the SIGINT signal by printing a message and exiting the program.

textdetectionwriter.py

class sndid.textdetectionwriter.TextDetectionWriter(detections_file, detections_dir)

Bases: object

A class for writing text detections to a file.

This class uses a separate thread to periodically check a queue for new text detections and write them to a file. This allows the main program to continue running without being blocked by the time-consuming process of writing detections to a file.

Parameters:

• detections_file (str) – The file to which text detections will be written.

• detections_dir (str) – The directory in which the detections file is located.

text_writer

An instance of the TextWriter class, which handles the actual writing of text detections to a file.

Type:

TextWriter

detection_queue

A queue for storing text detections waiting to be written to a file.

Type:

queue.Queue

_stop_event

An event that can be set to stop the writer thread.

Type:

threading.Event

_writer_thread

A thread for periodically checking the detection queue and writing detections to a file.

Type:

threading.Thread

stop()

Stop the writer thread and join it.

This method sets the stop event, causing the writer thread to exit its loop, and then joins the thread to wait for it to finish.

write_detections(detections, detection_timestamp, input_index)

Write text detections to the queue.

This method writes the given text detections to the detection queue, allowing them to be written to a file by the writer thread.

Parameters:

• detections (list) – A list of text detections, where each detection is a dictionary with keys ‘text’, ‘confidence’, and ‘bounding_box’.

• detection_timestamp (float) – The timestamp associated with the text detections.

• input_index (int) – The index of the input from which the text detections were obtained.

textwriter.py

class sndid.textwriter.TextWriter(detections_file)

Bases: object

Class for writing detections to a text file.

detections_file

File for storing detections.

Type:

str

write_detections(detections, detection_timestamp, input_index)

Write detections to a text file.

Parameters:

• detections (list) – List of detections.

• detection_timestamp (datetime) – Timestamp of detections.

• input_index (int) – Index of input.