fractopo.general module

Contains general calculation and plotting tools.

class fractopo.general.Col(*values)

Bases: Enum

GeoDataFrame column names for attributes.

AZIMUTH = 'azimuth'

AZIMUTH_SET = 'azimuth_set'

LENGTH = 'length'

LENGTH_NON_WEIGHTED = 'length non-weighted'

LENGTH_SET = 'length_set'

LENGTH_WEIGHTS = 'boundary_weight'

class fractopo.general.Param(*values)

Bases: Enum

Column names for geometric and topological parameters.

The ParamInfo instances contain additional metadata.

AREA = ParamInfo(name='Area', plot_as_log=False, unit='$m^2$', needs_topology=False, aggregator=<function sum_aggregation>)

AREAL_FREQUENCY_B20 = ParamInfo(name='Areal Frequency B20', plot_as_log=True, unit='$\\frac{1}{m^2}$', needs_topology=True, aggregator=<function mean_aggregation>)

AREAL_FREQUENCY_P20 = ParamInfo(name='Areal Frequency P20', plot_as_log=True, unit='$\\frac{1}{m^2}$', needs_topology=True, aggregator=<function mean_aggregation>)

BRANCH_MAX_LENGTH = ParamInfo(name='Branch Max Length', plot_as_log=True, unit='$m$', needs_topology=True, aggregator=<function mean_aggregation>)

BRANCH_MEAN_LENGTH = ParamInfo(name='Branch Mean Length', plot_as_log=True, unit='$m$', needs_topology=True, aggregator=<function mean_aggregation>)

BRANCH_MIN_LENGTH = ParamInfo(name='Branch Min Length', plot_as_log=True, unit='$m$', needs_topology=True, aggregator=<function mean_aggregation>)

CIRCLE_COUNT = ParamInfo(name='Circle Count', plot_as_log=False, unit='-', needs_topology=False, aggregator=<function sum_aggregation>)

CONNECTIONS_PER_BRANCH = ParamInfo(name='Connections per Branch', plot_as_log=False, unit='$\\frac{1}{n}$', needs_topology=True, aggregator=<function mean_aggregation>)

CONNECTIONS_PER_TRACE = ParamInfo(name='Connections per Trace', plot_as_log=False, unit='$\\frac{1}{n}$', needs_topology=True, aggregator=<function mean_aggregation>)

CONNECTION_FREQUENCY = ParamInfo(name='Connection Frequency', plot_as_log=False, unit='$\\frac{1}{m^2}$', needs_topology=True, aggregator=<function mean_aggregation>)

DIMENSIONLESS_INTENSITY_B22 = ParamInfo(name='Dimensionless Intensity B22', plot_as_log=False, unit='-', needs_topology=True, aggregator=<function mean_aggregation>)

DIMENSIONLESS_INTENSITY_P22 = ParamInfo(name='Dimensionless Intensity P22', plot_as_log=False, unit='-', needs_topology=False, aggregator=<function mean_aggregation>)

FRACTURE_DENSITY_MAULDON = ParamInfo(name='Fracture Density (Mauldon)', plot_as_log=True, unit='$\\frac{1}{m^2}$', needs_topology=True, aggregator=<function mean_aggregation>)

FRACTURE_INTENSITY_B21 = ParamInfo(name='Fracture Intensity B21', plot_as_log=True, unit='$\\frac{m}{m^2}$', needs_topology=False, aggregator=<function mean_aggregation>)

FRACTURE_INTENSITY_MAULDON = ParamInfo(name='Fracture Intensity (Mauldon)', plot_as_log=True, unit='$\\frac{m}{m^2}$', needs_topology=True, aggregator=<function mean_aggregation>)

FRACTURE_INTENSITY_P21 = ParamInfo(name='Fracture Intensity P21', plot_as_log=True, unit='$\\frac{m}{m^2}$', needs_topology=False, aggregator=<function mean_aggregation>)

NUMBER_OF_BRANCHES = ParamInfo(name='Number of Branches', plot_as_log=False, unit='-', needs_topology=True, aggregator=<function sum_aggregation>)

