"""
Managing existing history files included within mapset
(C) 2023 by the GRASS Development Team
This program is free software under the GNU General Public
License (>=v2). Read the file COPYING that comes with GRASS
for details.
.. sectionauthor:: Linda Karlovska (Kladivova) linda.karlovska@seznam.cz
"""
import json
import shutil
from enum import Enum
from pathlib import Path
from datetime import datetime
import grass.script as gs
[docs]class Status(Enum):
"""Enum representing a set of status constants
that are used to represent various states or command outcomes."""
ABORTED = "aborted"
FAILED = "failed"
RUNNING = "running"
SUCCESS = "success"
UNKNOWN = "unknown"
[docs]def get_current_mapset_gui_history_path():
"""Return path to the current mapset history file.
This function does not ensure that the file exists.
"""
env = gs.gisenv()
base_path = Path(env["GISDBASE"]) / env["LOCATION_NAME"] / env["MAPSET"]
history_filename = ".wxgui_history"
txt_path = base_path / history_filename
json_path = base_path / (history_filename + ".json")
# Return path txt only if it exists and json does not
return txt_path if txt_path.exists() and not json_path.exists() else json_path
[docs]def get_history_file_extension(history_path):
"""Get extension of the history file.
The standard long-term format is plain text.
Newly it is stored as JSON-formatted.
:param str history_path: path to the history log file
:return str extension: None (plain text) or .json
"""
file_path = Path(history_path)
extension = file_path.suffix
return extension
[docs]def ensure_history_file(history_path):
"""Set up a new GUI history file if it doesn't exist.
:param str history_path: path to the history file
"""
if not history_path.exists():
try:
with open(history_path, encoding="utf-8", mode="w"):
pass
except OSError as e:
raise OSError(
_("Unable to create history file {}").format(history_path)
) from e
def _read_from_plain_text(history_path):
"""Read content of plain text history file.
:param str history_path: path to the history log file
:return content_list: list of dictionaries
with 'command' and 'command_info' keys
'command_info' is always empty since plain text history file
stores only executed commands."""
content_list = list()
try:
with open(
history_path, encoding="utf-8", mode="r", errors="replace"
) as file_history:
content_list = [
{"command": line.strip(), "command_info": None}
for line in file_history.readlines()
]
except OSError as e:
raise OSError(
_("Unable to read from plain text history file {}").format(history_path)
) from e
return content_list
def _read_from_JSON(history_path):
"""Read content of JSON history file.
:param str history_path: path to the history log file
:return content_list: list of dictionaries
with 'command' and 'command_info' keys
"""
content_list = list()
try:
with open(
history_path, encoding="utf-8", mode="r", errors="replace"
) as file_history:
content = file_history.read()
if content:
try:
history_entries = json.loads(content)
except ValueError as ve:
raise ValueError(
_("Error decoding content of JSON history file {}").format(
history_path
)
) from ve
# Process the content as a list of dictionaries
content_list = [
{
"command": entry["command"],
"command_info": entry["command_info"],
}
for entry in history_entries
]
except OSError as e:
raise OSError(
_("Unable to read from JSON history file {}").format(history_path)
) from e
return content_list
[docs]def read(history_path):
"""Read the content of the history file.
:param str history_path: path to the history log file
:return content_list: list of dictionaries
with 'command' and 'command_info' keys
"""
if get_history_file_extension(history_path) == ".json":
return _read_from_JSON(history_path)
return _read_from_plain_text(history_path)
[docs]def filter(json_data, command, timestamp):
"""
Filter JSON history file based on provided command and the time of command launch.
:param json_data: List of dictionaries representing JSON entries
:param command: First filtering argument representing command as string
:param timestamp: Second filtering argument representing the time of command launch
:return: Index of entry matching the filter criteria.
"""
for index, entry in enumerate(json_data):
if entry["command_info"]:
if (
entry["command"] == command
and entry["command_info"]["timestamp"] == timestamp
):
return index
return None
def _remove_entry_from_plain_text(history_path, index):
"""Remove entry from plain text history file.
:param str history_path: path to the history log file
:param int index: index of the command to be removed
"""
try:
with open(history_path, encoding="utf-8", mode="r+") as file_history:
lines = file_history.readlines()
file_history.seek(0)
file_history.truncate()
for number, line in enumerate(lines):
if number not in [index]:
file_history.write(line)
except OSError as e:
raise OSError(
_("Unable to remove entry from plain text history file {}").format(
history_path
)
) from e
def _remove_entry_from_JSON(history_path, index):
"""Remove entry from JSON history file.
:param str history_path: path to the history log file
:param int index: index of the command to be removed
"""
try:
with open(history_path, encoding="utf-8", mode="r+") as file_history:
try:
history_entries = json.load(file_history)
# Check if the index is valid
if 0 <= index < len(history_entries):
# Remove the entry at the specified index
del history_entries[index]
# Write the modified history back to the file
file_history.seek(0)
file_history.truncate()
json.dump(history_entries, file_history, indent=2)
except ValueError as ve:
raise ValueError(
_("Error decoding content of JSON history file {}").format(
history_path
)
) from ve
except OSError as e:
raise OSError(
_("Unable to remove entry from JSON history file {}").format(history_path)
) from e
[docs]def remove_entry(history_path, index):
"""Remove entry from history file.
:param str history_path: path to the history log file
:param int index: index of the command to be removed
"""
if get_history_file_extension(history_path) == ".json":
_remove_entry_from_JSON(history_path, index)
else:
_remove_entry_from_plain_text(history_path, index)
[docs]def convert_plain_text_to_JSON(history_path):
"""Convert plain text history log to JSON format.
:param str history_path: path to the history log file
"""
try:
lines = _read_from_plain_text(history_path)
# Extract file path and name without extension
file_path = history_path.parent
file_name_no_ext = history_path.stem
# JSON file path
json_path = file_path / (file_name_no_ext + ".json")
# Write JSON data to a new file
with open(json_path, encoding="utf-8", mode="w") as json_outfile:
json.dump(lines, json_outfile, indent=2)
except OSError as e:
raise OSError(
_("Unable to convert plain text history file {} to JSON format").format(
history_path
)
) from e
[docs]def get_initial_command_info(env_run):
"""Get information about the launched command.
:param env_run: environment needed for processing grass command
:return cmd_info dict: initial information about the launched command
"""
# Execution timestamp in ISO 8601 format
exec_time = datetime.now().isoformat()
# 2D raster MASK presence
env = gs.gisenv(env_run)
mapset_path = Path(env["GISDBASE"]) / env["LOCATION_NAME"] / env["MAPSET"]
mask2d_present = (mapset_path / "cell" / "MASK").exists()
# 3D raster MASK presence
mask3d_present = (mapset_path / "grid3" / "RASTER3D_MASK").exists()
# Computational region settings
region_settings = gs.region(env=env_run)
# Finalize the command info dictionary
cmd_info = {
"timestamp": exec_time,
"mask2d": mask2d_present,
"mask3d": mask3d_present,
"region": region_settings,
"status": Status.RUNNING.value,
}
return cmd_info
def _add_entry_to_JSON(history_path, entry):
"""Add entry to JSON history file.
:param str history_path: path to the history log file
:param dict entry: entry consisting of 'command' and 'command_info' keys
"""
try:
with open(history_path, encoding="utf-8", mode="r") as file_history:
existing_data = json.load(file_history)
except (OSError, ValueError):
existing_data = []
existing_data.append(entry)
try:
with open(history_path, encoding="utf-8", mode="w") as file_history:
json.dump(existing_data, file_history, indent=2)
except ValueError as ve:
raise ValueError(
_("Error decoding content of JSON history file {}").format(history_path)
) from ve
except OSError as e:
raise OSError(
_("Unable to add entry to JSON history file {}").format(history_path)
) from e
[docs]def add_entry(history_path, entry):
"""Add entry to history file.
:param str history_path: path to the history log file
:param dict entry: entry consisting of 'command' and 'command_info' keys
"""
if get_history_file_extension(history_path) == ".json":
_add_entry_to_JSON(history_path, entry)
else:
raise ValueError("Adding entries is supported only for JSON format.")
def _update_entry_in_JSON(history_path, command_info, index=None):
"""Update entry in JSON history file. If index is None it updates a last entry.
:param str history_path: path to the history log file
:param dict command_info: command info entry for update
:param int|None index: index of the command to be updated
"""
try:
with open(history_path, encoding="utf-8", mode="r+") as file_history:
try:
history_entries = json.load(file_history)
if index is None:
# Find index of last entry
index = len(history_entries) - 1
# Check if the index is valid
if 0 <= index < len(history_entries):
# Update entry
history_entries[index]["command_info"].update(command_info)
# Write the modified history back to the file
file_history.seek(0)
file_history.truncate()
json.dump(history_entries, file_history, indent=2)
except ValueError as ve:
raise ValueError(
_("Error decoding content of JSON history file {}").format(
history_path
)
) from ve
except OSError as e:
raise OSError(
_("Unable to update entry in JSON history file {}").format(history_path)
) from e
[docs]def update_entry(history_path, command_info, index=None):
"""Update entry in history file. If index is None it updates a last entry.
:param str history_path: path to the history log file
:param dict command_info: command info entry for update
:param int|None index: index of the command to be updated
"""
if get_history_file_extension(history_path) == ".json":
_update_entry_in_JSON(history_path, command_info, index)
else:
raise ValueError("Updating entries is supported only for JSON format.")
[docs]def copy(history_path, target_path):
"""Copy history file to the target location.
:param str history_path: path to the history log file
:param str target_path: target location for history log file
:return boolean: True if file is successfully copied
"""
try:
shutil.copyfile(history_path, target_path)
except OSError as e:
raise OSError(
_("Unable to copy history file {} to {}").format(history_path, target_path)
) from e
return True
