fractopo package

Subpackages

Submodules

fractopo.branches_and_nodes module

Functions for extracting branches and nodes from trace maps.

branches_and_nodes is the main entrypoint.

fractopo.branches_and_nodes.additional_snapping_func(trace, idx, additional_snapping)

Insert points into LineStrings to make sure trace abutting trace.

E.g.

Return type:: LineString

>>> trace = LineString([(0, 0), (1, 0), (2, 0), (3, 0)])
>>> idx = 0
>>> point = Point(2.25, 0.1)
>>> additional_snapping = [
...     (0, point),
... ]
>>> additional_snapping_func(trace, idx, additional_snapping).wkt
'LINESTRING (0 0, 1 0, 2 0, 2.25 0.1, 3 0)'

When idx doesn’t match -> no additional snapping

>>> trace = LineString([(0, 0), (1, 0), (2, 0), (3, 0)])
>>> idx = 1
>>> point = Point(2.25, 0.1)
>>> additional_snapping = [
...     (0, point),
... ]
>>> additional_snapping_func(trace, idx, additional_snapping).wkt
'LINESTRING (0 0, 1 0, 2 0, 3 0)'

fractopo.branches_and_nodes.angle_to_point(point, nearest_point, comparison_point)

Calculate the angle between two vectors.

Vectors are made from the given points: Both vectors have the same first point, nearest_point, and second point is either point or comparison_point.

Returns angle in degrees.

E.g.

Return type:: float

>>> point = Point(1, 1)
>>> nearest_point = Point(0, 0)
>>> comparison_point = Point(-1, 1)
>>> angle_to_point(point, nearest_point, comparison_point)
90.0

>>> point = Point(1, 1)
>>> nearest_point = Point(0, 0)
>>> comparison_point = Point(-1, 2)
>>> angle_to_point(point, nearest_point, comparison_point)
71.56505117707799

fractopo.branches_and_nodes.determine_branch_identity(number_of_i_nodes, number_of_xy_nodes, number_of_e_nodes)

Determine the identity of a branch.

Is based on the amount of I-, XY- and E-nodes and returns it as a string.

E.g.

Return type:: str

>>> determine_branch_identity(2, 0, 0)
'I - I'

>>> determine_branch_identity(1, 1, 0)
'C - I'

>>> determine_branch_identity(1, 0, 1)
'I - E'

fractopo.branches_and_nodes.determine_insert_approach(nearest_point_idx, trace_point_dists, snap_threshold, point, nearest_point)

Determine if to insert or replace point.

Return type:: tuple[int, bool]

fractopo.branches_and_nodes.filter_non_unique_traces(traces, snap_threshold)

Filter out traces that are not unique.

Return type:: GeoSeries

fractopo.branches_and_nodes.get_branch_identities(branches, nodes, node_identities, snap_threshold)

Determine the types of branches for a GeoSeries of branches.

i.e. C-C, C-I or I-I, + (C-E, E-E, I-E)

Return type:: list[str]

>>> branches = gpd.GeoSeries(
...     [
...         LineString([(1, 1), (2, 2)]),
...         LineString([(2, 2), (3, 3)]),
...         LineString([(3, 0), (2, 2)]),
...         LineString([(2, 2), (-2, 5)]),
...     ]
... )
>>> nodes = gpd.GeoSeries(
...     [
...         Point(2, 2),
...         Point(1, 1),
...         Point(3, 3),
...         Point(3, 0),
...         Point(-2, 5),
...     ]
... )
>>> node_identities = ["X", "I", "I", "I", "E"]
>>> snap_threshold = 0.001
>>> get_branch_identities(branches, nodes, node_identities, snap_threshold)
['C - I', 'C - I', 'C - I', 'C - E']

fractopo.branches_and_nodes.insert_point_to_linestring(trace, point, snap_threshold)

Insert/modify point to trace LineString.

The point location is determined to fit into the LineString without changing the geometrical order of LineString vertices (which only makes sense if LineString is sublinear.)

TODO: Does not work for 2.5D geometries (Z-coordinates). Z-coordinates will be lost.

E.g.

Return type:: LineString