NUMBER_OF_BRANCHES_TRUE = ParamInfo(name='Number of Branches (Real)', plot_as_log=False, unit='-', needs_topology=True, aggregator=<function sum_aggregation>)

NUMBER_OF_TRACES = ParamInfo(name='Number of Traces', plot_as_log=False, unit='-', needs_topology=True, aggregator=<function sum_aggregation>)

NUMBER_OF_TRACES_TRUE = ParamInfo(name='Number of Traces (Real)', plot_as_log=False, unit='-', needs_topology=True, aggregator=<function sum_aggregation>)

TRACE_MAX_LENGTH = ParamInfo(name='Trace Max Length', plot_as_log=True, unit='$m$', needs_topology=False, aggregator=<function mean_aggregation>)

TRACE_MEAN_LENGTH = ParamInfo(name='Trace Mean Length', plot_as_log=True, unit='$m$', needs_topology=False, aggregator=<function mean_aggregation>)

TRACE_MEAN_LENGTH_MAULDON = ParamInfo(name='Trace Mean Length (Mauldon)', plot_as_log=True, unit='$m$', needs_topology=True, aggregator=<function mean_aggregation>)

TRACE_MIN_LENGTH = ParamInfo(name='Trace Min Length', plot_as_log=True, unit='$m$', needs_topology=False, aggregator=<function mean_aggregation>)

class fractopo.general.ParamInfo(name: str, plot_as_log: bool, unit: str, needs_topology: bool, aggregator: Callable)

Bases: object

Parameter with name and metadata.

aggregator: Callable

name: str

needs_topology: bool

plot_as_log: bool

unit: str

class fractopo.general.ProcessResult(identifier: str, result: Any, error: bool)

Bases: object

Dataclass for multiprocessing result parsing.

error: bool

identifier: str

result: Any

fractopo.general.assign_branch_and_node_colors(feature_type: str) → str

Determine color for each branch and node type.

Defaults to red.

>>> assign_branch_and_node_colors("C-C")
'red'

fractopo.general.avg_calc(data)

Calculate average for radial data.

TODO: Should take length into calculations. Not real average atm

fractopo.general.azimu_half(degrees: float) → float

Transform azimuth from 180-360 range to range 0-180.

Parameters:: degrees – Degrees in range 0 - 360
Returns:: Degrees in range 0 - 180

fractopo.general.azimuth_to_unit_vector(azimuth: float) → ndarray: Convert azimuth to unit vector.

fractopo.general.bool_arrays_sum(arr_1: ndarray, arr_2: ndarray) → ndarray

Calculate integer sum of two arrays.

Resulting array consists only of integers 0, 1 and 2.

>>> arr_1 = np.array([True, False, False])
>>> arr_2 = np.array([True, True, False])
>>> bool_arrays_sum(arr_1, arr_2)
array([2, 1, 0])

>>> arr_1 = np.array([True, True])
>>> arr_2 = np.array([True, True])
>>> bool_arrays_sum(arr_1, arr_2)
array([2, 2])

fractopo.general.bounding_polygon(geoseries: GeoSeries | GeoDataFrame) → Polygon

Create bounding polygon around GeoSeries.

The geoseries geometries will always be completely enveloped by the polygon. The geometries will not intersect the polygon boundary.

>>> geom = LineString([(1, 0), (1, 1), (-1, -1)])
>>> geoseries = gpd.GeoSeries([geom])
>>> poly = bounding_polygon(geoseries)
>>> poly.wkt
'POLYGON ((2 -2, 2 2, -2 2, -2 -2, 2 -2))'
>>> geoseries.intersects(poly.boundary)
0    False
dtype: bool

fractopo.general.calc_circle_area(radius: float | int | Real) → float

Calculate area of circle.

>>> calc_circle_area(1.78)
9.953822163633902

fractopo.general.calc_circle_radius(area: float | int | Real) → float

Calculate radius from area.

>>> calc_circle_radius(10.0)
1.7841241161527712

fractopo.general.calc_strike(dip_direction: float) → float

Calculate strike from dip direction. Right-handed rule.

E.g.:

>>> calc_strike(50.0)
320.0

>>> calc_strike(180.0)
90.0

