diff --git a/ecoscope/plotting/plot.py b/ecoscope/plotting/plot.py index 092b7a9c..ec743533 100644 --- a/ecoscope/plotting/plot.py +++ b/ecoscope/plotting/plot.py @@ -286,155 +286,6 @@ def speed(trajectory): return fig -def plot_collar_voltage( - relocations, - start_time, - extract_fn=extract_voltage, - output_folder=None, - layout_kwargs=None, - hline_kwargs=None, -): - # @TODO Complete black-box re-write - assigned_range = ( - relocations["extra__subjectsource__assigned_range"] - .apply(pd.Series) - .add_prefix("extra.extra.subjectsource__assigned_range.") - ) - relocations = relocations.merge(assigned_range, right_index=True, left_index=True) - - groups = relocations.groupby(by=["extra__subject__id", "extra__subjectsource__id"]) - - for group, dataframe in groups: - subject_name = relocations.loc[relocations["extra__subject__id"] == group[0]]["extra__subject__name"].unique()[ - 0 - ] - - dataframe["extra__subjectsource__assigned_range__upper"] = pd.to_datetime( - dataframe["extra__subjectsource__assigned_range"].str["upper"], - errors="coerce", - ) - subjectsource_upperbound = dataframe["extra__subjectsource__assigned_range__upper"].unique() - - is_source_active = subjectsource_upperbound >= start_time or pd.isna(subjectsource_upperbound)[0] - if is_source_active: - logger.info(subject_name) - - dataframe = dataframe.sort_values(by=["fixtime"]) - dataframe["voltage"] = np.array(dataframe.apply(extract_fn, axis=1), dtype=np.float64) - - time = dataframe[dataframe.fixtime >= start_time].fixtime.tolist() - voltage = dataframe[dataframe.fixtime >= start_time].voltage.tolist() - - # Calculate the historic voltage - hist_voltage = dataframe[dataframe.fixtime <= start_time].voltage.tolist() - if hist_voltage: - volt_upper, volt_lower = np.nanpercentile(hist_voltage, [97.5, 2.5]) - hist_voltage_mean = np.nanmean(hist_voltage) - else: - volt_upper, volt_lower = np.nan, np.nan - hist_voltage_mean = None - volt_diff = volt_upper - volt_lower - volt_upper = np.full((len(time)), volt_upper, dtype=np.float32) - volt_lower = np.full((len(time)), volt_lower, dtype=np.float32) - - if np.all(volt_diff == 0): - # jitter = np.random.random_sample((len(volt_upper,))) - volt_upper = volt_upper + 0.025 * max(volt_upper) - volt_lower = volt_lower - 0.025 * max(volt_lower) - - if not any(hist_voltage or voltage): - continue - - try: - lower_y = min(np.nanmin(np.array(hist_voltage)), np.nanmin(np.array(voltage))) - upper_y = max(np.nanmax(np.array(hist_voltage)), np.nanmax(np.array(voltage))) - except ValueError: - lower_y = min(hist_voltage or voltage) - upper_y = max(hist_voltage or voltage) - finally: - lower_y = lower_y - 0.1 * lower_y - upper_y = upper_y + 0.1 * upper_y - - if not len(voltage): - continue - - if not layout_kwargs: - layout = go.Layout( - xaxis={"title": "Time"}, - yaxis={"title": "Collar Voltage"}, - margin={"l": 40, "b": 40, "t": 50, "r": 50}, - hovermode="closest", - ) - else: - layout = go.Layout(**layout_kwargs) - - # Add the current voltage - trace = go.Scatter( - x=time, - y=voltage, - fill=None, - showlegend=True, - mode="lines", - line={ - "width": 1, - "shape": "spline", - }, - line_color="rgb(0,0,246)", - marker={ - "colorscale": "Viridis", - "color": voltage, - "colorbar": dict(title="Colorbar"), - "cmax": np.max(voltage), - "cmin": np.min(voltage), - }, - name=subject_name, - ) - - # Add the historical lower HPD value - trace_lower = go.Scatter( - x=time, - y=volt_lower, - fill=None, - line_color="rgba(255,255,255,0)", - mode="lines", - showlegend=False, - ) - - # Add the historical max HPD value - trace_upper = go.Scatter( - x=time, - y=volt_upper, - fill="tonexty", # fill area between trace0 and trace1 - mode="lines", - fillcolor="rgba(0,176,246,0.2)", - line_color="rgba(255,255,255,0)", - showlegend=True, - name="Historic 2.5% - 97.5%", - ) - - fig = go.Figure(layout=layout) - fig.add_trace(trace_lower) - fig.add_trace(trace_upper) - fig.add_trace(trace) - if hist_voltage_mean: - if not hline_kwargs: - fig.add_hline( - y=hist_voltage_mean, - line_dash="dot", - line_width=1.5, - line_color="Red", - annotation_text="Historic Mean", - annotation_position="right", - ) - else: - fig.add_hline(**hline_kwargs) - fig.update_layout(yaxis=dict(range=[lower_y, upper_y])) - if output_folder: - fig.write_image(os.path.join(f"{output_folder}/_{group}_{str(uuid.uuid4())[:4]}.png")) - else: - fig.show() - - def plot_seasonal_dist(ndvi_vals, cuts, bandwidth=0.05, output_file=None): x = ndvi_vals.sort_values().to_numpy().reshape(-1, 1) kde = KernelDensity(kernel="gaussian", bandwidth=bandwidth).fit(x)