>>> trace = LineString([(0, 0), (1, 0), (2, 0), (3, 0)])
>>> point = Point(1.25, 0.1)
>>> insert_point_to_linestring(trace, point, 0.01).wkt
'LINESTRING (0 0, 1 0, 1.25 0.1, 2 0, 3 0)'

>>> trace = LineString([(0, 0), (1, 0), (2, 0), (3, 0)])
>>> point = Point(2.25, 0.1)
>>> insert_point_to_linestring(trace, point, 0.01).wkt
'LINESTRING (0 0, 1 0, 2 0, 2.25 0.1, 3 0)'

fractopo.branches_and_nodes.is_endpoint_close_to_boundary(endpoint, areas, snap_threshold)

Check if endpoint is within snap_threshold of areas boundaries.

Return type:: bool

fractopo.branches_and_nodes.node_identities_from_branches(branches, areas, snap_threshold)

Resolve node identities from branch data.

Return type:: tuple[list[Point], list[str]]

>>> branches_list = [
...     LineString([(0, 0), (1, 1)]),
...     LineString([(2, 2), (1, 1)]),
...     LineString([(2, 0), (1, 1)]),
... ]
>>> area_polygon = Polygon([(-5, -5), (-5, 5), (5, 5), (5, -5)])
>>> branches = gpd.GeoSeries(branches_list)
>>> areas = gpd.GeoSeries([area_polygon])
>>> snap_threshold = 0.001
>>> nodes, identities = node_identities_from_branches(
...     branches, areas, snap_threshold
... )
>>> [node.wkt for node in nodes]
['POINT (0 0)', 'POINT (1 1)', 'POINT (2 2)', 'POINT (2 0)']
>>> identities
['I', 'Y', 'I', 'I']

fractopo.branches_and_nodes.node_identity(endpoint, idx, areas, endpoints_geoseries, endpoints_spatial_index, snap_threshold)

Determine node identity of endpoint.

Return type:: str

fractopo.branches_and_nodes.report_snapping_loop(loops, allowed_loops): Report snapping looping.

fractopo.branches_and_nodes.resolve_trace_candidates(trace, idx, traces_spatial_index, traces, snap_threshold)

Resolve spatial index intersection to actual intersection candidates.

Return type:: list[LineString]

fractopo.branches_and_nodes.simple_snap(trace, trace_candidates, snap_threshold)

Modify conditionally trace to snap to any of trace_candidates.

E.g.

Return type:: tuple[LineString, bool]

>>> trace = LineString([(0, 0), (1, 0), (2, 0), (3, 0)])
>>> trace_candidates = (
...     [LineString([(3.0001, -3), (3.0001, 0), (3, 3)])]
... )
>>> snap_threshold = 0.001
>>> snapped = simple_snap(trace, trace_candidates, snap_threshold)
>>> snapped[0].wkt, snapped[1]
('LINESTRING (0 0, 1 0, 2 0, 3.0001 0)', True)

Do not snap overlapping.

>>> trace = LineString([(0, 0), (1, 0), (2, 0), (3.0002, 0)])
>>> trace_candidates = (
...     [LineString([(3.0001, -3), (3.0001, 0), (3, 3)])]
... )
>>> snap_threshold = 0.001
>>> snapped = simple_snap(trace, trace_candidates, snap_threshold)
>>> snapped[0].wkt, snapped[1]
('LINESTRING (0 0, 1 0, 2 0, 3.0002 0)', False)

fractopo.branches_and_nodes.snap_others_to_trace(idx, trace, snap_threshold, traces, traces_spatial_index, areas, final_allowed_loop=False)

Determine whether and how to snap trace to traces.

E.g.

Trace gets new coordinates to snap other traces to it:

Return type:: tuple[LineString, bool]

>>> idx = 0
>>> trace = LineString([(0, 0), (1, 0), (2, 0), (3, 0)])
>>> snap_threshold = 0.001
>>> traces = [trace, LineString([(1.5, 3), (1.5, 0.00001)])]
>>> traces_spatial_index = gpd.GeoSeries(traces).sindex
>>> areas = None
>>> snapped = snap_others_to_trace(
...     idx, trace, snap_threshold, traces, traces_spatial_index, areas
... )
>>> snapped[0].wkt, snapped[1]
('LINESTRING (0 0, 1 0, 1.5 1e-05, 2 0, 3 0)', True)