Parameters:: dip_direction – The direction of dip.
Returns:: Converted strike.

fractopo.general.check_for_wrong_geometries(traces: GeoDataFrame, area: GeoDataFrame): Check that traces are line geometries and area contains area geometries.

fractopo.general.check_for_z_coordinates(geodata: GeoDataFrame | GeoSeries) → bool: Check if geopandas data contains Z-coordinates.

fractopo.general.compare_unit_vector_orientation(vec_1: ndarray, vec_2: ndarray, threshold_angle: float | int | Real) → bool: If vec_1 and vec_2 are too different in orientation, will return False.

fractopo.general.convert_list_columns(gdf: GeoDataFrame, allow: bool = True) → GeoDataFrame: Convert list type columns to string.

fractopo.general.create_unit_vector(start_point: Point, end_point: Point) → ndarray

Create numpy unit vector from two shapely Points.

Parameters:

start_point – The start point.
end_point – The end point.

Returns:

The unit vector that points from start_point to end_point.

fractopo.general.define_length_set(length: float, set_df: DataFrame) → str: Define sets based on the length of the traces or branches.

fractopo.general.determine_azimuth(line: LineString, halved: bool) → float

Calculate azimuth of given line.

If halved -> return is in range [0, 180] Else -> [0, 360]

e.g.: Accepts LineString

>>> determine_azimuth(LineString([(0, 0), (1, 1)]), True)
45.0

>>> determine_azimuth(LineString([(0, 0), (0, 1)]), True)
0.0

>>> determine_azimuth(LineString([(0, 0), (-1, -1)]), False)
225.0

>>> determine_azimuth(LineString([(0, 0), (-1, -1)]), True)
45.0

Parameters:

line – The line of which azimuth is determined.
halved – Whether to return result in range [0, 180] (halved=True) or [0, 360] (halved=False).

Returns:

The determined azimuth.

fractopo.general.determine_boundary_intersecting_lines(line_gdf: GeoDataFrame, area_gdf: GeoDataFrame, snap_threshold: float) → tuple[ndarray, ndarray]: Determine lines that intersect any target area boundary.

fractopo.general.determine_regression_azimuth(line: LineString) → float

Determine azimuth of line LineString with linear regression.

A scikit-learn LinearRegression is fitted to the x, y coordinates of the given and the azimuth of the fitted linear line is returned.

The azimuth is returned in range [0, 180].

E.g.

>>> line = LineString([(0, 0), (1, 1), (2, 2), (3, 3)])
>>> determine_regression_azimuth(line)
45.0

>>> line = LineString([(-1, -5), (3, 3)])
>>> round(determine_regression_azimuth(line), 3)
26.565

>>> line = LineString([(0, 0), (0, 3)])
>>> determine_regression_azimuth(line)
0.0

Parameters:: line – The line of which azimuth is determined.
Returns:: The determined azimuth in range [0, 180].
Raises:: ValueError – When LinearRegression returns unexpected coefficients.

fractopo.general.determine_set(value: float | int | Real, value_ranges: Sequence[tuple[float | int | Real, float | int | Real]], set_names: Sequence[str], loop_around: bool) → str

Determine which named value range, if any, value is within.

loop_around defines behavior expected for radial data i.e. when value range can loop back around e.g. [160, 50]

E.g.

>>> determine_set(10.0, ((0, 20), (30, 160)), ("0-20", "30-160"), False)
'0-20'

Example with

>>> determine_set(50.0, ((0, 20), (160, 60)), ("0-20", "160-60"), True)
'160-60'

Parameters:

value – Value to determine set of.
value_ranges – Ranges of each set.
set_names – Names of each set.
loop_around – Whether the sets loop around. This is the case for radial data such as azimuths but not the case for length data.

Returns:

Set string in which value belongs.

Raises:

ValueError – When set value ranges overlap.

fractopo.general.determine_valid_intersection_points(intersection_geoms: GeoSeries) → list[Point]

Filter intersection points between trace candidates and geom.

Only allows Point geometries as intersections. LineString intersections would be possible if geometries are stacked.

Parameters:: intersection_geoms – GeoSeries of intersection (Point) geometries.
Returns:: The valid intersections (Points).

