# Libreprinter is a software allowing to use the Centronics and serial printing
# functions of vintage computers on modern equipement through a tiny hardware
# interface.
# Copyright (C) 2020-2024 Ysard
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Gestion of files (creation and processing) and directories used by the project"""
# Standard imports
import os
import shutil
from pathlib import Path
# Custom imports
from libreprinter.commons import OUTPUT_DIRS, SHARED_MEM_NAME
[docs]
def init_directories(output_path, output_dirs=OUTPUT_DIRS):
"""Init default output directories (raw, png, pdf, txt, pcl)
:param output_path: Path were directories will be created.
:param output_dirs: Override default output directories. Useful if callled
by plugins during their initialisation.
:type output_path: str
:type output_dirs: list[str]
"""
# test if output path exists
if not os.path.isdir(output_path):
os.mkdir(output_path)
for directory in output_dirs:
try:
os.mkdir(output_path + directory, mode=0o755)
except FileExistsError:
pass
[docs]
def cleanup_directories(output_path):
"""Delete all directories initialised by this project in the given path
Deletes also /dev/shm/ shared memory.
:param output_path: Path were directories will be created.
:type output_path: str
"""
for directory in OUTPUT_DIRS:
try:
shutil.rmtree(output_path + directory)
except FileNotFoundError:
pass
# Clean shared memory
try:
os.remove("/dev/shm/" + SHARED_MEM_NAME)
except FileNotFoundError:
pass
[docs]
def get_job_number(
output_path, extensions=(".txt", ".raw", ".eps", ".csv", ".hpgl", ".ps")
):
"""Return the number of the current job based on files found in output directories
- Get only non empty files
- Get only expected extensions (txt, raw, eps)
- Search highest file number in all folders
TODO: handle properly pdf dir data: page1-1.pdf, page1-2.pdf, page2-1.pdf...
TODO: handle new naming convention based on the date to avoid overwriting
if raw files are deleted.
=> will allow to only search raw files
:return: Current job number
:rtype: int
"""
g = Path(output_path).rglob("*.*")
# Prune empty files, get basenames and filter extensions
pruned = [
filepath.stem for filepath in g
if filepath.suffix in extensions # Filter extensions
and filepath.stem.isnumeric() # Get only files named with digits
and filepath.stat().st_size # Get non empty files
]
# Get the higher name
pruned = sorted(map(int, pruned))
# Increment the value to get the current job number
return int(pruned[-1]) + 1 if pruned else 1
[docs]
def convert_data_line_ending(in_data, line_ending):
r"""Replace line endings of in_data and return the result
- unix => windows: `\\n => \\r\\n`
- windows => unix: `\\r\\n => \\n`
:param in_data: Data to be processed
:param line_ending: Destination line ending.
:type in_data: bytes
:type line_ending: bytes
:return: Processed data
:rtype: bytes
"""
conv_mapping = {
b"\n": b"\r\n",
b"\r\n": b"\n",
}
return in_data.replace(conv_mapping[line_ending], line_ending)
[docs]
def convert_file_line_ending(in_file, out_file, line_ending):
"""Replace line endings of in_file and put data in out_file
.. seealso:: :meth:`convert_data_line_ending`
:param in_file: Path of processed but not modified file.
:param out_file: Path of result file.
:param line_ending: Destination line ending
:type line_ending: str
"""
# Convert to bytes for binary files
line_ending = line_ending.encode()
with open(in_file, "rb") as in_file_fd, open(out_file, "wb") as out_file_fd:
out_file_fd.write(convert_data_line_ending(in_file_fd.read(), line_ending))