Trace itself is not snapped by snap_others_to_trace:

>>> idx = 0
>>> trace = LineString([(0, 0), (1, 0), (2, 0), (3, 0)])
>>> snap_threshold = 0.001
>>> traces = [trace, LineString([(3.0001, -3), (3.0001, 0), (3, 3)])]
>>> traces_spatial_index = gpd.GeoSeries(traces).sindex
>>> areas = None
>>> snapped = snap_others_to_trace(
...     idx, trace, snap_threshold, traces, traces_spatial_index, areas
... )
>>> snapped[0].wkt, snapped[1]
('LINESTRING (0 0, 1 0, 2 0, 3 0)', False)

fractopo.branches_and_nodes.snap_trace_simple(idx, trace, snap_threshold, traces, traces_spatial_index, final_allowed_loop=False)

Determine whether and how to perform simple snap.

Return type:: tuple[LineString, bool]

fractopo.branches_and_nodes.snap_trace_to_another(trace_endpoints, another, snap_threshold)

Add point to another trace to snap trace to end at another trace.

I.e. modifies and returns another

Return type:: tuple[LineString, bool]

fractopo.branches_and_nodes.snap_traces(traces, snap_threshold, areas=None, final_allowed_loop=False)

Snap traces to end exactly at other traces.

Return type:: tuple[list[LineString], bool]

fractopo.cli module

Command-line integration of fractopo with click.

class fractopo.cli.LogLevel(*values)

Bases: Enum

Enums for log levels.

CRITICAL = 'CRITICAL'

DEBUG = 'DEBUG'

ERROR = 'ERROR'

INFO = 'INFO'

WARNING = 'WARNING'

fractopo.cli.default_network_output_paths(network_name, general_output, branches_output, nodes_output, parameters_output): Determine default network output paths.

fractopo.cli.describe_results(validated, error_column, console=<console width=80 ColorSystem.EIGHT_BIT>): Describe validation results to stdout.

fractopo.cli.fractopo_callback(log_level=<typer.models.OptionInfo object>, version=<typer.models.OptionInfo object>): Use fractopo command-line utilities.

fractopo.cli.get_click_path_args(exists=True, **kwargs): Get basic click path args.

fractopo.cli.info(): Print out information about fractopo installation and python environment.

fractopo.cli.make_output_dir(base_path)

Make timestamped output dir.

Return type:: Path

fractopo.cli.network(trace_file=<typer.models.ArgumentInfo object>, area_file=<typer.models.ArgumentInfo object>, snap_threshold=<typer.models.OptionInfo object>, determine_branches_nodes=<typer.models.OptionInfo object>, name=<typer.models.OptionInfo object>, circular_target_area=<typer.models.OptionInfo object>, truncate_traces=<typer.models.OptionInfo object>, censoring_area=<typer.models.OptionInfo object>, output_path=<typer.models.OptionInfo object>): Analyze the geometry and topology of trace network.

fractopo.cli.rich_table_from_parameters(parameters)

Generate rich Table from network parameters.

Return type:: Table

fractopo.cli.tracevalidate(trace_file=<typer.models.ArgumentInfo object>, area_file=<typer.models.ArgumentInfo object>, allow_fix=<typer.models.OptionInfo object>, summary=<typer.models.OptionInfo object>, snap_threshold=<typer.models.OptionInfo object>, output=<typer.models.OptionInfo object>, only_area_validation=<typer.models.OptionInfo object>, allow_empty_area=<typer.models.OptionInfo object>)

Validate trace data delineated by target area data.

If allow_fix is True, some automatic fixing will be done to e.g. convert MultiLineStrings to LineStrings.

fractopo.fractopo_utils module

Miscellaneous utilities and scripts of fractopo.

class fractopo.fractopo_utils.LineMerge

Bases: object

Merge lines conditionally.

static conditional_linemerge(first, second, tolerance, buffer_value)

Conditionally merge two LineStrings (first and second).

