Skip to content
GitLab
Commit d4767d38 authored by Matteo's avatar Matteo
Browse files

refactor to pythonic package

parent da5c2d12
Expand all Hide whitespace changes
Inline Side-by-side
.editorconfig 0 → 100644
View file @ d4767d38
# EditorConfig is awesome: https://EditorConfig.org
# top-most EditorConfig file
root = true
[*]
indent_style = space
indent_size = 4
end_of_line = lf
charset = utf-8
trim_trailing_whitespace = false
insert_final_newline = false
\ No newline at end of file
config/args.yaml → .env
View file @ d4767d38
# Working path
WORKING_PATH: "/data"
WORKING_PATH="../data"
# Name of the Preservation Audio File (without extension)
FILES_NAME: "sample12_W7N_R15C"
FILES_NAME="BERIO100"
# PRESERVATION_FILE_NAME: "BornToDie"
# PRESERVATION_FILE_NAME: ""
# Equalisation standard used to record the tape.
# Accepted values: ("NAB", "CCIR")
STANDARD_W: "NAB"
STANDARD_W="NAB"
# Recording tape speed [ips].
# Accepted values: (3.75, 7.5, 15, 30)
SPEED_W: 7.5
SPEED_W=15
# Equalization standard used to read the tape.
# Accepted values: ("NAB", "CCIR")
STANDARD_R: "CCIR"
STANDARD_R="NAB"
# Reading tape speed [ips].
# Accepted values: (3.75, 7.5, 15, 30)
SPEED_R: 15
\ No newline at end of file
SPEED_R=15
\ No newline at end of file
config/logger.yaml deleted 100644 → 0
View file @ da5c2d12
version: 1
formatters:
brief:
format: '%(message)s'
precise:
format: '[%(levelname)s %(name)s %(module)s:%(lineno)s - %(funcName)s() - %(asctime)s]\n\t %(message)s \n'
datefmt: '%d/%m/%Y %H:%M:%S'
handlers:
file:
class: logging.handlers.RotatingFileHandler
formatter: precise
filename: /var/log/tape-audio-restoration/server.log
maxBytes: 10485760
backupCount: 3
console:
class: logging.StreamHandler
formatter: brief
loggers:
uvicorn:
level: INFO
handlers: [file, console]
propagate: no
\ No newline at end of file
poetry.lock
View file @ d4767d38
This diff is collapsed.
pyproject.toml
View file @ d4767d38
[tool.poetry]
name = "tape-audio-restoration"
version = "1.0.1"
version = "1.1.0"
description = "MPAI CAE-ARP Tape Audio Restoration."
authors = ["Matteo Spanio <dev2@audioinnova.com>", "Nadir Dalla Pozza <nadir.dallapozza@unipd.it>"]
authors = [
"Matteo Spanio <dev2@audioinnova.com>",
"Nadir Dalla Pozza <nadir.dallapozza@unipd.it>"
]
license = "GPLv3"
readme = "README.md"
packages = [{ include = "tape_audio_restoration", from = "src" }]
[tool.poetry.dependencies]
python = ">=3.10,<3.12"
numpy = "1.23.3"
mpai-cae-arp = "^0.2.2"
control = "0.9.3.post2"
scipy = "^1.10.1"
fastapi = {extras = ["standard"], version = "^0.95.0"}
uvicorn = "^0.21.1"
mpai-cae-arp = "^0.3.2"
rich = "^13.3.4"
[build-system]
......@@ -21,4 +24,4 @@ requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
[tool.poetry.scripts]
tape-audio-restoration = 'tapeAudioRestoration:main'
\ No newline at end of file
tape-audio-restoration = 'tape_audio_restoration.cli:main'
\ No newline at end of file
src/server.py deleted 100644 → 0
View file @ da5c2d12
import subprocess
from fastapi import FastAPI, Response, status, Query
from typing import Annotated
from mpai_cae_arp.types.schema import Info
from mpai_cae_arp.files import FileType, File
info = File("config/server.yaml", FileType.YAML).get_content()
app = FastAPI(**info)
@app.get("/")
async def list_all_entrypoints() -> list[str]:
"""
Get the list of available main routes.
"""
return ["restore", "description", "docs"]
@app.get("/description")
async def get_server_info() -> Info:
"""
Retrieve all the informations about the software running on the server.
"""
return info
@app.get("/restore", status_code=200)
async def restore(
response: Response,
equalization_w: Annotated[str, Query(
description="Actual equalization of the audio tape applied during the recording.",
example="CCIR"
)],
equalization_r: Annotated[str, Query(
description="Actual equalization of the audio tape applied during the playback.",
example="NAB"
)],
speed_w: Annotated[str, Query(
description="Actual speed of the audio tape applied during the recording.",
example="7.5"
)],
speed_r: Annotated[str, Query(
description="Actual speed of the audio tape applied during the playback.",
example="15"
)],
files_name: Annotated[str, Query(
description="Name of the audio/video file to be analyzed without the extension.",
example="filename"
)],
working_path: Annotated[str, Query(
description="Path to the directory containing the preservation files.",
examples={
"Absolute path": {"value": "/home/user/preservation"},
"Relative path": {"value": "preservation"}
}
)] = None
) -> dict:
process = [
"python",
"src/tapeAudioRestoration.py",
"--files_name", files_name,
"--equalization-w", equalization_w,
"--equalization-r", equalization_r,
"--speed-w", speed_w,
"--speed-r", speed_r
]
if working_path is not None:
process.extend(["--working_path", working_path])
else:
config = File("config/args.yaml", FileType.YAML).get_content()
process.extend(["--working_path", config["WORKING_PATH"]])
cprocess = subprocess.run(process, capture_output=True)
if cprocess.returncode == 0:
editing_list = File("EditingList.json", FileType.JSON).get_content()
return editing_list
else:
response.status_code = status.HTTP_412_PRECONDITION_FAILED
return {"error": {
"returncode": cprocess.returncode,
"stdout": cprocess.stdout.decode("utf-8"),
"stderr": cprocess.stderr.decode("utf-8"),
}}
src/tape_audio_restoration/cli.py 0 → 100644
View file @ d4767d38
import os
import sys
from argparse import ArgumentParser, RawTextHelpFormatter
from typing import NoReturn
from rich.console import Console
from scipy.io import wavfile
from mpai_cae_arp.files import File, FileType
from mpai_cae_arp.io import Color, Style, pprint
from mpai_cae_arp.audio import AudioWave
from .lib import correction, get_correction_filter, save_file, check_input
def get_args() -> tuple[str, str, str, float, str, float]:
"""
Method to obtain arguments from environment variables or command line.
Default environment, ignored if a command line argument is passed.
Raises
------
ValueError
If no arguments are passed and at least one environment variable is not set.
Returns
-------
tuple[str, str, str, float, str, float]
tuple consisting of six variables:
* str specifying the working path;
* str specifying the name of the Preservation files, which is key element to retrieve necessary files;
* str specifying the equalization standard used when the tape was recorded;
* float specifying the speed used when the tape was recorded;
* str specifying the equalization standard used when the tape was read;
* float specifying the speed used when the tape was read.
"""
if len(sys.argv) > 1:
# Read from command line
parser = ArgumentParser(
prog="python3 tapeAudioRestoration.py",
formatter_class=RawTextHelpFormatter,
description="A tool that implements MPAI CAE-ARP Tape Audio Restoration Technical Specification.\n"
"By default, the configuration parameters are loaded from the environment,\n"
"but, alternately, you can pass command line arguments to replace them."
)
parser.add_argument(
"-w",
"--working-path",
help="Specify the Working Path, where all input files are stored",
required=True
)
parser.add_argument(
"-f",
"--files-name",
help="Specify the name of the Preservation files (without extension)",
required=True
)
parser.add_argument(
"-ew",
"--equalization-w",
help="Specify the name of the equalization standard used when the tape was recorded",
required=True
)
parser.add_argument(
"-sw",
"--speed-w",
help="Specify the speed used when the tape was recorded",
required=True
)
parser.add_argument(
"-er",
"--equalization-r",
help="Specify the name of the equalization standard used when the tape was read",
required=True
)
parser.add_argument(
"-sr",
"--speed-r",
help="Specify the speed used when the tape was read",
required=True
)
args = parser.parse_args()
working_path = args.working_path
files_name = args.files_name
standard_w = args.equalization_w
speed_w = args.speed_w
standard_r = args.equalization_r
speed_r = args.speed_r
else:
arg_names = ['WORKING_PATH', 'FILES_NAME', 'STANDARD_W', 'SPEED_W', 'STANDARD_R', 'SPEED_R']
config = {argument: os.getenv(argument) for argument in arg_names}
working_path, files_name, standard_w, speed_w, standard_r, speed_r = map(config.get, arg_names)
if any(value is None for value in config.values()):
raise ValueError("Please, set the environment variables: WORKING_PATH, FILES_NAME, STANDARD_W, SPEED_W, STANDARD_R, SPEED_R.")
return working_path, files_name, standard_w, float(speed_w), standard_r, float(speed_r)
def exit_with_error(error_message: str, error: int, console: Console) -> NoReturn:
console.print(f"[red bold]Error: {error_message}")
quit(error)
def main():
console = Console()
console.print("[bold]\nWelcome to ARP Tape Audio Restoration!")
console.print(f"You are using Python version: {sys.version}")
try:
working_path, files_name, standard_w, speed_w, standard_r, speed_r = get_args()
paf_path, temp_path, standard_w, standard_r = check_input(working_path, files_name, standard_w, speed_w, standard_r, speed_r)
# open the Preservation Audio file
with console.status("[bold]Opening Preservation Audio file...", spinner="dots"):
sample_rate, preservation_audio_file = wavfile.read(paf_path)
# get the correction filter
with console.status("[bold]Applying correction filter...", spinner="dots"):
a, b, new_sample_rate, _ = get_correction_filter(standard_w, speed_w, standard_r, speed_r, sample_rate)
restored_audio_file = preservation_audio_file
# if needed, apply the correction filter to the Preservation Audio file
if len(a) != 0:
restored_audio_file = correction(a, b, preservation_audio_file, round(new_sample_rate))
with console.status("[bold]Saving Restoration Audio file...", spinner="dots"):
save_file(restored_audio_file, round(new_sample_rate), temp_path, f"{files_name}.wav")
except ValueError:
exit_with_error(
"{}\n{}".format(
"Working directory or files name not specified!",
"Try -h/--help to know more about Tape Audio Restoration usage"),
os.EX_USAGE, console)
except Exception as e:
exit_with_error(str(e), os.EX_SOFTWARE, console)
console.print("[green] Success! :tada:")
quit(os.EX_OK)
if __name__ == '__main__':
main()
src/tapeAudioRestoration.py → src/tape_audio_restoration/lib.py
View file @ d4767d38
#!./venv/bin/activate
"""
MPAI CAE-ARP Tape Audio Restoration.
......@@ -14,109 +12,31 @@ import numpy as np
import os
import shutil
import sys
from argparse import ArgumentParser, RawTextHelpFormatter
from statistics import mode
from control import c2d, TransferFunction
from numpy import ndarray
from scipy.io import wavfile
from scipy.signal import tf2zpk, zpk2tf, lfilter
from mpai_cae_arp.io import Color, Style, pprint
from mpai_cae_arp.files import File, FileType
from mpai_cae_arp.io import Color, Style, pprint
from mpai_cae_arp.types.irregularity import IrregularityFile, Irregularity, IrregularityProperties
__author__ = "Nadir Dalla Pozza"
__copyright__ = "Copyright 2022, Audio Innova S.r.l."
__credits__ = ["Niccolò Pretto", "Nadir Dalla Pozza", "Sergio Canazza"]
__license__ = "GPL v3.0"
__version__ = "1.0.1"
__maintainer__ = "Nadir Dalla Pozza"
__email__ = "nadir.dallapozza@unipd.it"
__status__ = "Production"
def get_arguments() -> tuple[str, str, str, float, str, float]:
"""
Method to obtain arguments from config/args.yaml file or command line.
Default config/args.yaml, ignored if a command line argument is passed.
:return: tuple consisting of nine variables:
1) str specifying the working path;
2) str specifying the name of the Preservation files, which is key element to retrieve necessary files;
3) str specifying the equalization standard used when the tape was recorded;
4) float specifying the speed used when the tape was recorded;
5) str specifying the equalization standard used when the tape was read;
6) float specifying the speed used when the tape was read.
"""
def verify_path(working_path: str, files_name: str) -> str:
raise NotImplementedError
if len(sys.argv) > 1:
# Read from command line
parser = ArgumentParser(
prog="python3 tapeAudioRestoration.py",
formatter_class=RawTextHelpFormatter,
description="A tool that implements MPAI CAE-ARP Tape Audio Restoration Technical Specification.\n"
"By default, the configuration parameters are loaded from ./config/args.yaml file,\n"
"but, alternately, you can pass command line arguments to replace them."
)
parser.add_argument(
"-w",
"--working-path",
help="Specify the Working Path, where all input files are stored",
required=True
)
parser.add_argument(
"-f",
"--files-name",
help="Specify the name of the Preservation files (without extension)",
required=True
)
parser.add_argument(
"-ew",
"--equalization-w",
help="Specify the name of the equalization standard used when the tape was recorded",
required=True
)
parser.add_argument(
"-sw",
"--speed-w",
help="Specify the speed used when the tape was recorded",
required=True
)
parser.add_argument(
"-er",
"--equalization-r",
help="Specify the name of the equalization standard used when the tape was read",
required=True
)
parser.add_argument(
"-sr",
"--speed-r",
help="Specify the speed used when the tape was read",
required=True
)
args = parser.parse_args()
working_path = args.working_path
files_name = args.files_name
standard_w = args.equalization_w
speed_w = float(args.speed_w)
standard_r = args.equalization_r
speed_r = float(args.speed_r)
else:
# Read configuration file
arg_names = ['WORKING_PATH', 'FILES_NAME', 'STANDARD_W', 'SPEED_W', 'STANDARD_R', 'SPEED_R']
config = {}
try:
config = File('config/args.yaml', FileType.YAML).get_content()
for argument in arg_names:
if argument not in config:
pprint(f'{argument} key not found in config/args.yaml!', color=Color.RED)
quit(os.EX_CONFIG)
except FileNotFoundError:
pprint('config/args.yaml file not found!', color=Color.RED)
quit(os.EX_NOINPUT)
# unpack config dictionary
working_path, files_name, standard_w, speed_w, standard_r, speed_r = map(config.get, arg_names)
return working_path, files_name, standard_w, speed_w, standard_r, speed_r
def verify_equalization_and_speed(standard_w: str, speed_w: float, standard_r: str, speed_r: float) -> tuple[str, str]:
"""
Method to verify that the equalization and speed parameters are correct.
"""
raise NotImplementedError
def check_input(working_path: str, files_name: str, standard_w: str, speed_w: float, standard_r: str, speed_r: float) -> tuple[str, str, str, str]:
......@@ -221,16 +141,29 @@ def check_input(working_path: str, files_name: str, standard_w: str, speed_w: fl
def get_correction_filter(standard_w: str, speed_w: float, standard_r: str, speed_r: float, fs: int) -> tuple[list, list, float, int]:
"""
Method to establish correct filter transfer function coefficients;
:param standard_w: str specifying the equalization standard used when the tape was recorded,
:param speed_w: float specifying the speed used when the tape was recorded,
:param standard_r: str specifying the equalization standard used when the tape was read,
:param speed_r: float specifying the speed used when the tape was read,
:param fs: float specifying the sampling frequency.
:return: tuple consisting of four variables:
1) array representing the filter numerator coefficients;
2) array representing the filter denominator coefficients;
3) float specifying the operating sampling frequency;
4) int informing about the case number.
Parameters
----------
standard_w : str
String specifying the equalization standard used when the tape was recorded.
speed_w : float
Float specifying the speed used when the tape was recorded.
standard_r : str
String specifying the equalization standard used when the tape was read.
speed_r : float
Float specifying the speed used when the tape was read.
fs : int
Integer specifying the sampling frequency.
Returns
-------
tuple
Tuple consisting of four variables:
* list of the filter numerator coefficients;
* list of the filter denominator coefficients;
* float specifying the operating sampling frequency;
* int informing about the case number.
"""
# CCIR time constants.
t2_30 = 17.5 * 10 ** (-6) # time constant CCIR_30
......@@ -296,7 +229,7 @@ def get_correction_filter(standard_w: str, speed_w: float, standard_r: str, spee
# Case 31
if speed_r == 30:
print('Reference case: 31')
print(CC.GREEN + 'Nothing to do!' + CC.END)
pprint('Nothing to do!', color=Color.GREEN)
quit(os.EX_OK)
# Case 15
elif speed_r == 15:
......@@ -658,11 +591,22 @@ def get_correction_filter(standard_w: str, speed_w: float, standard_r: str, spee
def correction(a: array, b: array, paf: ndarray, fs: int) -> ndarray:
"""
Apply a correction filter to a Preservation Audio File;
:param a: array of coefficients, specifying the numerator of filter transfer function,
:param b: array of coefficients, specifying in the denominator of filter transfer function,
:param paf: ndarray specifying the raw audio data of the Preservation Audio File,
:param fs: int specifying the operational sampling frequency.
:return: the corrected audio as a Restored Audio File.
Parameters
----------
a : array
array of coefficients, specifying the numerator of filter transfer function,
b : array
array of coefficients, specifying in the denominator of filter transfer function,
paf : ndarray
specifying the raw audio data of the Preservation Audio File,
fs : int
specifying the operational sampling frequency.
Returns
-------
ndarray
the corrected audio as a Restored Audio File.
"""
# Analog transfer function
......@@ -707,11 +651,21 @@ def correction(a: array, b: array, paf: ndarray, fs: int) -> ndarray:
def save_file(file: ndarray, fs: int, temp_path: str, name: str):
"""
Save an audio file to the given path with name 1.wav;
:param file: ndarray specifying the raw audio data,
:param fs: int specifying the operational sampling frequency,
:param temp_path: str specifying the path where the file will be saved,
:param name: str specifying the file name.
:return: exit codes corresponding to the execution status.
Parameters
----------
file : ndarray
Array containing the raw audio data.
fs : int
Sampling frequency.
temp_path : str
Path where the file will be saved.
name : str
File name.
Returns
-------
exit codes corresponding to the execution status.
"""
raf_path = os.path.join(temp_path, 'RestoredAudioFiles')
make_raf = False
......@@ -737,55 +691,100 @@ def save_file(file: ndarray, fs: int, temp_path: str, name: str):
wavfile.write(os.path.join(raf_path, name + '.wav'), fs, file)
def main():
"""
Main execution method.
:return: exit codes corresponding to the execution status.
def read_params_from_irregularity_file(irregularity_file_path: str) -> tuple[str, float, str, float]:
"""Get the parameters needed to apply the correction filter from the specified irregularity file.
Parameters
----------
irregularity_file_path : str
Path to the irregularity file.
Returns
-------
tuple[str, float, str, float]
Tuple containing the parameters needed to apply the correction filter.
It is assumed that the audio file considered contains only one recording session (so only one speed and equalisation).
Since it is only one recording session, the possible correction to be applied is only one, this fact allow us to
consider the mode of the equalisation and speed detected as the correct parameters.
If there are no irregularities properties in the irregularity file, the parameters default to:
- reading equalisation: NAB
- reading speed: 7.5
- writing equalisation: NAB
- writing speed: 7.5
"""
pprint("\nWelcome to ARP Tape Audio Restoration!", styles=[Style.BOLD])
print("You are using Python version: " + sys.version)
# Get the input from config/args.yaml or command line
working_path, files_name, standard_w, speed_w, standard_r, speed_r = get_arguments()
# Check if input is correct
paf_path, temp_path, standard_w, standard_r = check_input(working_path, files_name, standard_w, speed_w, standard_r, speed_r)
# Display input parameters
print('\nInput parameters:')
print(' WORKING_PATH: ' + working_path)
print(' FILES_NAME: ' + files_name)
print(' STANDARD_W: ' + standard_w)
print(' SPEED_W: ' + str(speed_w) + ' ips')
print(' STANDARD_R: ' + standard_r)
print(' SPEED_R: ' + str(speed_r) + ' ips')
# Preservation Audio File check
print("Opening '%s'..." % paf_path)
fs, paf = wavfile.read(paf_path)
print('Preservation Audio File opened!')
print(' FS: ' + str(fs) + ' Hz\n')
# Decision stage
a, b, fs, case = get_correction_filter(standard_w, speed_w, standard_r, speed_r, fs)
# Casting FS to int because wavfile.write() is stupid
fs = round(fs)
print('Reference case: ' + str(case))
print('Operational FS: ' + str(fs) + ' Hz.')
# Correction phase
if len(a) != 0:
# Not all cases present a correction filter!
raf = correction(a, b, paf, fs)
save_file(raf, fs, temp_path, '1')
else:
# Just save Restored Audio File, but with modified fs
save_file(paf, fs, temp_path, '1')
# End
pprint("Success!\n", color=Color.GREEN, styles=[Style.BOLD])
if __name__ == '__main__':
main()
json_content = File(irregularity_file_path, FileType.JSON).get_content()
irregularity_file = IrregularityFile.from_json(json_content)
write_speeds = []
read_speeds = []
write_eq = []
read_eq = []
for irr in irregularity_file.irregularities:
if irr.irregularity_properties is not None:
read_eq.append(irr.irregularity_properties.reading_equalisation)
read_speeds.append(irr.irregularity_properties.reading_speed)
write_eq.append(irr.irregularity_properties.writing_equalisation)
write_speeds.append(irr.irregularity_properties.writing_speed)
else:
read_eq.append(None)
read_speeds.append(None)
write_eq.append(None)
write_speeds.append(None)
result_ws, result_rs, result_we, result_re = mode(write_speeds), mode(read_speeds), mode(write_eq), mode(read_eq)
if all(x is None for x in [result_rs, result_ws]) or len(write_speeds) == 0:
result_ws, result_rs = 7.5, 7.5
if all(x is None for x in [result_we, result_re]) or len(write_eq) == 0:
result_we, result_re = 'NAB', 'NAB'
return result_we, result_ws, result_re, result_rs
# def main():
# """
# Main execution method.
# :return: exit codes corresponding to the execution status.
# """
# pprint("\nWelcome to ARP Tape Audio Restoration!", styles=[Style.BOLD])
# print("You are using Python version: " + sys.version)
# # Get the input from config/args.yaml or command line
# working_path, files_name, standard_w, speed_w, standard_r, speed_r = get_args()
# # Check if input is correct
# paf_path, temp_path, standard_w, standard_r = check_input(working_path, files_name, standard_w, speed_w, standard_r, speed_r)
# # Display input parameters
# print('\nInput parameters:')
# print(' WORKING_PATH: ' + working_path)
# print(' FILES_NAME: ' + files_name)
# print(' STANDARD_W: ' + standard_w)
# print(' SPEED_W: ' + str(speed_w) + ' ips')
# print(' STANDARD_R: ' + standard_r)
# print(' SPEED_R: ' + str(speed_r) + ' ips')
# # Preservation Audio File check
# print("Opening '%s'..." % paf_path)
# fs, paf = wavfile.read(paf_path)
# print('Preservation Audio File opened!')
# print(' FS: ' + str(fs) + ' Hz\n')
# # Decision stage
# a, b, fs, case = get_correction_filter(standard_w, speed_w, standard_r, speed_r, fs)
# # Casting FS to int because wavfile.write() is stupid
# fs = round(fs)
# print('Reference case: ' + str(case))
# print('Operational FS: ' + str(fs) + ' Hz.')
# # Correction phase
# if len(a) != 0:
# # Not all cases present a correction filter!
# raf = correction(a, b, paf, fs)
# save_file(raf, fs, temp_path, '1')
# else:
# # Just save Restored Audio File, but with modified fs
# save_file(paf, fs, temp_path, '1')
# # End
# pprint("Success!\n", color=Color.GREEN, styles=[Style.BOLD])
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment