Skip to content
GitLab
Commit bbd5af95 authored by Greg Harris's avatar Greg Harris
Browse files

Updating file endings to CRLF

parent 4776f9c9
Pipeline #99658 passed with stages
in 1 minute and 7 seconds
Expand all Hide whitespace changes
Inline Side-by-side
src/wbdutil/common/configuration.py
View file @ bbd5af95
......@@ -62,12 +62,12 @@ class Configuration:
rtype: Dict[str, Any]
"""
return self._config.get("welllog_mapping")
@property
def las_file_mapping(self) -> Union[Dict[str, Any], None]:
"""
Gets the las_file_mapping.
return: the las_file_mapping
rtype: Dict[str, Any]
"""
return self._config.get("las_file_mapping")
@property
def las_file_mapping(self) -> Union[Dict[str, Any], None]:
"""
Gets the las_file_mapping.
return: the las_file_mapping
rtype: Dict[str, Any]
"""
return self._config.get("las_file_mapping")
src/wbdutil/common/exceptions.py
View file @ bbd5af95
from typing import List
class WbdutilException(Exception):
"""
Base class for exceptions raised within Wdutil
"""
def __init__(self, *args):
"""
Create a new instance of a WdutilException
:param *args object: The Exception arguments
"""
super().__init__(args)
class WbdutilValidationException(Exception):
"""
General validation exception raised within Wdutil
"""
def __init__(self, error_messages: List[str], *args):
"""
Create a new instance of a WdutilValidationException
:param error_messages List[str]: The validation errors
:param *args object: The Exception arguments
"""
self._error_messages = error_messages
super().__init__(args)
@property
def error_messages(self) -> List[str]:
"""
Gets the list of validation errors
return: the validation errors
rtype: List[str]
"""
return self._error_messages
class WbdutilAttributeError(WbdutilException):
"""
Common exception class for attribute errors raised in the Wbdutil
"""
def __init__(self, message: str, inner_exception: Exception, *args):
"""
Create a new instance of a ExtendedPropertiesAttributeError
:param str message: The exception message
:param inner_exception Exception: The inner exception
:param args object: The Exception arguments
"""
self.message = message
self.inner_exception = inner_exception
super().__init__(args)
class WbdutilValueError(Exception):
def __init__(self, *args):
"""
Create a new instance of a WbdutilValueError
:param *args object: The Exception arguments
"""
super().__init__(args)
from typing import List
class WbdutilException(Exception):
"""
Base class for exceptions raised within Wdutil
"""
def __init__(self, *args):
"""
Create a new instance of a WdutilException
:param *args object: The Exception arguments
"""
super().__init__(args)
class WbdutilValidationException(Exception):
"""
General validation exception raised within Wdutil
"""
def __init__(self, error_messages: List[str], *args):
"""
Create a new instance of a WdutilValidationException
:param error_messages List[str]: The validation errors
:param *args object: The Exception arguments
"""
self._error_messages = error_messages
super().__init__(args)
@property
def error_messages(self) -> List[str]:
"""
Gets the list of validation errors
return: the validation errors
rtype: List[str]
"""
return self._error_messages
class WbdutilAttributeError(WbdutilException):
"""
Common exception class for attribute errors raised in the Wbdutil
"""
def __init__(self, message: str, inner_exception: Exception, *args):
"""
Create a new instance of a ExtendedPropertiesAttributeError
:param str message: The exception message
:param inner_exception Exception: The inner exception
:param args object: The Exception arguments
"""
self.message = message
self.inner_exception = inner_exception
super().__init__(args)
class WbdutilValueError(Exception):
def __init__(self, *args):
"""
Create a new instance of a WbdutilValueError
:param *args object: The Exception arguments
"""
super().__init__(args)
src/wbdutil/common/file_loader.py
View file @ bbd5af95
import os
from pathlib import Path
from typing import Dict, List
import lasio
from lasio.las import LASFile
from knack.log import get_logger
from abc import ABC, abstractmethod
import json
logger = get_logger(__name__)
class IFileLoader(ABC):
@abstractmethod
def load(self, filepath: str) -> str:
"""
Open and load a file
:param path: str The path and filename of the data.
:return: the contents of the file
:rtype: str
:raises FileNotFoundError: If the file does not exist.
"""
pass
class IDictionaryLoader(ABC):
@abstractmethod
def load(self, path: str) -> Dict[str, any]:
"""
Load and parse a json file
:param path str : The path and filename of the file to load.
"""
pass
class LocalFileLoader(IFileLoader):
def load(self, path: str) -> str:
"""
Open and load a file from the local file system,
:param str path: The path and filename of the data.
:return: the contents of the file
:rtype: str
:raises FileNotFoundError: If the file does not exist.
"""
if not os.path.isfile(path):
raise FileNotFoundError(f'The file "{path}" does not exist')
with open(path, 'r') as file:
return file.read()
class FileValidationError(Exception):
"""
Custom error for file validation.
"""
def __init__(self, message="File must have a valid Well Name populated."):
super().__init__(message)
class JsonLoader(IDictionaryLoader):
def __init__(self, file_loader: IFileLoader):
"""
Construct a new instance of JsonLoader.
:param file_loader IFileLoader : an instance of a file loader.
"""
self._file_loader = file_loader
def load(self, path: str) -> Dict[str, any]:
"""
Load and parse a json file
:param path str : The path and filename of the file to load.
"""
return json.loads(self._file_loader.load(path))
class LasParser:
def __init__(self, file_loader: IFileLoader):
"""
Construct a new instance of LasParser.
:param file Union[IO, str]: Either a filename (inc. path), an open file object, or a string containing the contents of a file.
"""
self._file_loader = file_loader
def validate_las_file(self, las: LASFile):
"""
Validate that the Well Name attribute of the inputted LAS file has been populated.
:param las LASFile: The LASFile object to be validated.
"""
well_name = las.well.WELL.value
if not well_name or well_name == " ":
raise FileValidationError
def validate_las_file_against_record(self, las: LASFile, record_well_name: str, rec_curve_mnemonics: List[str]):
"""
More extensive validation of a LAS file if it is to be used to write data to an existing record (as
opposed to creating new wellbore and well log records).
In addition to confirming the Well Name attribute of the LAS file has been populated, the well
name and curves are validated against the existing Wellbore and Well Log records respectively.
:param las LASFile: The LASFile object to be validated.
:param record_well_name str: Well name associated with the well log record that is to have data written to it.
:param rec_curve_mnemonics List[str]: List of available curves in well log record.
"""
self.validate_las_file(las)
las_well_name = las.well.WELL.value
if not las_well_name == record_well_name:
raise FileValidationError("Well name associated with well log record does not match well name in LAS file.")
las_curve_mnemonics = [curve.mnemonic for curve in las.curves] if las.curves else []
if not all(curve in rec_curve_mnemonics for curve in las_curve_mnemonics):
raise FileValidationError("Curves available in well log record do not match those in LAS file.")
def load_las_file(self, path: str) -> LASFile:
"""
Parse a LAS format data into a LASFile object.
:param file str: A path and filename of a las file.
:return: A LASFile object.
"""
las = lasio.read(self._file_loader.load(path))
try:
self.validate_las_file(las)
except FileValidationError as e:
print(f"LAS file validation failed: {str(e)}")
raise e
except Exception as e:
print(f"Unexpected error validating file: {str(e)}")
raise e
return las
class FileUtilities:
@staticmethod
def get_filenames(path: Path, suffix: str = '') -> List[Path]:
"""
Get the paths of the files in a path and with a suffix.
If Path itself is a file returns a single item List that contains Path.
If Path is a directory return a List of all the paths of all the files in the directory with suffix.
:param Path path: A director or file path
:return: A List of paths.
:rtype: List[Path]
"""
if path.is_file():
return [path]
elif path.is_dir():
return list(path.glob(f'**/*{suffix}'))
else:
logger.error(
"Invalid input paths. Please confirm you have provided the required inputs correctly."
)
import os
from pathlib import Path
from typing import Dict, List
import lasio
from lasio.las import LASFile
from knack.log import get_logger
from abc import ABC, abstractmethod
import json
logger = get_logger(__name__)
class IFileLoader(ABC):
@abstractmethod
def load(self, filepath: str) -> str:
"""
Open and load a file
:param path: str The path and filename of the data.
:return: the contents of the file
:rtype: str
:raises FileNotFoundError: If the file does not exist.
"""
pass
class IDictionaryLoader(ABC):
@abstractmethod
def load(self, path: str) -> Dict[str, any]:
"""
Load and parse a json file
:param path str : The path and filename of the file to load.
"""
pass
class LocalFileLoader(IFileLoader):
def load(self, path: str) -> str:
"""
Open and load a file from the local file system,
:param str path: The path and filename of the data.
:return: the contents of the file
:rtype: str
:raises FileNotFoundError: If the file does not exist.
"""
if not os.path.isfile(path):
raise FileNotFoundError(f'The file "{path}" does not exist')
with open(path, 'r') as file:
return file.read()
class FileValidationError(Exception):
"""
Custom error for file validation.
"""
def __init__(self, message="File must have a valid Well Name populated."):
super().__init__(message)
class JsonLoader(IDictionaryLoader):
def __init__(self, file_loader: IFileLoader):
"""
Construct a new instance of JsonLoader.
:param file_loader IFileLoader : an instance of a file loader.
"""
self._file_loader = file_loader
def load(self, path: str) -> Dict[str, any]:
"""
Load and parse a json file
:param path str : The path and filename of the file to load.
"""
return json.loads(self._file_loader.load(path))
class LasParser:
def __init__(self, file_loader: IFileLoader):
"""
Construct a new instance of LasParser.
:param file Union[IO, str]: Either a filename (inc. path), an open file object, or a string containing the contents of a file.
"""
self._file_loader = file_loader
def validate_las_file(self, las: LASFile):
"""
Validate that the Well Name attribute of the inputted LAS file has been populated.
:param las LASFile: The LASFile object to be validated.
"""
well_name = las.well.WELL.value
if not well_name or well_name == " ":
raise FileValidationError
def validate_las_file_against_record(self, las: LASFile, record_well_name: str, rec_curve_mnemonics: List[str]):
"""
More extensive validation of a LAS file if it is to be used to write data to an existing record (as
opposed to creating new wellbore and well log records).
In addition to confirming the Well Name attribute of the LAS file has been populated, the well
name and curves are validated against the existing Wellbore and Well Log records respectively.
:param las LASFile: The LASFile object to be validated.
:param record_well_name str: Well name associated with the well log record that is to have data written to it.
:param rec_curve_mnemonics List[str]: List of available curves in well log record.
"""
self.validate_las_file(las)
las_well_name = las.well.WELL.value
if not las_well_name == record_well_name:
raise FileValidationError("Well name associated with well log record does not match well name in LAS file.")
las_curve_mnemonics = [curve.mnemonic for curve in las.curves] if las.curves else []
if not all(curve in rec_curve_mnemonics for curve in las_curve_mnemonics):
raise FileValidationError("Curves available in well log record do not match those in LAS file.")
def load_las_file(self, path: str) -> LASFile:
"""
Parse a LAS format data into a LASFile object.
:param file str: A path and filename of a las file.
:return: A LASFile object.
"""
las = lasio.read(self._file_loader.load(path))
try:
self.validate_las_file(las)
except FileValidationError as e:
print(f"LAS file validation failed: {str(e)}")
raise e
except Exception as e:
print(f"Unexpected error validating file: {str(e)}")
raise e
return las
class FileUtilities:
@staticmethod
def get_filenames(path: Path, suffix: str = '') -> List[Path]:
"""
Get the paths of the files in a path and with a suffix.
If Path itself is a file returns a single item List that contains Path.
If Path is a directory return a List of all the paths of all the files in the directory with suffix.
:param Path path: A director or file path
:return: A List of paths.
:rtype: List[Path]
"""
if path.is_file():
return [path]
elif path.is_dir():
return list(path.glob(f'**/*{suffix}'))
else:
logger.error(
"Invalid input paths. Please confirm you have provided the required inputs correctly."
)
src/wbdutil/common/reflection_utilities.py
View file @ bbd5af95
from typing import Any, Dict, List
from knack.log import get_logger
from .exceptions import WbdutilAttributeError, WbdutilValueError
logger = get_logger(__name__)
class ReflectionHelper:
def getattr_recursive(source: any, source_field_heirachy: List[str]) -> any:
"""
Get the specified field from the source object in a recursive manner.
:param any destination: The source object
:param List[str] source_field_heirachy: List of fields, from lowest to the deepest in the heirachy.
:return: The data value.
:rtype: any
"""
field_name = source_field_heirachy.pop(0)
try:
if isinstance(source, dict):
subfield = source[field_name]
else:
subfield = getattr(source, field_name)
except (KeyError, AttributeError):
if field_name.endswith(']') and '[' in field_name:
# probably a reference to a list item
logger.debug(f"Attempting to parse the mapping field {field_name} as an array")
subfield = ReflectionHelper._getattr_list(source, field_name)
else:
message = f"The attribute '{field_name}' was not found in the source data. An empty value for this field will be returned."
logger.error(message)
return None
if len(source_field_heirachy) == 0:
return subfield
else:
return ReflectionHelper.getattr_recursive(subfield, source_field_heirachy)
def _getattr_list(source: any, source_field_name: str) -> any:
try:
segments = source_field_name.split('[')
if isinstance(source, dict):
subfield = source[segments.pop(0)]
else:
subfield = getattr(source, segments.pop(0))
for ind in segments:
subfield = subfield[int(ind.rstrip(']'))]
return subfield
except (KeyError, AttributeError, ValueError) as ex:
message = f"The attribute '{source_field_name}' could not be parsed as an array"
logger.error(message)
raise WbdutilAttributeError(message, ex)
def setattr_recursive(value: any, destination: Dict[str, Any], destination_field_heirachy: List[str]) -> None:
"""
Set the specified field to the given value, recursively create member objects where needed. Update in place.
:param any value: The data value
:param any destination: The destination object
:param List[str] destination_field_heirachy: List of fields in the hierarchy, ordered from top to bottom.
"""
if destination is None:
message = "The destination dictionary must not be null"
logger.critical(message)
raise WbdutilValueError(message)
field_name = destination_field_heirachy.pop(0)
if len(destination_field_heirachy) == 0:
destination[field_name] = value
return destination
if field_name not in destination:
destination[field_name] = {}
ReflectionHelper.setattr_recursive(value, destination[field_name], destination_field_heirachy)
from typing import Any, Dict, List
from knack.log import get_logger
from .exceptions import WbdutilAttributeError, WbdutilValueError
logger = get_logger(__name__)
class ReflectionHelper:
def getattr_recursive(source: any, source_field_heirachy: List[str]) -> any:
"""
Get the specified field from the source object in a recursive manner.
:param any destination: The source object
:param List[str] source_field_heirachy: List of fields, from lowest to the deepest in the heirachy.
:return: The data value.
:rtype: any
"""
field_name = source_field_heirachy.pop(0)
try:
if isinstance(source, dict):
subfield = source[field_name]
else:
subfield = getattr(source, field_name)
except (KeyError, AttributeError):
if field_name.endswith(']') and '[' in field_name:
# probably a reference to a list item
logger.debug(f"Attempting to parse the mapping field {field_name} as an array")
subfield = ReflectionHelper._getattr_list(source, field_name)
else:
message = f"The attribute '{field_name}' was not found in the source data. An empty value for this field will be returned."
logger.error(message)
return None
if len(source_field_heirachy) == 0:
return subfield
else:
return ReflectionHelper.getattr_recursive(subfield,</