Merge occurs if 1) their endpoints are within buffer_value of each other and 2) their total orientations are within tolerance (degrees) of each other.

Merges by joining their coordinates. The endpoint (that is within buffer_value of endpoint of first) of the second LineString is trimmed from the resulting coordinates.

E.g. with merging:

Return type:: Optional[LineString]

>>> first = LineString([(0, 0), (0, 2)])
>>> second = LineString([(0, 2.001), (0, 4)])
>>> tolerance = 5
>>> buffer_value = 0.01
>>> LineMerge.conditional_linemerge(first, second, tolerance, buffer_value).wkt
'LINESTRING (0 0, 0 2, 0 4)'

Without merging:

>>> first = LineString([(0, 0), (0, 2)])
>>> second = LineString([(0, 2.1), (0, 4)])
>>> tolerance = 5
>>> buffer_value = 0.01
>>> LineMerge.conditional_linemerge(
...     first, second, tolerance, buffer_value
... ) is None
True

static conditional_linemerge_collection(traces, tolerance, buffer_value)

Conditionally linemerge within a collection of LineStrings.

Returns the linemerged traces and the idxs of traces that were linemerged.

E.g.

Return type:: tuple[list[LineString], list[int]]

>>> first = LineString([(0, 0), (0, 2)])
>>> second = LineString([(0, 2.001), (0, 4)])
>>> traces = gpd.GeoSeries([first, second])
>>> tolerance = 5
>>> buffer_value = 0.01
>>> new_traces, idx = LineMerge.conditional_linemerge_collection(
...     traces, tolerance, buffer_value
... )
>>> [trace.wkt for trace in new_traces], idx
(['LINESTRING (0 0, 0 2, 0 4)'], [0, 1])

static integrate_replacements(traces, new_traces, modified_idx)

Add linemerged and remove the parts that were linemerged.

E.g.

Return type:: GeoDataFrame

>>> first = LineString([(0, 0), (0, 2)])
>>> second = LineString([(0, 2.001), (0, 4)])
>>> traces = gpd.GeoDataFrame(geometry=[first, second])
>>> new_traces = [LineString([(0, 0), (0, 2), (0, 4)])]
>>> modified_idx = [0, 1]
>>> LineMerge.integrate_replacements(traces, new_traces, modified_idx)
                     geometry
0  LINESTRING (0 0, 0 2, 0 4)

static run_loop(traces, tolerance, buffer_value)

Run multiple conditional linemerge iterations for GeoDataFrame.

This is the main entrypoint.

GeoDataFrame should contain LineStrings.

E.g.

Return type:: GeoDataFrame

>>> first = LineString([(0, 0), (0, 2)])
>>> second = LineString([(0, 2.001), (0, 4)])
>>> traces = gpd.GeoDataFrame(geometry=[first, second])
>>> tolerance = 5
>>> buffer_value = 0.01
>>> LineMerge.run_loop(traces, tolerance, buffer_value)
                     geometry
0  LINESTRING (0 0, 0 2, 0 4)

fractopo.fractopo_utils.remove_identical_sindex(geosrs, snap_threshold)

Remove stacked nodes by using a search buffer the size of snap_threshold.

Return type:: GeoSeries

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, plot_as_log, unit, needs_topology, aggregator)

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, result, error)

Bases: object

Dataclass for multiprocessing result parsing.

error: bool

identifier: str

result: Any

fractopo.general.assign_branch_and_node_colors(feature_type)

Determine color for each branch and node type.

Defaults to red.

Return type:: str

>>> 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)

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

Parameters:: degrees (float) – Degrees in range 0 - 360
Return type:: float
Returns:: Degrees in range 0 - 180

fractopo.general.azimuth_to_unit_vector(azimuth)

Convert azimuth to unit vector.

Return type:: ndarray

fractopo.general.bool_arrays_sum(arr_1, arr_2)

Calculate integer sum of two arrays.

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

Return type:: ndarray

>>> 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)

Create bounding polygon around GeoSeries.

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

Return type:: Polygon

>>> 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)

Calculate area of circle.

Return type:: float

>>> calc_circle_area(1.78)
9.953822163633902