fractopo.general.determine_valid_intersection_points_no_vnode(trace_candidates: GeoSeries, geom: LineString) → list[Point]

Filter intersection points between trace candidates and geom with no vnodes.

V-node intersections are validated by looking at the endpoints. If V-nodes were kept as intersection points the VNodeValidator could not find V-node errors.

fractopo.general.dissolve_multi_part_traces(traces: GeoDataFrame | GeoSeries) → GeoDataFrame | GeoSeries

Dissolve MultiLineStrings in GeoDataFrame or GeoSeries.

Copies all attribute data of rows with MultiLineStrings to new LineStrings.

fractopo.general.extend_bounds(min_x: float | int | Real, min_y: float | int | Real, max_x: float | int | Real, max_y: float | int | Real, extend_amount: float | int | Real) → tuple[float | int | Real, float | int | Real, float | int | Real, float | int | Real]

Extend bounds by addition and reduction.

>>> extend_bounds(0, 0, 10, 10, 10)
(-10, -10, 20, 20)

fractopo.general.fallback_aggregation(values) → str: Fallback aggregation where values are simply joined into a string.

fractopo.general.flatten_tuples(list_of_tuples: list[tuple[Any, ...]]) → tuple[list[int], list[Any]]

Flatten collection of tuples and return index references.

Indexes are from original tuple groupings.

E.g.

>>> tuples = [(1, 1, 1), (2, 2, 2, 2), (3,)]
>>> flatten_tuples(tuples)
([0, 0, 0, 1, 1, 1, 1, 2], [1, 1, 1, 2, 2, 2, 2, 3])

fractopo.general.focus_plot_to_bounds(ax: Axes, total_bounds: ndarray) → Axes: Focus plot to given bounds.

Get LineString or Polygon bounds.

>>> geom_bounds(LineString([(-10, -10), (10, 10)]))
(-10.0, -10.0, 10.0, 10.0)

fractopo.general.get_next_point_in_trace(trace: LineString, point: Point) → Point: Determine next coordinate point towards middle of LineString from point.

fractopo.general.get_trace_coord_points(trace: LineString) → list[Point]

Get all coordinate Points of a LineString.

>>> trace = LineString([(0, 0), (2, 0), (3, 0)])
>>> coord_points = get_trace_coord_points(trace)
>>> print([p.wkt for p in coord_points])
['POINT (0 0)', 'POINT (2 0)', 'POINT (3 0)']

fractopo.general.get_trace_endpoints(trace: LineString) → tuple[Point, Point]: Return endpoints (shapely.geometry.Point) of a given LineString.

fractopo.general.intersection_count_to_boundary_weight(intersection_count: int) → int

Get actual weight factor for boundary intersection count.

>>> intersection_count_to_boundary_weight(2)
0
>>> intersection_count_to_boundary_weight(0)
1
>>> intersection_count_to_boundary_weight(1)
2

fractopo.general.is_azimuth_close(first: float | int | Real, second: float | int | Real, tolerance: float | int | Real, halved: bool = True) → bool

Determine are azimuths first and second within tolerance.

Takes into account the radial nature of azimuths.

>>> is_azimuth_close(0, 179, 15)
True

>>> is_azimuth_close(166, 179, 15)
True

>>> is_azimuth_close(20, 179, 15)
False

Parameters:

first – First azimuth value to compare.
second – Second azimuth value to compare.
tolerance – Tolerance for closeness.
halved – Are the azimuths azial (i.e. halved=True) or vectors.

fractopo.general.is_empty_area(area: GeoDataFrame, traces: GeoDataFrame): Check if any traces intersect the area(s) in area GeoDataFrame.

fractopo.general.is_set(value: float | int | Real, value_range: tuple[float | int | Real, float | int | Real], loop_around: bool) → bool

Determine if value fits within the given value_range.

If the value range has the possibility of looping around loop_around can be set to true.

>>> is_set(5, (0, 10), False)
True

>>> is_set(5, (175, 15), True)
True

Parameters:

