-
Notifications
You must be signed in to change notification settings - Fork 63
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #114 from usnistgov/4.1.0.dev
4.1.0.dev
- Loading branch information
Showing
123 changed files
with
29,279 additions
and
5,920 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,229 @@ | ||
from copy import deepcopy | ||
|
||
from django import forms | ||
from django.contrib import admin, messages | ||
from django.contrib.admin import register | ||
from django.contrib.admin.decorators import display | ||
from django.contrib.admin.utils import display_for_value | ||
from django.urls import reverse | ||
from django.utils.safestring import mark_safe | ||
|
||
from NEMO.apps.sensors.models import ( | ||
Sensor, | ||
SensorAlertEmail, | ||
SensorAlertLog, | ||
SensorCard, | ||
SensorCardCategory, | ||
SensorCategory, | ||
SensorData, | ||
) | ||
|
||
|
||
def duplicate_sensor_configuration(model_admin, request, queryset): | ||
for sensor in queryset: | ||
original_name = sensor.name | ||
new_name = "Copy of " + sensor.name | ||
try: | ||
existing_sensor = Sensor.objects.filter(name=new_name) | ||
if existing_sensor.exists(): | ||
messages.error( | ||
request, | ||
mark_safe( | ||
f'There is already a copy of {original_name} as <a href="{reverse("admin:sensors_sensor_change", args=[existing_sensor.first().id])}">{new_name}</a>. Change the copy\'s name and try again' | ||
), | ||
) | ||
continue | ||
else: | ||
new_sensor: Sensor = deepcopy(sensor) | ||
new_sensor.name = new_name | ||
new_sensor.read_frequency = 0 | ||
new_sensor.id = None | ||
new_sensor.pk = None | ||
new_sensor.save() | ||
messages.success( | ||
request, | ||
mark_safe( | ||
f'A duplicate of {original_name} has been made as <a href="{reverse("admin:sensors_sensor_change", args=[new_sensor.id])}">{new_sensor.name}</a>' | ||
), | ||
) | ||
except Exception as error: | ||
messages.error( | ||
request, f"{original_name} could not be duplicated because of the following error: {str(error)}" | ||
) | ||
|
||
|
||
def read_selected_sensors(model_admin, request, queryset): | ||
for sensor in queryset: | ||
try: | ||
response = sensor.read_data(raise_exception=True) | ||
if isinstance(response, SensorData): | ||
messages.success(request, f"{sensor} data read: {response.value}") | ||
elif isinstance(response, str): | ||
messages.warning(request, response) | ||
except Exception as error: | ||
messages.error(request, f"{sensor} data could not be read due to the following error: {str(error)}") | ||
|
||
|
||
class SensorCardAdminForm(forms.ModelForm): | ||
class Meta: | ||
model = SensorCard | ||
widgets = {"password": forms.PasswordInput(render_value=True)} | ||
fields = "__all__" | ||
|
||
def clean(self): | ||
if any(self.errors): | ||
return | ||
cleaned_data = super().clean() | ||
category = cleaned_data["category"] | ||
from NEMO.apps.sensors import sensors | ||
|
||
sensors.get(category).clean_sensor_card(self) | ||
return cleaned_data | ||
|
||
|
||
@register(SensorCard) | ||
class SensorCardAdmin(admin.ModelAdmin): | ||
form = SensorCardAdminForm | ||
list_display = ("name", "enabled", "server", "port", "category") | ||
|
||
|
||
class SensorAdminForm(forms.ModelForm): | ||
class Meta: | ||
model = Sensor | ||
fields = "__all__" | ||
|
||
def clean(self): | ||
if any(self.errors): | ||
return | ||
cleaned_data = super().clean() | ||
|
||
card = ( | ||
self.cleaned_data["sensor_card"] | ||
if "sensor_card" in self.cleaned_data | ||
else self.cleaned_data["interlock_card"] | ||
) | ||
if card: | ||
category = card.category | ||
from NEMO.apps.sensors import sensors | ||
|
||
sensors.get(category).clean_sensor(self) | ||
return cleaned_data | ||
|
||
|
||
class SensorCategoryAdminForm(forms.ModelForm): | ||
class Meta: | ||
model = SensorCategory | ||
fields = "__all__" | ||
|
||
def __init__(self, *args, **kwargs): | ||
super().__init__(*args, **kwargs) | ||
if self.instance.pk: | ||
children_ids = [child.id for child in self.instance.all_children()] | ||
self.fields["parent"].queryset = SensorCategory.objects.exclude(id__in=[self.instance.pk, *children_ids]) | ||
|
||
|
||
@register(SensorCategory) | ||
class SensorCategoryAdmin(admin.ModelAdmin): | ||
form = SensorCategoryAdminForm | ||
list_display = ("name", "get_parent", "get_children") | ||
|
||
@display(ordering="children", description="Children") | ||
def get_children(self, category: SensorCategory) -> str: | ||
return mark_safe( | ||
", ".join( | ||
[ | ||
f'<a href="{reverse("admin:sensors_sensorcategory_change", args=[child.id])}">{child.name}</a>' | ||
for child in category.children.all() | ||
] | ||
) | ||
) | ||
|
||
@display(ordering="parent", description="Parent") | ||
def get_parent(self, category: SensorCategory) -> str: | ||
if not category.parent: | ||
return "" | ||
return mark_safe( | ||
f'<a href="{reverse("admin:sensors_sensorcategory_change", args=[category.parent.id])}">{category.parent.name}</a>' | ||
) | ||
|
||
def formfield_for_foreignkey(self, db_field, request, **kwargs): | ||
""" Filter list of potential parents """ | ||
if db_field.name == "parent": | ||
kwargs["queryset"] = SensorCategory.objects.filter() | ||
return super().formfield_for_foreignkey(db_field, request, **kwargs) | ||
|
||
|
||
@register(Sensor) | ||
class SensorAdmin(admin.ModelAdmin): | ||
form = SensorAdminForm | ||
list_display = ( | ||
"id", | ||
"name", | ||
"visible", | ||
"card", | ||
"get_card_enabled", | ||
"sensor_category", | ||
"unit_id", | ||
"read_address", | ||
"number_of_values", | ||
"get_read_frequency", | ||
"get_last_read", | ||
"get_last_read_at", | ||
) | ||
actions = [duplicate_sensor_configuration, read_selected_sensors] | ||
|
||
@display(boolean=True, ordering="sensor_card__enabled", description="Card Enabled") | ||
def get_card_enabled(self, obj: Sensor): | ||
return obj.card.enabled | ||
|
||
@display(description="Last read") | ||
def get_last_read(self, obj: Sensor): | ||
last_data_point = obj.last_data_point() | ||
return last_data_point.value if last_data_point else "" | ||
|
||
@display(description="Last read at") | ||
def get_last_read_at(self, obj: Sensor): | ||
last_data_point = obj.last_data_point() | ||
return last_data_point.created_date if last_data_point else "" | ||
|
||
@display(ordering="read_frequency", description="Read frequency") | ||
def get_read_frequency(self, obj: Sensor): | ||
return obj.read_frequency if obj.read_frequency != 0 else display_for_value(False, "", boolean=True) | ||
|
||
|
||
@register(SensorCardCategory) | ||
class SensorCardCategoryAdmin(admin.ModelAdmin): | ||
list_display = ("name", "key") | ||
|
||
|
||
@register(SensorData) | ||
class SensorDataAdmin(admin.ModelAdmin): | ||
list_display = ("created_date", "sensor", "value", "get_display_value") | ||
date_hierarchy = "created_date" | ||
list_filter = ("sensor", "sensor__sensor_category") | ||
|
||
@display(ordering="sensor__data_prefix", description="Display value") | ||
def get_display_value(self, obj: SensorData): | ||
return obj.display_value() | ||
|
||
|
||
@register(SensorAlertEmail) | ||
class SensorAlertEmailAdmin(admin.ModelAdmin): | ||
list_display = ("sensor", "enabled", "trigger_condition", "trigger_no_data", "additional_emails", "triggered_on") | ||
readonly_fields = ("triggered_on",) | ||
|
||
|
||
@register(SensorAlertLog) | ||
class SensorAlertLogAdmin(admin.ModelAdmin): | ||
list_display = ["id", "time", "sensor", "reset", "value"] | ||
list_filter = ["sensor", "value", "reset"] | ||
date_hierarchy = "time" | ||
|
||
def has_delete_permission(self, request, obj=None): | ||
return False | ||
|
||
def has_add_permission(self, request): | ||
return False | ||
|
||
def has_change_permission(self, request, obj=None): | ||
return False |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,15 @@ | ||
from django.core.validators import validate_email | ||
|
||
from NEMO.decorators import customization | ||
from NEMO.views.customization import CustomizationBase | ||
|
||
|
||
@customization(key="sensors", title="Sensor Data") | ||
class SensorCustomization(CustomizationBase): | ||
variables = {"sensor_default_daterange": "", "sensor_default_refresh_rate": "0", "sensor_alert_emails": ""} | ||
|
||
def validate(self, name, value): | ||
if name == "sensor_alert_emails": | ||
recipients = tuple([e for e in value.split(",") if e]) | ||
for email in recipients: | ||
validate_email(email) |
Oops, something went wrong.