fractopo.general.calc_circle_radius(area)

Calculate radius from area.

Return type:: float

>>> calc_circle_radius(10.0)
1.7841241161527712

fractopo.general.calc_strike(dip_direction)

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 (float) – The direction of dip.
Return type:: float
Returns:: Converted strike.

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

fractopo.general.check_for_z_coordinates(geodata)

Check if geopandas data contains Z-coordinates.

Return type:: bool

fractopo.general.compare_unit_vector_orientation(vec_1, vec_2, threshold_angle)

If vec_1 and vec_2 are too different in orientation, will return False.

Return type:: bool

fractopo.general.convert_list_columns(gdf, allow=True)

Convert list type columns to string.

Return type:: GeoDataFrame

fractopo.general.create_unit_vector(start_point, end_point)

Create numpy unit vector from two shapely Points.

Parameters:

start_point (Point) – The start point.
end_point (Point) – The end point.

Return type:

ndarray

Returns:

The unit vector that points from start_point to end_point.

fractopo.general.define_length_set(length, set_df)

Define sets based on the length of the traces or branches.

Return type:: str

fractopo.general.determine_azimuth(line, halved)

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 (LineString) – The line of which azimuth is determined.
halved (bool) – Whether to return result in range [0, 180] (halved=True) or [0, 360] (halved=False).

Return type:

float

Returns:

The determined azimuth.

fractopo.general.determine_boundary_intersecting_lines(line_gdf, area_gdf, snap_threshold)

Determine lines that intersect any target area boundary.

Return type:: tuple[ndarray, ndarray]

fractopo.general.determine_regression_azimuth(line)

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 (LineString) – The line of which azimuth is determined.
Return type:: float
Returns:: The determined azimuth in range [0, 180].
Raises:: ValueError – When LinearRegression returns unexpected coefficients.

fractopo.general.determine_set(value, value_ranges, set_names, loop_around)

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 (Union[float, int, Real]) – Value to determine set of.
value_ranges (Sequence[tuple[Union[float, int, Real], Union[float, int, Real]]]) – Ranges of each set.
set_names (Sequence[str]) – Names of each set.
loop_around (bool) – Whether the sets loop around. This is the case for radial data such as azimuths but not the case for length data.

Return type:

str

Returns:

Set string in which value belongs.

Raises:

ValueError – When set value ranges overlap.

fractopo.general.determine_valid_intersection_points(intersection_geoms)

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) – GeoSeries of intersection (Point) geometries.
Return type:: list[Point]
Returns:: The valid intersections (Points).

fractopo.general.determine_valid_intersection_points_no_vnode(trace_candidates, geom)

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.

Return type:: list[Point]

fractopo.general.dissolve_multi_part_traces(traces)

Dissolve MultiLineStrings in GeoDataFrame or GeoSeries.

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

Return type:: Union[GeoDataFrame, GeoSeries]

fractopo.general.extend_bounds(min_x, min_y, max_x, max_y, extend_amount)

Extend bounds by addition and reduction.

Return type:: tuple[Union[float, int, Real], Union[float, int, Real], Union[float, int, Real], Union[float, int, Real]]

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

fractopo.general.fallback_aggregation(values)

Fallback aggregation where values are simply joined into a string.

Return type:: str

fractopo.general.flatten_tuples(list_of_tuples)

Flatten collection of tuples and return index references.

Indexes are from original tuple groupings.

E.g.

Return type:: tuple[list[int], list[Any]]

>>> 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, total_bounds)

Focus plot to given bounds.

Return type:: Axes

fractopo.general.geom_bounds(geom)

Get LineString or Polygon bounds.

Return type:: tuple[Union[float, int, Real], Union[float, int, Real], Union[float, int, Real], Union[float, int, Real]]

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

fractopo.general.get_next_point_in_trace(trace, point)

Determine next coordinate point towards middle of LineString from point.

Return type:: Point

fractopo.general.get_trace_coord_points(trace)

Get all coordinate Points of a LineString.

Return type:: list[Point]

>>> 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)

Return endpoints (shapely.geometry.Point) of a given LineString.

Return type:: tuple[Point, Point]

