from typing import Optional, List, Literal, TYPE_CHECKING, Any, Union, Tuple, Type, IO, Dict, cast
import numpy as np
from geogeometry.shared.observer.ObserverTag import ObserverTag
from geoassistant.drillholes.DrillholeDataCollection import DrillholeDataCollection
from geoassistant.drillholes.CollectionParametersHandler import CollectionParametersHandler
from geoassistant.drillholes.intervalscollection_components.IntervalsCollectionProperties import IntervalsCollectionProperties
from geoassistant.drillholes.intervalscollection_components.IntervalsCollectionReader import IntervalsCollectionReader
from geoassistant.drillholes.intervalscollection_components.IntervalsCollectionGeneralQuerier import IntervalsCollectionGeneralQuerier
from geoassistant.drillholes.intervalscollection_components.IntervalsCollectionSubSetter import IntervalsCollectionSubSetter
from geoassistant.drillholes.intervalscollection_components.IntervalsCollectionExporter import IntervalsCollectionExporter
from geoassistant.drillholes.intervalscollection_components.IntervalsCollectionDataProfiler import IntervalsCollectionDataProfiler
from geoassistant.drillholes.intervalscollection_components.IntervalsCollectionHistogrammer import IntervalsCollectionHistogrammer
from geoassistant.lab_testing.plt.PLTTestsCollection import PLTTestsCollection
from geoassistant.statistics.ParameterStatsObject import ParameterStatsObject
from geoassistant.statistics.correlation.CorrelationPlotFactory import CorrelationPlotFactory
from geoassistant.utils.collections import from_list_to_collection
from geoassistant.drillholes.LoggedDiscontinuitiesCollection import LoggedDiscontinuitiesCollection
if TYPE_CHECKING:
import geogeometry
from geoassistant.drillholes.Interval import Interval
from geoassistant.drillholes.Drillhole import Drillhole
from geoassistant.drillholes.DrillholesCollection import DrillholesCollection
from geoassistant.drillholes.LoggedMajorStructuresCollection import LoggedMajorStructuresCollection
from geoassistant.block_model.BlockModel import BlockModel
[docs]
class IntervalsCollection(DrillholeDataCollection['IntervalsCollection', 'Interval'], IntervalsCollectionProperties):
"""
A collection of Interval objects with filtering and querying utilities.
Subsetter methods return new lists of Interval objects and never
mutate the original collection.
"""
def __init__(self, name: Optional[str] = None):
super().__init__(name=name)
self.subsetter: IntervalsCollectionSubSetter = IntervalsCollectionSubSetter(self)
self.general_querier: IntervalsCollectionGeneralQuerier = IntervalsCollectionGeneralQuerier(self)
self.parameters_handler: CollectionParametersHandler = CollectionParametersHandler(self)
self.exporter: IntervalsCollectionExporter = IntervalsCollectionExporter(self)
self.data_profiler: IntervalsCollectionDataProfiler = IntervalsCollectionDataProfiler(self)
self.histogrammer: IntervalsCollectionHistogrammer = IntervalsCollectionHistogrammer(self)
[docs]
def checkCalculations(self) -> None:
for interval in self:
interval.checkCalculations()
def __getitem__(self, identifier: Union[int, str]) -> Any:
if isinstance(identifier, (int, np.integer)):
return self.elements[int(identifier)]
else:
return self.getParameterValues(parameter_id=identifier)
[docs]
def addInterval(self, interval: 'Interval') -> None:
self.addElement(element=interval)
[docs]
def getLoggedDiscontinuities(self) -> LoggedDiscontinuitiesCollection:
dc = LoggedDiscontinuitiesCollection()
for interval in self:
if interval.getDiscontinuities() is not None:
dc += interval.getDiscontinuities()
return dc
[docs]
def getMajorStructures(self) -> 'LoggedMajorStructuresCollection':
from geoassistant.drillholes.LoggedMajorStructuresCollection import LoggedMajorStructuresCollection
m_structs = LoggedMajorStructuresCollection()
for interval in self:
if interval.getMajorStructures() is not None:
m_structs += interval.getMajorStructures()
return m_structs
[docs]
def setDrillhole(self, drillhole: 'Drillhole') -> None:
for interval in self:
interval.drillhole = drillhole # Direct assignment does NOT run single geometry calculations
all_depths = np.array([d for interval in self for d in (interval.getFrom(), interval.getTo())])
points = drillhole.calculatePointsBySurveyDepths(depths=all_depths)
for i, interval in enumerate(self):
interval.setNodes(nodes=points[[2*i, 2*i + 1]])
[docs]
def assignDrillholes(self, drillholes: 'DrillholesCollection') -> None:
IntervalsCollectionReader.assignIntervalsToDrillholes(intervals=self, drillholes=drillholes)
[docs]
def assignDiscontinuities(self, discontinuities: 'LoggedDiscontinuitiesCollection') -> None:
for interval in self:
subset = discontinuities.getSubsetBetweenDepths(d0=interval.getFrom(), d1=interval.getTo())
if len(subset):
interval.setDiscontinuities(discontinuities=subset)
[docs]
def assignMajorStructures(self, major_structures: 'LoggedMajorStructuresCollection') -> None:
for interval in self:
subset = major_structures.getSubsetBetweenDepths(d0=interval.getFrom(), d1=interval.getTo())
if len(subset):
interval.setMajorStructures(major_structures=subset)
[docs]
def assignPLTTests(self, plt_tests: 'PLTTestsCollection') -> None:
for _plt in plt_tests:
for interval in self:
if interval.getFrom() <= _plt.getDepth() < interval.getTo():
if interval.getPLTTests() is None:
interval.setPLTTests(plt_tests=PLTTestsCollection())
interval.getPLTTests().addElement(_plt)
break
else:
if _plt.getDepth() == self[-1].getTo():
if self[-1].getPLTTests() is None:
self[-1].setPLTTests(plt_tests=PLTTestsCollection())
self[-1].getPLTTests().addElement(_plt)
[docs]
def assignSystemsJointCondition(self, method: Literal["worst"]) -> None:
for interval in self:
interval.assignSystemsJointCondition(method=method)
[docs]
def getPLTTests(self) -> PLTTestsCollection:
all_plts = PLTTestsCollection()
for interval in self:
if interval.getPLTTests() is not None:
all_plts += interval.getPLTTests()
return all_plts
[docs]
def getMinimumFrom(self) -> float:
return min([interval.getFrom() for interval in self])
[docs]
def getMaximumTo(self) -> float:
return max([interval.getTo() for interval in self])
[docs]
def calculateJointFrequencyFromDiscontinuities(self, method: Literal["core_correction", "raw"],
considered_types: Tuple[Literal['J', 'CJ', 'FJ']] = ('J', 'FJ')) -> None:
"""
Assigns FF to intervals based on its registered LoggedDiscontinuitiesCollection.
"""
for interval in self:
interval.calculateJointFrequencyFromDiscontinuities(method=method, considered_types=considered_types)
[docs]
def findIntervalByCredentials(self, drillhole_name: str, _from: float, _to: float) -> Optional['Interval']:
dh_intervals = cast("IntervalsCollection", self.getSubsetByDrillholeName(drillhole_name=drillhole_name))
return dh_intervals.findIntervalByFromTo(_from=_from, _to=_to)
[docs]
def findIntervalByFromTo(self, _from: float, _to: float) -> Optional['Interval']:
for interval in self:
if interval.getFrom() == _from and interval.getTo() == _to:
return interval
return None
[docs]
@from_list_to_collection
def getSubsetByDrillholeName(self, drillhole_name: str) -> List['Interval']:
return self.subsetter.getSubsetByDrillholeName(drillhole_name=drillhole_name)
[docs]
@from_list_to_collection
def getSubsetFromParameterRange(self, parameter_id: str, parameter_range: List[Optional[float]],
include_start: bool = False, include_end: bool = False) -> List['Interval']:
"""
Return intervals whose parameter value falls within a given range.
This method filters the collection by evaluating the value
associated with ``parameter_id`` for each interval and selecting
those that lie within the specified numeric bounds.
Open bounds can be expressed using ``None``.
Args:
parameter_id:
Identifier of the parameter to evaluate.
parameter_range:
A two-element list ``[min, max]`` defining the range.
Use ``None`` to indicate an open lower or upper bound.
include_start:
Whether the lower bound is inclusive.
include_end:
Whether the upper bound is inclusive.
Returns:
A list of matching ``Interval`` objects.
See Also:
getSubsetFromParameterValue
getSubsetFromBounds
"""
return self.subsetter.getSubsetFromParameterRange(parameter_id=parameter_id, parameter_range=parameter_range,
include_start=include_start, include_end=include_end)
[docs]
@from_list_to_collection
def getSubsetByParameterDefinition(self, parameter_id: str) -> List['Interval']:
return self.subsetter.getSubsetByParameterDefinition(parameter_id=parameter_id)
[docs]
@from_list_to_collection
def getSubsetFromParameterValue(self, parameter_id: str, value: Any) -> List['Interval']:
"""
Returns a subset of intervals matching a specific parameter value.
This method filters the collection based on the value associated with
the specified parameter identifier and returns the matching intervals.
Args:
parameter_id (str): Identifier of the parameter used for filtering.
value (Any): Value used as the filter criterion.
Returns:
List[Interval]: List of intervals matching the specified parameter value.
"""
return self.subsetter.getSubsetFromParameterValue(parameter_id=parameter_id, value=value)
[docs]
@from_list_to_collection
def getSubsetWithDiscontinuities(self) -> List['Interval']:
return self.subsetter.getSubsetWithDiscontinuities()
[docs]
@from_list_to_collection
def getSubsetWithMajorStructures(self) -> List['Interval']:
return self.subsetter.getSubsetWithMajorStructures()
[docs]
@from_list_to_collection
def filterByMajorStructures(self, limit_length_percentage: float) -> List['Interval']:
return self.subsetter.filterByMajorStructures(limit_length_percentage=limit_length_percentage)
[docs]
@from_list_to_collection
def getSubsetFromPlaneOffset(self, plane: 'geogeometry.Plane', offset: float) -> List['Interval']:
return self.subsetter.getSubsetFromPlaneOffset(plane=plane, offset=offset)
[docs]
def createParameterHistogram(self, parameter_id: str, frequency_type: Literal['meters', 'intervals'] = 'meters',
savepath: Optional[str] = None) -> None:
self.histogrammer.createParameterHistogram(parameter_id=parameter_id, frequency_type=frequency_type, savepath=savepath)
[docs]
def createCorrelationPlot(self, x_parameter: str, y_parameter: str) -> None:
plot = CorrelationPlotFactory.createCorrelationPlot(x_parameter=x_parameter, y_parameter=y_parameter)
plot.setXParameterValues(values=self.getParameterValues(parameter_id=x_parameter))
plot.setYParameterValues(values=self.getParameterValues(parameter_id=y_parameter))
plot.show()
[docs]
def plotParameterAlongAxis(self, parameter_id: str, axis: Literal['x', 'y', 'z'],
y_limits: Optional[Tuple[float, float]] = None) -> None:
self.data_profiler.plotParameterAlongAxis(parameter_id=parameter_id, axis=axis, y_limits=y_limits)
[docs]
@classmethod
def readIntervalsExcelSheet(cls: Type['IntervalsCollection'],
source: Union[str, IO[bytes]],
sheetname: str,
from_col: str, to_col: str,
min_col: str = 'A', max_col: str = None,
min_row: int = 2, max_row: int = None,
id_col: Optional[str] = None, hole_id: Optional[str] = None,
intervals_name: Optional[str] = None,
drillholes: Optional['DrillholesCollection'] = None) -> 'IntervalsCollection':
return IntervalsCollectionReader.readIntervalsExcelSheet(cls, source=source, sheetname=sheetname,
from_col=from_col, to_col=to_col,
min_col=min_col, max_col=max_col,
min_row=min_row, max_row=max_row,
id_col=id_col, hole_id=hole_id,
intervals_name=intervals_name,
drillholes=drillholes)
[docs]
@classmethod
def readIntervalsCsvFile(cls,
source: Union[str, IO[str]],
from_key: str, to_key: str,
id_key: Optional[str] = None, hole_id: Optional[str] = None,
intervals_name: Optional[str] = None,
drillholes: Optional['DrillholesCollection'] = None) -> 'IntervalsCollection':
return IntervalsCollectionReader.readIntervalsCsvFile(cls, source=source,
from_key=from_key, to_key=to_key,
id_key=id_key, hole_id=hole_id,
intervals_name=intervals_name,
drillholes=drillholes)
[docs]
def export(self, savepath: str, parameters: Optional[List[str]] = None, metadata_keys: Optional[List[str]] = None) -> None:
self.exporter.export(savepath=savepath, parameters=parameters, metadata_keys=metadata_keys)
[docs]
def onElementChange(self, interval: 'Interval', **kwargs) -> None:
self.notifyObservers(tags=[ObserverTag.CONTAINER])
[docs]
def getParameterStats(self, parameter_id: str, name: Optional[str] = None) -> ParameterStatsObject:
stats = ParameterStatsObject(parameter_id=parameter_id, name=name)
stats.setData(data=self.getParameterValues(parameter_id=parameter_id), weights=self.getLengths())
return stats
[docs]
def getParameterRepresentativeValue(self, parameter_id: str, statistic: str = 'average') -> Optional[float]:
if len(self):
stats = self.getParameterStats(parameter_id=parameter_id)
if not len(stats.getFilteredData()):
return np.nan
if statistic == 'average':
if "QBarton" in parameter_id:
return stats.getStatsDict()['avg_from_log']
else:
return stats.getStatsDict()['average']
else:
raise ValueError("TO IMPLEMENT")
else:
return np.nan
