Mercurial > repos > bimib > cobraxy
changeset 335:2b7911a8366f draft
Uploaded
| author | luca_milaz |
|---|---|
| date | Thu, 04 Sep 2025 12:05:10 +0000 |
| parents | c561c060a55f |
| children | b89091ae2484 |
| files | COBRAxy/utils/general_utils.py |
| diffstat | 1 files changed, 643 insertions(+), 574 deletions(-) [+] |
line wrap: on
line diff
--- a/COBRAxy/utils/general_utils.py Thu Sep 04 12:04:42 2025 +0000 +++ b/COBRAxy/utils/general_utils.py Thu Sep 04 12:05:10 2025 +0000 @@ -1,575 +1,644 @@ -import math -import re -import sys -import csv -import pickle -import lxml.etree as ET - -from enum import Enum -from itertools import count -from typing import Any, Callable, Dict, Generic, List, Optional, TypeVar, Union - -import pandas as pd -import cobra - -# FILES -class FileFormat(Enum): - """ - Encodes possible file extensions to conditionally save data in a different format. - """ - DAT = ("dat",) # this is how galaxy treats all your files! - CSV = ("csv",) # this is how most editable input data is written - TSV = ("tsv",) # this is how most editable input data is ACTUALLY written TODO:more support pls!! - - SVG = ("svg",) # this is how most metabolic maps are written - PNG = ("png",) # this is a common output format for images (such as metabolic maps) - PDF = ("pdf",) # this is also a common output format for images, as it's required in publications. - - XML = ("xml",) # this is one main way cobra models appear in - JSON = ("json",) # this is the other - - TXT = ("txt",) # this is how most output data is written - - PICKLE = ("pickle", "pk", "p") # this is how all runtime data structures are saved - #TODO: we're in a pickle (ba dum tss), there's no point in supporting many extensions internally. The - # issue will never be solved for user-uploaded files and those are saved as .dat by galaxy anyway so it - # doesn't matter as long as we CAN recognize these 3 names as valid pickle extensions. We must however - # agree on an internal standard and use only that one, otherwise constructing usable paths becomes a nightmare. - @classmethod - def fromExt(cls, ext :str) -> "FileFormat": - """ - Converts a file extension string to a FileFormat instance. - - Args: - ext : The file extension as a string. - - Returns: - FileFormat: The FileFormat instance corresponding to the file extension. - """ - variantName = ext.upper() - if variantName in FileFormat.__members__: return FileFormat[variantName] - - variantName = variantName.lower() - for member in cls: - if variantName in member.value: return member - - raise ValueErr("ext", "a valid FileFormat file extension", ext) - - def __str__(self) -> str: - """ - (Private) converts to str representation. Good practice for usage with argparse. - - Returns: - str : the string representation of the file extension. - """ - return self.value[-1] #TODO: fix, it's the dumb pickle thing - -class FilePath(): - """ - Represents a file path. View this as an attempt to standardize file-related operations by expecting - values of this type in any process requesting a file path. - """ - def __init__(self, filePath :str, ext :FileFormat, *, prefix = "") -> None: - """ - (Private) Initializes an instance of FilePath. - - Args: - path : the end of the path, containing the file name. - ext : the file's extension. - prefix : anything before path, if the last '/' isn't there it's added by the code. - - Returns: - None : practically, a FilePath instance. - """ - self.ext = ext - self.filePath = filePath - - if prefix and prefix[-1] != '/': prefix += '/' - self.prefix = prefix - - @classmethod - def fromStrPath(cls, path :str) -> "FilePath": - """ - Factory method to parse a string from which to obtain, if possible, a valid FilePath instance. - - Args: - path : the string containing the path - - Raises: - PathErr : if the provided string doesn't represent a valid path. - - Returns: - FilePath : the constructed instance. - """ - # This method is often used to construct FilePath instances from ARGS UI arguments. These arguments *should* - # always be correct paths and could be used as raw strings, however most if not all functions that work with - # file paths request the FilePath objects specifically, which is a very good thing in any case other than this. - # What ends up happening is we spend time parsing a string into a FilePath so that the function accepts it, only - # to call show() immediately to bring back the string and open the file it points to. - # TODO: this is an indication that the arguments SHOULD BE OF TYPE FilePath if they are filepaths, this ENSURES - # their correctness when modifying the UI and avoids the pointless back-and-forth. - result = re.search(r"^(?P<prefix>.*\/)?(?P<name>.*)\.(?P<ext>[^.]*)$", path) - if not result or not result["name"] or not result["ext"]: - raise PathErr(path, "cannot recognize folder structure or extension in path") - - prefix = result["prefix"] if result["prefix"] else "" - return cls(result["name"], FileFormat.fromExt(result["ext"]), prefix = prefix) - - def show(self) -> str: - """ - Shows the path as a string. - - Returns: - str : the path shown as a string. - """ - return f"{self.prefix}{self.filePath}.{self.ext}" - - def __str__(self) -> str: return self.show() - -# ERRORS -def terminate(msg :str) -> None: - """ - Terminate the execution of the script with an error message. - - Args: - msg (str): The error message to be displayed. - - Returns: - None - """ - sys.exit(f"Execution aborted: {msg}\n") - -def logWarning(msg :str, loggerPath :str) -> None: - """ - Log a warning message to an output log file and print it to the console. The final period and a - newline is added by the function. - - Args: - s (str): The warning message to be logged and printed. - loggerPath : The file path of the output log file. Given as a string, parsed to a FilePath and - immediately read back (beware relative expensive operation, log with caution). - - Returns: - None - """ - # building the path and then reading it immediately seems useless, but it's actually a way of - # validating that reduces repetition on the caller's side. Besides, logging a message by writing - # to a file is supposed to be computationally expensive anyway, so this is also a good deterrent from - # mindlessly logging whenever something comes up, log at the very end and tell the user everything - # that went wrong. If you don't like it: implement a persistent runtime buffer that gets dumped to - # the file only at the end of the program's execution. - with open(FilePath.fromStrPath(loggerPath).show(), 'a') as log: log.write(f"{msg}.\n") - -class CustomErr(Exception): - """ - Custom error class to handle exceptions in a structured way, with a unique identifier and a message. - """ - __idGenerator = count() - errName = "Custom Error" - def __init__(self, msg :str, details = "", explicitErrCode = -1) -> None: - """ - (Private) Initializes an instance of CustomErr. - - Args: - msg (str): Error message to be displayed. - details (str): Informs the user more about the error encountered. Defaults to "". - explicitErrCode (int): Explicit error code to be used. Defaults to -1. - - Returns: - None : practically, a CustomErr instance. - """ - self.msg = msg - self.details = details - - self.id = max(explicitErrCode, next(CustomErr.__idGenerator)) - - def throw(self, loggerPath = "") -> None: - """ - Raises the current CustomErr instance, logging a warning message before doing so. - - Raises: - self: The current CustomErr instance. - - Returns: - None - """ - if loggerPath: logWarning(str(self), loggerPath) - raise self - - def abort(self) -> None: - """ - Aborts the execution of the script. - - Returns: - None - """ - terminate(str(self)) - - def __str__(self) -> str: - """ - (Private) Returns a string representing the current CustomErr instance. - - Returns: - str: A string representing the current CustomErr instance. - """ - return f"{CustomErr.errName} #{self.id}: {self.msg}, {self.details}." - -class ArgsErr(CustomErr): - """ - CustomErr subclass for UI arguments errors. - """ - errName = "Args Error" - def __init__(self, argName :str, expected :Any, actual :Any, msg = "no further details provided") -> None: - super().__init__(f"argument \"{argName}\" expected {expected} but got {actual}", msg) - -class DataErr(CustomErr): - """ - CustomErr subclass for data formatting errors. - """ - errName = "Data Format Error" - def __init__(self, fileName :str, msg = "no further details provided") -> None: - super().__init__(f"file \"{fileName}\" contains malformed data", msg) - -class PathErr(CustomErr): - """ - CustomErr subclass for filepath formatting errors. - """ - errName = "Path Error" - def __init__(self, path :FilePath, msg = "no further details provided") -> None: - super().__init__(f"path \"{path}\" is invalid", msg) - -class ValueErr(CustomErr): - """ - CustomErr subclass for any value error. - """ - errName = "Value Error" - def __init__(self, valueName: str, expected :Any, actual :Any, msg = "no further details provided") -> None: - super().__init__("value " + f"\"{valueName}\" " * bool(valueName) + f"was supposed to be {expected}, but got {actual} instead", msg) - -# RESULT -T = TypeVar('T') -E = TypeVar('E', bound = CustomErr) # should bind to Result.ResultErr but python happened! -class Result(Generic[T, E]): - class ResultErr(CustomErr): - """ - CustomErr subclass for all Result errors. - """ - errName = "Result Error" - def __init__(self, msg = "no further details provided") -> None: - super().__init__(msg) - """ - Class to handle the result of an operation, with a value and a boolean flag to indicate - whether the operation was successful or not. - """ - def __init__(self, value :Union[T, E], isOk :bool) -> None: - """ - (Private) Initializes an instance of Result. - - Args: - value (Union[T, E]): The value to be stored in the Result instance. - isOk (bool): A boolean flag to indicate whether the operation was successful or not. - - Returns: - None : practically, a Result instance. - """ - self.isOk = isOk - self.isErr = not isOk - self.value = value - - @classmethod - def Ok(cls, value :T) -> "Result": - """ - Constructs a new Result instance with a successful operation. - - Args: - value (T): The value to be stored in the Result instance, set as successful. - - Returns: - Result: A new Result instance with a successful operation. - """ - return Result(value, isOk = True) - - @classmethod - def Err(cls, value :E) -> "Result": - """ - Constructs a new Result instance with a failed operation. - - Args: - value (E): The value to be stored in the Result instance, set as failed. - - Returns: - Result: A new Result instance with a failed operation. - """ - return Result(value, isOk = False) - - def unwrap(self) -> T: - """ - Unwraps the value of the Result instance, if the operation was successful. - - Raises: - ResultErr: If the operation was not successful. - - Returns: - T: The value of the Result instance, if the operation was successful. - """ - if self.isOk: return self.value - raise Result.ResultErr(f"Unwrapped Result.Err : {self.value}") - - def unwrapOr(self, default :T) -> T: - """ - Unwraps the value of the Result instance, if the operation was successful, otherwise - it returns a default value. - - Args: - default (T): The default value to be returned if the operation was not successful. - - Returns: - T: The value of the Result instance, if the operation was successful, - otherwise the default value. - """ - return self.value if self.isOk else default - - def expect(self, err :"Result.ResultErr") -> T: - """ - Expects that the value of the Result instance is successful, otherwise it raises an error. - - Args: - err (Exception): The error to be raised if the operation was not successful. - - Raises: - err: The error raised if the operation was not successful. - - Returns: - T: The value of the Result instance, if the operation was successful. - """ - if self.isOk: return self.value - raise err - - U = TypeVar("U") - def map(self, mapper: Callable[[T], U]) -> "Result[U, E]": - """ - Maps the value of the current Result to whatever is returned by the mapper function. - If the Result contained an unsuccessful operation to begin with it remains unchanged - (a reference to the current instance is returned). - If the mapper function panics the returned result instance will be of the error kind. - - Args: - mapper (Callable[[T], U]): The mapper operation to be applied to the Result value. - - Returns: - Result[U, E]: The result of the mapper operation applied to the Result value. - """ - if self.isErr: return self - try: return Result.Ok(mapper(self.value)) - except Exception as e: return Result.Err(e) - - D = TypeVar("D", bound = "Result.ResultErr") - def mapErr(self, mapper :Callable[[E], D]) -> "Result[T, D]": - """ - Maps the error of the current Result to whatever is returned by the mapper function. - If the Result contained a successful operation it remains unchanged - (a reference to the current instance is returned). - If the mapper function panics this method does as well. - - Args: - mapper (Callable[[E], D]): The mapper operation to be applied to the Result error. - - Returns: - Result[U, E]: The result of the mapper operation applied to the Result error. - """ - if self.isOk: return self - return Result.Err(mapper(self.value)) - - def __str__(self): - return f"Result::{'Ok' if self.isOk else 'Err'}({self.value})" - -# FILES -def read_dataset(path :FilePath, datasetName = "Dataset (not actual file name!)") -> pd.DataFrame: - """ - Reads a .csv or .tsv file and returns it as a Pandas DataFrame. - - Args: - path : the path to the dataset file. - datasetName : the name of the dataset. - - Raises: - DataErr: If anything goes wrong when trying to open the file, if pandas thinks the dataset is empty or if - it has less than 2 columns. - - Returns: - pandas.DataFrame: The dataset loaded as a Pandas DataFrame. - """ - # I advise against the use of this function. This is an attempt at standardizing bad legacy code rather than - # removing / replacing it to avoid introducing as many bugs as possible in the tools still relying on this code. - # First off, this is not the best way to distinguish between .csv and .tsv files and Galaxy itself makes it really - # hard to implement anything better. Also, this function's name advertizes it as a dataset-specific operation and - # contains dubious responsibility (how many columns..) while being a file-opening function instead. My suggestion is - # TODO: stop using dataframes ever at all in anything and find a way to have tight control over file extensions. - try: dataset = pd.read_csv(path.show(), sep = '\t', header = None, engine = "python") - except: - try: dataset = pd.read_csv(path.show(), sep = ',', header = 0, engine = "python") - except Exception as err: raise DataErr(datasetName, f"encountered empty or wrongly formatted data: {err}") - - if len(dataset.columns) < 2: raise DataErr(datasetName, "a dataset is always meant to have at least 2 columns") - return dataset - -def readPickle(path :FilePath) -> Any: - """ - Reads the contents of a .pickle file, which needs to exist at the given path. - - Args: - path : the path to the .pickle file. - - Returns: - Any : the data inside a pickle file, could be anything. - """ - with open(path.show(), "rb") as fd: return pickle.load(fd) - -def writePickle(path :FilePath, data :Any) -> None: - """ - Saves any data in a .pickle file, created at the given path. - - Args: - path : the path to the .pickle file. - data : the data to be written to the file. - - Returns: - None - """ - with open(path.show(), "wb") as fd: pickle.dump(data, fd) - -def readCsv(path :FilePath, delimiter = ',', *, skipHeader = True) -> List[List[str]]: - """ - Reads the contents of a .csv file, which needs to exist at the given path. - - Args: - path : the path to the .csv file. - delimiter : allows other subformats such as .tsv to be opened by the same method (\\t delimiter). - skipHeader : whether the first row of the file is a header and should be skipped. - - Returns: - List[List[str]] : list of rows from the file, each parsed as a list of strings originally separated by commas. - """ - with open(path.show(), "r", newline = "") as fd: return list(csv.reader(fd, delimiter = delimiter))[skipHeader:] - -def readSvg(path :FilePath, customErr :Optional[Exception] = None) -> ET.ElementTree: - """ - Reads the contents of a .svg file, which needs to exist at the given path. - - Args: - path : the path to the .svg file. - - Raises: - DataErr : if the map is malformed. - - Returns: - Any : the data inside a svg file, could be anything. - """ - try: return ET.parse(path.show()) - except (ET.XMLSyntaxError, ET.XMLSchemaParseError) as err: - raise customErr if customErr else err - -def writeSvg(path :FilePath, data:ET.ElementTree) -> None: - """ - Saves svg data opened with lxml.etree in a .svg file, created at the given path. - - Args: - path : the path to the .svg file. - data : the data to be written to the file. - - Returns: - None - """ - with open(path.show(), "wb") as fd: fd.write(ET.tostring(data)) - -# UI ARGUMENTS -class Bool: - def __init__(self, argName :str) -> None: - self.argName = argName - - def __call__(self, s :str) -> bool: return self.check(s) - - def check(self, s :str) -> bool: - s = s.lower() - if s == "true" : return True - if s == "false": return False - raise ArgsErr(self.argName, "boolean string (true or false, not case sensitive)", f"\"{s}\"") - -class Float: - def __init__(self, argName = "Dataset values, not an argument") -> None: - self.argName = argName - - def __call__(self, s :str) -> float: return self.check(s) - - def check(self, s :str) -> float: - try: return float(s) - except ValueError: - s = s.lower() - if s == "nan" or s == "none": return math.nan - raise ArgsErr(self.argName, "numeric string or \"None\" or \"NaN\" (not case sensitive)", f"\"{s}\"") - -# MODELS -OldRule = List[Union[str, "OldRule"]] -class Model(Enum): - """ - Represents a metabolic model, either custom or locally supported. Custom models don't point - to valid file paths. - """ - - Recon = "Recon" - ENGRO2 = "ENGRO2" - ENGRO2_no_legend = "ENGRO2_no_legend" - HMRcore = "HMRcore" - HMRcore_no_legend = "HMRcore_no_legend" - Custom = "Custom" # Exists as a valid variant in the UI, but doesn't point to valid file paths. - - def __raiseMissingPathErr(self, path :Optional[FilePath]) -> None: - if not path: raise PathErr("<<MISSING>>", "it's necessary to provide a custom path when retrieving files from a custom model") - - def getRules(self, toolDir :str, customPath :Optional[FilePath] = None) -> Dict[str, Dict[str, OldRule]]: - """ - Open "rules" file for this model. - - Returns: - Dict[str, Dict[str, OldRule]] : the rules for this model. - """ - path = customPath if self is Model.Custom else FilePath(f"{self.name}_rules", FileFormat.PICKLE, prefix = f"{toolDir}/local/pickle files/") - self.__raiseMissingPathErr(path) - return readPickle(path) - - def getTranslator(self, toolDir :str, customPath :Optional[FilePath] = None) -> Dict[str, Dict[str, str]]: - """ - Open "gene translator (old: gene_in_rule)" file for this model. - - Returns: - Dict[str, Dict[str, str]] : the translator dict for this model. - """ - path = customPath if self is Model.Custom else FilePath(f"{self.name}_genes", FileFormat.PICKLE, prefix = f"{toolDir}/local/pickle files/") - self.__raiseMissingPathErr(path) - return readPickle(path) - - def getMap(self, toolDir = ".", customPath :Optional[FilePath] = None) -> ET.ElementTree: - path = customPath if self is Model.Custom else FilePath(f"{self.name}_map", FileFormat.SVG, prefix = f"{toolDir}/local/svg metabolic maps/") - self.__raiseMissingPathErr(path) - return readSvg(path, customErr = DataErr(path, f"custom map in wrong format")) - - def getCOBRAmodel(self, toolDir = ".", customPath :Optional[FilePath] = None, customExtension :Optional[FilePath]=None)->cobra.Model: - if(self is Model.Custom): - return self.load_custom_model(customPath, customExtension) - else: - return cobra.io.read_sbml_model(FilePath(f"{self.name}", FileFormat.XML, prefix = f"{toolDir}/local/models/").show()) - - def load_custom_model(self, file_path :FilePath, ext :Optional[FileFormat] = None) -> cobra.Model: - ext = ext if ext else file_path.ext - try: - if ext is FileFormat.XML: - return cobra.io.read_sbml_model(file_path.show()) - - if ext is FileFormat.JSON: - return cobra.io.load_json_model(file_path.show()) - - except Exception as e: raise DataErr(file_path, e.__str__()) - raise DataErr(file_path, - f"Fomat \"{file_path.ext}\" is not recognized, only JSON and XML files are supported.") - +import math +import re +import sys +import csv +import pickle +import lxml.etree as ET + +from enum import Enum +from itertools import count +from typing import Any, Callable, Dict, Generic, List, Literal, Optional, TypeVar, Union + +import pandas as pd +import cobra + +import zipfile +import gzip +import bz2 +from io import StringIO + +# FILES +class FileFormat(Enum): + """ + Encodes possible file extensions to conditionally save data in a different format. + """ + DAT = ("dat",) # this is how galaxy treats all your files! + CSV = ("csv",) # this is how most editable input data is written + TSV = ("tsv",) # this is how most editable input data is ACTUALLY written + + SVG = ("svg",) # this is how most metabolic maps are written + PNG = ("png",) # this is a common output format for images (such as metabolic maps) + PDF = ("pdf",) # this is also a common output format for images, as it's required in publications. + + XML = ("xml","xml.gz", "xml.zip", "xml.bz2") # SBML files are XML files, sometimes compressed + JSON = ("json","json.gz", "json.zip", "json.bz2") # COBRA models can be stored as JSON files, sometimes compressed + + TXT = ("txt",) # this is how most output data is written + + PICKLE = ("pickle", "pk", "p") # this is how all runtime data structures are saved + + def __init__(self): + self.original_extension = "" + + + @classmethod + def fromExt(cls, ext :str) -> "FileFormat": + """ + Converts a file extension string to a FileFormat instance. + + Args: + ext : The file extension as a string. + + Returns: + FileFormat: The FileFormat instance corresponding to the file extension. + """ + variantName = ext.upper() + if variantName in FileFormat.__members__: + instance = FileFormat[variantName] + instance.original_extension = ext + return instance + + variantName = variantName.lower() + for member in cls: + if variantName in member.value: + member.original_extension = ext + return member + + raise ValueErr("ext", "a valid FileFormat file extension", ext) + + def __str__(self) -> str: + """ + (Private) converts to str representation. Good practice for usage with argparse. + + Returns: + str : the string representation of the file extension. + """ + + if(self.values[-1] in ["json", "xml"]): #return the original string extension for compressed files + return self.original_extension + else: + return self.value[-1] # for all other formats and pickle + +class FilePath(): + """ + Represents a file path. View this as an attempt to standardize file-related operations by expecting + values of this type in any process requesting a file path. + """ + def __init__(self, filePath :str, ext :FileFormat, *, prefix = "") -> None: + """ + (Private) Initializes an instance of FilePath. + + Args: + path : the end of the path, containing the file name. + ext : the file's extension. + prefix : anything before path, if the last '/' isn't there it's added by the code. + + Returns: + None : practically, a FilePath instance. + """ + self.ext = ext + self.filePath = filePath + + if prefix and prefix[-1] != '/': prefix += '/' + self.prefix = prefix + + @classmethod + def fromStrPath(cls, path :str) -> "FilePath": + """ + Factory method to parse a string from which to obtain, if possible, a valid FilePath instance. + It detects double extensions such as .json.gz and .xml.bz2, which are common in COBRA models. + These double extensions are not supported for other file types such as .csv. + + Args: + path : the string containing the path + + Raises: + PathErr : if the provided string doesn't represent a valid path. + + Returns: + FilePath : the constructed instance. + """ + # This method is often used to construct FilePath instances from ARGS UI arguments. These arguments *should* + # always be correct paths and could be used as raw strings, however most if not all functions that work with + # file paths request the FilePath objects specifically, which is a very good thing in any case other than this. + # What ends up happening is we spend time parsing a string into a FilePath so that the function accepts it, only + # to call show() immediately to bring back the string and open the file it points to. + # TODO: this is an indication that the arguments SHOULD BE OF TYPE FilePath if they are filepaths, this ENSURES + # their correctness when modifying the UI and avoids the pointless back-and-forth. + result = re.search(r"^(?P<prefix>.*\/)?(?P<name>.*)\.(?P<ext>[^.]*)$", path) + if not result or not result["name"] or not result["ext"]: + raise PathErr(path, "cannot recognize folder structure or extension in path") + + prefix = result["prefix"] if result["prefix"] else "" + name, ext = result["name"], result["ext"] + + # Split path into parts + parts = path.split(".") + if len(parts) >= 3: + penultimate = parts[-2] + last = parts[-1] + if penultimate in {"json", "xml"}: + name = ".".join(parts[:-2]) + ext = f"{penultimate}.{last}" + + return cls(name, FileFormat.fromExt(ext), prefix=prefix) + + def show(self) -> str: + """ + Shows the path as a string. + + Returns: + str : the path shown as a string. + """ + return f"{self.prefix}{self.filePath}.{self.ext}" + + def __str__(self) -> str: return self.show() + +# ERRORS +def terminate(msg :str) -> None: + """ + Terminate the execution of the script with an error message. + + Args: + msg (str): The error message to be displayed. + + Returns: + None + """ + sys.exit(f"Execution aborted: {msg}\n") + +def logWarning(msg :str, loggerPath :str) -> None: + """ + Log a warning message to an output log file and print it to the console. The final period and a + newline is added by the function. + + Args: + s (str): The warning message to be logged and printed. + loggerPath : The file path of the output log file. Given as a string, parsed to a FilePath and + immediately read back (beware relative expensive operation, log with caution). + + Returns: + None + """ + # building the path and then reading it immediately seems useless, but it's actually a way of + # validating that reduces repetition on the caller's side. Besides, logging a message by writing + # to a file is supposed to be computationally expensive anyway, so this is also a good deterrent from + # mindlessly logging whenever something comes up, log at the very end and tell the user everything + # that went wrong. If you don't like it: implement a persistent runtime buffer that gets dumped to + # the file only at the end of the program's execution. + with open(FilePath.fromStrPath(loggerPath).show(), 'a') as log: log.write(f"{msg}.\n") + +class CustomErr(Exception): + """ + Custom error class to handle exceptions in a structured way, with a unique identifier and a message. + """ + __idGenerator = count() + errName = "Custom Error" + def __init__(self, msg :str, details = "", explicitErrCode = -1) -> None: + """ + (Private) Initializes an instance of CustomErr. + + Args: + msg (str): Error message to be displayed. + details (str): Informs the user more about the error encountered. Defaults to "". + explicitErrCode (int): Explicit error code to be used. Defaults to -1. + + Returns: + None : practically, a CustomErr instance. + """ + self.msg = msg + self.details = details + + self.id = max(explicitErrCode, next(CustomErr.__idGenerator)) + + def throw(self, loggerPath = "") -> None: + """ + Raises the current CustomErr instance, logging a warning message before doing so. + + Raises: + self: The current CustomErr instance. + + Returns: + None + """ + if loggerPath: logWarning(str(self), loggerPath) + raise self + + def abort(self) -> None: + """ + Aborts the execution of the script. + + Returns: + None + """ + terminate(str(self)) + + def __str__(self) -> str: + """ + (Private) Returns a string representing the current CustomErr instance. + + Returns: + str: A string representing the current CustomErr instance. + """ + return f"{CustomErr.errName} #{self.id}: {self.msg}, {self.details}." + +class ArgsErr(CustomErr): + """ + CustomErr subclass for UI arguments errors. + """ + errName = "Args Error" + def __init__(self, argName :str, expected :Any, actual :Any, msg = "no further details provided") -> None: + super().__init__(f"argument \"{argName}\" expected {expected} but got {actual}", msg) + +class DataErr(CustomErr): + """ + CustomErr subclass for data formatting errors. + """ + errName = "Data Format Error" + def __init__(self, fileName :str, msg = "no further details provided") -> None: + super().__init__(f"file \"{fileName}\" contains malformed data", msg) + +class PathErr(CustomErr): + """ + CustomErr subclass for filepath formatting errors. + """ + errName = "Path Error" + def __init__(self, path :FilePath, msg = "no further details provided") -> None: + super().__init__(f"path \"{path}\" is invalid", msg) + +class ValueErr(CustomErr): + """ + CustomErr subclass for any value error. + """ + errName = "Value Error" + def __init__(self, valueName: str, expected :Any, actual :Any, msg = "no further details provided") -> None: + super().__init__("value " + f"\"{valueName}\" " * bool(valueName) + f"was supposed to be {expected}, but got {actual} instead", msg) + +# RESULT +T = TypeVar('T') +E = TypeVar('E', bound = CustomErr) # should bind to Result.ResultErr but python happened! +class Result(Generic[T, E]): + class ResultErr(CustomErr): + """ + CustomErr subclass for all Result errors. + """ + errName = "Result Error" + def __init__(self, msg = "no further details provided") -> None: + super().__init__(msg) + """ + Class to handle the result of an operation, with a value and a boolean flag to indicate + whether the operation was successful or not. + """ + def __init__(self, value :Union[T, E], isOk :bool) -> None: + """ + (Private) Initializes an instance of Result. + + Args: + value (Union[T, E]): The value to be stored in the Result instance. + isOk (bool): A boolean flag to indicate whether the operation was successful or not. + + Returns: + None : practically, a Result instance. + """ + self.isOk = isOk + self.isErr = not isOk + self.value = value + + @classmethod + def Ok(cls, value :T) -> "Result": + """ + Constructs a new Result instance with a successful operation. + + Args: + value (T): The value to be stored in the Result instance, set as successful. + + Returns: + Result: A new Result instance with a successful operation. + """ + return Result(value, isOk = True) + + @classmethod + def Err(cls, value :E) -> "Result": + """ + Constructs a new Result instance with a failed operation. + + Args: + value (E): The value to be stored in the Result instance, set as failed. + + Returns: + Result: A new Result instance with a failed operation. + """ + return Result(value, isOk = False) + + def unwrap(self) -> T: + """ + Unwraps the value of the Result instance, if the operation was successful. + + Raises: + ResultErr: If the operation was not successful. + + Returns: + T: The value of the Result instance, if the operation was successful. + """ + if self.isOk: return self.value + raise Result.ResultErr(f"Unwrapped Result.Err : {self.value}") + + def unwrapOr(self, default :T) -> T: + """ + Unwraps the value of the Result instance, if the operation was successful, otherwise + it returns a default value. + + Args: + default (T): The default value to be returned if the operation was not successful. + + Returns: + T: The value of the Result instance, if the operation was successful, + otherwise the default value. + """ + return self.value if self.isOk else default + + def expect(self, err :"Result.ResultErr") -> T: + """ + Expects that the value of the Result instance is successful, otherwise it raises an error. + + Args: + err (Exception): The error to be raised if the operation was not successful. + + Raises: + err: The error raised if the operation was not successful. + + Returns: + T: The value of the Result instance, if the operation was successful. + """ + if self.isOk: return self.value + raise err + + U = TypeVar("U") + def map(self, mapper: Callable[[T], U]) -> "Result[U, E]": + """ + Maps the value of the current Result to whatever is returned by the mapper function. + If the Result contained an unsuccessful operation to begin with it remains unchanged + (a reference to the current instance is returned). + If the mapper function panics the returned result instance will be of the error kind. + + Args: + mapper (Callable[[T], U]): The mapper operation to be applied to the Result value. + + Returns: + Result[U, E]: The result of the mapper operation applied to the Result value. + """ + if self.isErr: return self + try: return Result.Ok(mapper(self.value)) + except Exception as e: return Result.Err(e) + + D = TypeVar("D", bound = "Result.ResultErr") + def mapErr(self, mapper :Callable[[E], D]) -> "Result[T, D]": + """ + Maps the error of the current Result to whatever is returned by the mapper function. + If the Result contained a successful operation it remains unchanged + (a reference to the current instance is returned). + If the mapper function panics this method does as well. + + Args: + mapper (Callable[[E], D]): The mapper operation to be applied to the Result error. + + Returns: + Result[U, E]: The result of the mapper operation applied to the Result error. + """ + if self.isOk: return self + return Result.Err(mapper(self.value)) + + def __str__(self): + return f"Result::{'Ok' if self.isOk else 'Err'}({self.value})" + +# FILES +def read_dataset(path :FilePath, datasetName = "Dataset (not actual file name!)") -> pd.DataFrame: + """ + Reads a .csv or .tsv file and returns it as a Pandas DataFrame. + + Args: + path : the path to the dataset file. + datasetName : the name of the dataset. + + Raises: + DataErr: If anything goes wrong when trying to open the file, if pandas thinks the dataset is empty or if + it has less than 2 columns. + + Returns: + pandas.DataFrame: The dataset loaded as a Pandas DataFrame. + """ + # I advise against the use of this function. This is an attempt at standardizing bad legacy code rather than + # removing / replacing it to avoid introducing as many bugs as possible in the tools still relying on this code. + # First off, this is not the best way to distinguish between .csv and .tsv files and Galaxy itself makes it really + # hard to implement anything better. Also, this function's name advertizes it as a dataset-specific operation and + # contains dubious responsibility (how many columns..) while being a file-opening function instead. My suggestion is + # TODO: stop using dataframes ever at all in anything and find a way to have tight control over file extensions. + try: dataset = pd.read_csv(path.show(), sep = '\t', header = None, engine = "python") + except: + try: dataset = pd.read_csv(path.show(), sep = ',', header = 0, engine = "python") + except Exception as err: raise DataErr(datasetName, f"encountered empty or wrongly formatted data: {err}") + + if len(dataset.columns) < 2: raise DataErr(datasetName, "a dataset is always meant to have at least 2 columns") + return dataset + +def readPickle(path :FilePath) -> Any: + """ + Reads the contents of a .pickle file, which needs to exist at the given path. + + Args: + path : the path to the .pickle file. + + Returns: + Any : the data inside a pickle file, could be anything. + """ + with open(path.show(), "rb") as fd: return pickle.load(fd) + +def writePickle(path :FilePath, data :Any) -> None: + """ + Saves any data in a .pickle file, created at the given path. + + Args: + path : the path to the .pickle file. + data : the data to be written to the file. + + Returns: + None + """ + with open(path.show(), "wb") as fd: pickle.dump(data, fd) + +def readCsv(path :FilePath, delimiter = ',', *, skipHeader = True) -> List[List[str]]: + """ + Reads the contents of a .csv file, which needs to exist at the given path. + + Args: + path : the path to the .csv file. + delimiter : allows other subformats such as .tsv to be opened by the same method (\\t delimiter). + skipHeader : whether the first row of the file is a header and should be skipped. + + Returns: + List[List[str]] : list of rows from the file, each parsed as a list of strings originally separated by commas. + """ + with open(path.show(), "r", newline = "") as fd: return list(csv.reader(fd, delimiter = delimiter))[skipHeader:] + +def readSvg(path :FilePath, customErr :Optional[Exception] = None) -> ET.ElementTree: + """ + Reads the contents of a .svg file, which needs to exist at the given path. + + Args: + path : the path to the .svg file. + + Raises: + DataErr : if the map is malformed. + + Returns: + Any : the data inside a svg file, could be anything. + """ + try: return ET.parse(path.show()) + except (ET.XMLSyntaxError, ET.XMLSchemaParseError) as err: + raise customErr if customErr else err + +def writeSvg(path :FilePath, data:ET.ElementTree) -> None: + """ + Saves svg data opened with lxml.etree in a .svg file, created at the given path. + + Args: + path : the path to the .svg file. + data : the data to be written to the file. + + Returns: + None + """ + with open(path.show(), "wb") as fd: fd.write(ET.tostring(data)) + +# UI ARGUMENTS +class Bool: + def __init__(self, argName :str) -> None: + self.argName = argName + + def __call__(self, s :str) -> bool: return self.check(s) + + def check(self, s :str) -> bool: + s = s.lower() + if s == "true" : return True + if s == "false": return False + raise ArgsErr(self.argName, "boolean string (true or false, not case sensitive)", f"\"{s}\"") + +class Float: + def __init__(self, argName = "Dataset values, not an argument") -> None: + self.argName = argName + + def __call__(self, s :str) -> float: return self.check(s) + + def check(self, s :str) -> float: + try: return float(s) + except ValueError: + s = s.lower() + if s == "nan" or s == "none": return math.nan + raise ArgsErr(self.argName, "numeric string or \"None\" or \"NaN\" (not case sensitive)", f"\"{s}\"") + +# MODELS +OldRule = List[Union[str, "OldRule"]] +class Model(Enum): + """ + Represents a metabolic model, either custom or locally supported. Custom models don't point + to valid file paths. + """ + + Recon = "Recon" + ENGRO2 = "ENGRO2" + ENGRO2_no_legend = "ENGRO2_no_legend" + HMRcore = "HMRcore" + HMRcore_no_legend = "HMRcore_no_legend" + Custom = "Custom" # Exists as a valid variant in the UI, but doesn't point to valid file paths. + + def __raiseMissingPathErr(self, path :Optional[FilePath]) -> None: + if not path: raise PathErr("<<MISSING>>", "it's necessary to provide a custom path when retrieving files from a custom model") + + def getRules(self, toolDir :str, customPath :Optional[FilePath] = None) -> Dict[str, Dict[str, OldRule]]: + """ + Open "rules" file for this model. + + Returns: + Dict[str, Dict[str, OldRule]] : the rules for this model. + """ + path = customPath if self is Model.Custom else FilePath(f"{self.name}_rules", FileFormat.PICKLE, prefix = f"{toolDir}/local/pickle files/") + self.__raiseMissingPathErr(path) + return readPickle(path) + + def getTranslator(self, toolDir :str, customPath :Optional[FilePath] = None) -> Dict[str, Dict[str, str]]: + """ + Open "gene translator (old: gene_in_rule)" file for this model. + + Returns: + Dict[str, Dict[str, str]] : the translator dict for this model. + """ + path = customPath if self is Model.Custom else FilePath(f"{self.name}_genes", FileFormat.PICKLE, prefix = f"{toolDir}/local/pickle files/") + self.__raiseMissingPathErr(path) + return readPickle(path) + + def getMap(self, toolDir = ".", customPath :Optional[FilePath] = None) -> ET.ElementTree: + path = customPath if self is Model.Custom else FilePath(f"{self.name}_map", FileFormat.SVG, prefix = f"{toolDir}/local/svg metabolic maps/") + self.__raiseMissingPathErr(path) + return readSvg(path, customErr = DataErr(path, f"custom map in wrong format")) + + def getCOBRAmodel(self, toolDir = ".", customPath :Optional[FilePath] = None, customExtension :Optional[FilePath]=None)->cobra.Model: + if(self is Model.Custom): + return self.load_custom_model(customPath, customExtension) + else: + return cobra.io.read_sbml_model(FilePath(f"{self.name}", FileFormat.XML, prefix = f"{toolDir}/local/models/").show()) + + def load_custom_model(self, file_path :FilePath, ext :Optional[FileFormat] = None) -> cobra.Model: + ext = ext if ext else file_path.ext + try: + if ext in FileFormat.XML: + return cobra.io.read_sbml_model(file_path.show()) + + if ext in FileFormat.JSON: + # Compressed files are not automatically handled by cobra + if(ext == "json"): + return cobra.io.load_json_model(file_path.show()) + else: + return self.extract_json_model(file_path, ext) + + except Exception as e: raise DataErr(file_path, e.__str__()) + raise DataErr(file_path, + f"Fomat \"{file_path.ext}\" is not recognized, only JSON and XML files are supported.") + + + def extract_json_model(file_path:FilePath, ext :FileFormat) -> cobra.Model: + """ + Extract json COBRA model from a compressed file (zip, gz, bz2). + + Args: + file_path: File path of the model + ext: File extensions of class FileFormat (should be .zip, .gz or .bz2) + + Returns: + cobra.Model: COBRApy model + + Raises: + Exception: Extraction errors + """ + ext_str = str(ext) + + try: + if '.zip' in ext_str: + with zipfile.ZipFile(file_path.show(), 'r') as zip_ref: + with zip_ref.open(zip_ref.namelist()[0]) as json_file: + content = json_file.read().decode('utf-8') + return cobra.io.load_json_model(StringIO(content)) + elif '.gz' in ext_str: + with gzip.open(file_path.show(), 'rt', encoding='utf-8') as gz_ref: + return cobra.io.load_json_model(gz_ref) + elif '.bz2' in ext_str: + with bz2.open(file_path.show(), 'rt', encoding='utf-8') as bz2_ref: + return cobra.io.load_json_model(bz2_ref) + else: + raise ValueError(f"Compression format not supported: {ext_str}. Supported: .zip, .gz and .bz2") + + except Exception as e: + raise Exception(f"Error during model extraction: {str(e)}") + + + def __str__(self) -> str: return self.value \ No newline at end of file