fractopo.general.intersection_count_to_boundary_weight(intersection_count)

Get actual weight factor for boundary intersection count.

Return type:: int

>>> 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, second, tolerance, halved=True)

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 (Union[float, int, Real]) – First azimuth value to compare.
second (Union[float, int, Real]) – Second azimuth value to compare.
tolerance (Union[float, int, Real]) – Tolerance for closeness.
halved (bool) – Are the azimuths azial (i.e. halved=True) or vectors.

Return type:

bool

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

fractopo.general.is_set(value, value_range, loop_around)

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 (Union[float, int, Real]) – Value to determine.
Tuple[float, (value_range) – The range of values.
loop_around (bool) – Whether the range loops around. This is the case for radial data such as azimuths but not the case for length data.

Return type:

bool

Returns:

Is it within range.

fractopo.general.line_intersection_to_points(first, second)

Perform shapely intersection between two LineStrings.

Enforces only Point returns.

Return type:: list[Point]

fractopo.general.match_crs(first, second)

Match crs between two geopandas data structures.

Return type:: tuple[Union[GeoSeries, GeoDataFrame], Union[GeoSeries, GeoDataFrame]]

>>> 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)

Aggregate by calculating mean.

Return type:: Union[float, int, Real]

fractopo.general.mls_to_ls(multilinestrings)

Flattens a list of multilinestrings to a list of linestrings.

Return type:: list[LineString]

>>> 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, keyword_arguments, arguments_identifier=<function <lambda>>, repeats=0)

Process function calls in parallel.

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

Return type:: list[ProcessResult]

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

fractopo.general.point_to_point_unit_vector(point, other_point)

Create unit vector from point to other point.

Return type:: ndarray

>>> 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)

Get x and y coordinates of Point.

Return type:: tuple[float, float]

fractopo.general.prepare_geometry_traces(trace_series)

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.

Return type:: PreparedGeometry

>>> 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, y_predicted)

Score fit with r2 metric.

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

Return type:: float

fractopo.general.raise_determination_error(attribute, verb='determining', determine_target='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, num_points)

Get random points within Polygon.

Return type:: list[Point]

>>> 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)

Read a filepath for a GeoDataFrame representable geo-object.

Parameters:: path (Path) – pathlib.Path to a GeoDataFrame representable spatial file.
Return type:: GeoDataFrame
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)

Remove duplicate columns case-insensitively.

Return type:: set

fractopo.general.remove_z_coordinates(geometry)

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 (Union[BaseGeometry, BaseMultipartGeometry]) – Shapely geometry
Return type:: Any
Returns:: Shapely geometry with the same geometry type

fractopo.general.remove_z_coordinates_from_geodata(geodata)

Remove Z-coordinates from geometries in geopandasgeodata.

Return type:: Union[GeoDataFrame, GeoSeries]

fractopo.general.replace_coord_in_trace(trace, index, replacement)

Replace coordinate Point of LineString at index with replacement Point.

Return type:: LineString

fractopo.general.resolve_split_to_ls(geom, splitter)

Resolve split between two LineStrings to only LineString results.

Return type:: list[LineString]

fractopo.general.safe_buffer(geom, radius, **kwargs)

Get type checked Polygon buffer.

Return type:: Polygon

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

fractopo.general.sanitize_name(name)

Return only alphanumeric parts of name string.

Return type:: str

fractopo.general.save_fig(fig, results_dir, name)

Save figure as svg image to results dir.

Return type:: list[Path]

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): General method to silence output from general func.

fractopo.general.spatial_index_intersection(spatial_index, coordinates)

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?

Return type:: list[int]

fractopo.general.sum_aggregation(values, **_)

Aggregate by calculating sum.

Return type:: Union[float, int, Real]

fractopo.general.total_bounds(geodata)

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)

Return type:: tuple[float, float, float, float]

fractopo.general.within_bounds(x, y, min_x, min_y, max_x, max_y)

Are x and y within the bounds.

Return type:: bool

>>> 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, path, driver='GeoJSON', allow_list_column_transform=False)

Write geodata with driver.

Default is GeoJSON.

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

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

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

Module contents

fractopo.

Fracture Network Analysis