value – Value to determine.
Tuple[float, (value_range) – The range of values.
loop_around – Whether the range loops around. This is the case for radial data such as azimuths but not the case for length data.

Returns:

Is it within range.

fractopo.general.line_intersection_to_points(first: LineString, second: LineString) → list[Point]

Perform shapely intersection between two LineStrings.

Enforces only Point returns.

fractopo.general.match_crs(first: GeoSeries | GeoDataFrame, second: GeoSeries | GeoDataFrame) → tuple[GeoSeries | GeoDataFrame, GeoSeries | GeoDataFrame]

Match crs between two geopandas data structures.

>>> first = gpd.GeoSeries([Point(1, 1)], crs="EPSG:3067")
>>> second = gpd.GeoSeries([Point(1, 1)])
>>> m_first, m_second = match_crs(first, second)
>>> m_first.crs == m_second.crs
True

fractopo.general.mean_aggregation(values, weights) → float | int | Real: Aggregate by calculating mean.

fractopo.general.mls_to_ls(multilinestrings: list[MultiLineString]) → list[LineString]

Flattens a list of multilinestrings to a list of linestrings.

>>> multilinestrings = [
...     MultiLineString(
...         [
...             LineString([(1, 1), (2, 2), (3, 3)]),
...             LineString([(1.9999, 2), (-2, 5)]),
...         ]
...     ),
...     MultiLineString(
...         [
...             LineString([(1, 1), (2, 2), (3, 3)]),
...             LineString([(1.9999, 2), (-2, 5)]),
...         ]
...     ),
... ]
>>> result_linestrings = mls_to_ls(multilinestrings)
>>> print([ls.wkt for ls in result_linestrings])
['LINESTRING (1 1, 2 2, 3 3)', 'LINESTRING (1.9999 2, -2 5)',
'LINESTRING (1 1, 2 2, 3 3)', 'LINESTRING (1.9999 2, -2 5)']

fractopo.general.multiprocess(function_to_call: ~collections.abc.Callable, keyword_arguments: ~collections.abc.Sequence, arguments_identifier=<function <lambda>>, repeats: int = 0) → list[ProcessResult]

Process function calls in parallel.

Returns result as a list where the error is appended when execution fails.

fractopo.general.numpy_to_python_type(value: Any): Convert numpy dtype variable to Python type, if possible.

fractopo.general.point_to_point_unit_vector(point: Point, other_point: Point) → ndarray

Create unit vector from point to other point.

>>> point = Point(0, 0)
>>> other_point = Point(1, 1)
>>> point_to_point_unit_vector(point, other_point)
array([0.70710678, 0.70710678])

fractopo.general.point_to_xy(point: Point) → tuple[float, float]: Get x and y coordinates of Point.

fractopo.general.prepare_geometry_traces(trace_series: GeoSeries) → PreparedGeometry

Prepare trace_series geometries for a faster spatial analysis.

Assumes geometries are LineStrings which are consequently collected into a single MultiLineString which is prepared with shapely.prepared.prep.

>>> traces = gpd.GeoSeries(
...     [LineString([(0, 0), (1, 1)]), LineString([(0, 1), (0, -1)])]
... )
>>> prepare_geometry_traces(traces).context.wkt
'MULTILINESTRING ((0 0, 1 1), (0 1, 0 -1))'

fractopo.general.r2_scorer(y_true: ndarray, y_predicted: ndarray) → float

Score fit with r2 metric.

Changes the scoring to be best at a value of 0.

fractopo.general.raise_determination_error(attribute: str, verb: str = 'determining', determine_target: str = 'topology (=branches and nodes)')

Raise AttributeError if attribute cannot be determined.

>>> try:
...     raise_determination_error("parameters")
...     assert False
... except AttributeError as exc:
...     print(f"{str(exc)[0:20]}...")
...
Cannot determine par...

fractopo.general.random_points_within(poly: Polygon, num_points: int) → list[Point]

Get random points within Polygon.

>>> from pprint import pprint
>>> random.seed(10)
>>> poly = box(0, 0, 1, 1)
>>> result = random_points_within(poly, 2)
>>> pprint([point.within(poly) for point in result])
[True, True]

fractopo.general.read_geofile(path: Path) → GeoDataFrame

Read a filepath for a GeoDataFrame representable geo-object.

Parameters:: path – pathlib.Path to a GeoDataFrame representable spatial file.
Returns:: geopandas.GeoDataFrame read from the file.
Raises:: TypeError – If the file could not be parsed as a GeoDataFrame by geopandas.

fractopo.general.remove_duplicate_caseinsensitive_columns(columns) → set: Remove duplicate columns case-insensitively.

fractopo.general.remove_z_coordinates(geometry: BaseGeometry | BaseMultipartGeometry) → Any

Remove z-coordinates from a geometry.

A shapely geometry should be provided. Output will always be the same type of geometry with the z-coordinates removed.

Parameters:: geometry – Shapely geometry
Returns:: Shapely geometry with the same geometry type

fractopo.general.remove_z_coordinates_from_geodata(geodata: GeoDataFrame | GeoSeries) → GeoDataFrame | GeoSeries: Remove Z-coordinates from geometries in geopandasgeodata.

fractopo.general.replace_coord_in_trace(trace: LineString, index: int, replacement: Point) → LineString: Replace coordinate Point of LineString at index with replacement Point.

fractopo.general.resolve_split_to_ls(geom: LineString, splitter: LineString) → list[LineString]: Resolve split between two LineStrings to only LineString results.

fractopo.general.safe_buffer(geom: Point | LineString | Polygon, radius: float, **kwargs) → Polygon

Get type checked Polygon buffer.

>>> result = safe_buffer(Point(0, 0), 1)
>>> isinstance(result, Polygon), round(result.area, 3)
(True, 3.137)

fractopo.general.sanitize_name(name: str) → str: Return only alphanumeric parts of name string.

fractopo.general.save_fig(fig: Figure, results_dir: Path, name: str) → list[Path]: Save figure as svg image to results dir.

fractopo.general.sd_calc(data)

Calculate standard deviation for radial data.

TODO: Wrong results atm. Needs to take into account real length, not just orientation of unit vector. Calculates standard deviation for radial data (degrees)

E.g.

>>> sd_calc(np.array([2, 5, 8]))
(3.0, 5.00)

Parameters:: data (np.ndarray) – Array of degrees
Returns:: Standard deviation

fractopo.general.silent_output(name: str): General method to silence output from general func.

fractopo.general.spatial_index_intersection(spatial_index: Any, coordinates: tuple[float | int | Real, float | int | Real, float | int | Real, float | int | Real] | tuple[float | int | Real, float | int | Real]) → list[int]

Type-checked spatial index intersection.

TODO: Maybe use spatial_index.quary_bulk for faster execution? TODO: Maybe use a predicate (e.g., “contains”) to specify operation?

fractopo.general.sum_aggregation(values, **_) → float | int | Real: Aggregate by calculating sum.

fractopo.general.total_bounds(geodata: GeoSeries | GeoDataFrame) → tuple[float, float, float, float]

Get total bounds of geodataset.

>>> geodata = gpd.GeoSeries([Point(-10, 10), Point(10, 10)])

>>> total_bounds(geodata)
(-10.0, 10.0, 10.0, 10.0)

fractopo.general.within_bounds(x: float | int | Real, y: float | int | Real, min_x: float | int | Real, min_y: float | int | Real, max_x: float | int | Real, max_y: float | int | Real) → bool

Are x and y within the bounds.

>>> within_bounds(1, 1, 0, 0, 2, 2)
True

fractopo.general.wrap_silence(func)

Wrap function to capture and silence its output.

Used primarily to silence output from powerlaw functions and methods.

fractopo.general.write_geodata(gdf: GeoDataFrame, path: Path, driver: str = 'GeoJSON', allow_list_column_transform: bool = False)

Write geodata with driver.

Default is GeoJSON.

fractopo.general.write_geodataframe(geodataframe: GeoDataFrame, name: str, results_dir: Path): Save geodataframe as GeoPackage, GeoJSON and shapefile.

fractopo.general.write_topodata(gdf: GeoDataFrame, output_path: Path): Write branch or nodes GeoDataFrame to output_path.

fractopo.general.zip_equal(*iterables): Zip iterables of only equal lengths.