25 new sensors

sageranger/__init__.py

@@ -7,3 +7,5 @@
 from sageranger.post_monthly import post_monthly_obs
 from sageranger.post_camera_er import post_camera
 from sageranger.post_obs import post_observation
+from sageranger.unpack_info import get_config_info
+from sageranger.sensor_class import SensorInfo 

sageranger/config/sensor_info.yml

@@ -0,0 +1,12 @@
+path_csv: /path/to/file.csv # locally stored on your machine
+event_type: '<type of event>'
+auth_token: 'Bearer <token>' # you must include bearer 
+provider: '<subject provider' # ex: cougarvision
+subject_subtype: <subject subtype> # ex: Camera Trap
+subject_type: 'stationary-object'
+content_type: 'observations.subject'
+source_type: 'tracking-device'
+group_name: '<subject group>' # ex: camera_trap
+# change for sensor type
+# 1=camera 0= temp
+sensor_type: 0

sageranger/post_camera_er.py

@@ -28,59 +28,62 @@
 from datetime import datetime, UTC
 import requests
 import pandas as pd
+from sageranger.unpack_info import get_config_info
+from sageranger.sensor_class import SensorInfo
 from sageranger.post_obs import post_observation
 
 
 def post_camera():  # pylint: disable=too-many-locals
     """Adds Cameras to csv list of camera traps"""
 
-    auth = input('input authorization: ')
+    config = get_config_info(SensorInfo)
 
     hdr = {
-        'Authorization': auth,
+        'Authorization': config.auth_token,
         'Accept': 'application/json'
         }
 
     url = 'https://sagebrush.pamdas.org/api/v1.0/'
 
-    df = pd.read_csv('/path/to/csv',
+    df = pd.read_csv(config.path_csv,
                      delimiter=' ',
                      header=0)
 
-    cam = df.camera.tolist()
+    sen = df.sensor.tolist()
     lat = df.lat.tolist()
     longi = df.longi.tolist()
 
-    for i in enumerate(cam):
+    for i in enumerate(sen):
         i = i[0]
         current_time = datetime.now(UTC)
         formatted_time = current_time.strftime('%Y-%m-%dT%H:%M:%S.%f') + 'Z'
 
         # first create a source
         payload = {
-            "source_type": "seismic",
-            "manufacturer_id": cam[i],
-            "model_name": cam[i],
+            "source_type": config.source_type,
+            "manufacturer_id": sen[i],
+            "model_name": sen[i],
             "additional": {},
-            "provider": "cougar_vision",
+            "provider": config.provider,
             "subject": {
-                "name": cam[i],
-                "subject_subtype": "camera_trap"
+                "name": sen[i],
+                "subject_subtype": config.subject_subtype
             },
             "assigned_range": {}
         }
 
         url_2 = url + 'sources/'
         source = requests.post(url_2, headers=hdr, json=payload, timeout=10)
         response_js = source.json()
+        # print(response_js)
         source_id = response_js['data']['id']
 
         # Then create a subject
         payload = {
-            "content_type": "observations.subject",
-            "name": cam[i],
-            "subject_type": "stationary-object",
-            "subject_subtype": "camera_trap",
+            "content_type": config.content_type,
+            "name": sen[i],
+            "subject_type": config.subject_type,
+            "subject_subtype": config.subject_subtype,
             "additional": {},
             "created_at": formatted_time,
             "updated_at": formatted_time,
@@ -96,7 +99,7 @@ def post_camera():  # pylint: disable=too-many-locals
         payload = {
             "assigned_range": {},
             "source": source_id,
-            "source_type": "tracking-device",
+            "source_type": config.source_type,
             "additional": {},
             "location": {
                 "latitude": lat[i],
@@ -109,8 +112,41 @@ def post_camera():  # pylint: disable=too-many-locals
         response = requests.get(url_4, headers=hdr, timeout=10)
         source_2 = response.json()
 
+        # get subject group id
+        url_5 = url + '/subjectgroups/?group_name=' + config.group_name
+        response = requests.get(url_5, headers=hdr, timeout=20)
+        response_json = response.json()
+        group_id = response_json['data'][0]['id']
+
+        # post subject to subject group
+        payload = [{
+            "id": subject_id,
+            "name": sen[i],
+            "subject_type": config.subject_type,
+            "subject_subtype": config.subject_subtype,
+            "additional": {},
+            "created_at": formatted_time,
+            "updated_at": formatted_time,
+            "is_active": 1,
+            }]
+
+        url_6 = url + 'subjectgroup/' + group_id + '/subjects'
+        subject = requests.post(url_6, headers=hdr, json=payload, timeout=10)
+        response_json = subject.json()
+
+        # get the id of the default subject group
+        url_7 = url + '/subjectgroups/?group_name=Subjects'
+        response = requests.get(url_7, headers=hdr, timeout=20)
+        response_json = response.json()
+        subject_default = response_json['data'][0]['id']
+
+        # delete from default subject group
+        url_8 = url + 'subjectgroup/' + subject_default + '/subjects'
+        _ = requests.delete(url_8, headers=hdr, json=payload, timeout=10)
+
+        # post observation
         post_observation(subject_id, "", formatted_time, hdr)
 
         print("\nsubject id: " + subject_id)
         print("source id: " + source_2['data'][0]['id'])
-        print("camera trap " + cam[i] + " is uploaded to sagebrush\n")
+        print("sensor " + str(sen[i]) + " is uploaded to sagebrush\n")

sageranger/post_cougar_log.py

@@ -34,7 +34,8 @@ def is_target(cam_name, authorization, label):
     current_time = datetime.now(UTC)
     formatted_time = current_time.strftime('%Y-%m-%dT%H:%M:%S.%f') + 'Z'
 
-    url = 'https://sagebrush.pamdas.org/api/v1.0/subjects/?name=' + cam_name
+    url = ('https://sagebrush.pamdas.org/api/v1.0/subjects/?name='
+           + str(cam_name))
 
     response = requests.get(url, headers=headers, timeout=10)
     response_json = response.json()

sageranger/post_obs.py

@@ -8,6 +8,8 @@
 """
 
 import requests
+from unpack_info import get_config_info
+from sensor_class import SensorInfo
 
 
 def post_observation(subject_id, label, time, hdr):
@@ -26,15 +28,15 @@ def post_observation(subject_id, label, time, hdr):
             https://<YOUR INSTANCE>.pamdas.org/api/v1.0/docs/interactive/ )
 
     """
+    config = get_config_info(SensorInfo)
 
     url = 'https://sagebrush.pamdas.org/api/v1.0/'
 
     url_3 = url + 'subject/' + subject_id + '/sources/'
     response = requests.get(url_3, headers=hdr, timeout=10)
     response_json = response.json()
-
     source_id = response_json['data'][0]['id']
-    payload = {
+    payload_camera = {
         "location": {
             "longitude": 0,
             "latitude": 0},
@@ -44,6 +46,40 @@ def post_observation(subject_id, label, time, hdr):
             [{"value": "test", "label": "animal", "units": ""}],
         "additional": {"animal": label}
         }
+
+    payload_temp = {
+        "location": {
+            "longitude": 0,
+            "latitude": 0},
+        "recorded_at": time,
+        "source": source_id,
+        "device_status_properties": [{
+            "value": "",
+            "label": "int_temperature",
+            "units": "C"
+            },
+            {"value": "",
+             "label": "ext_temperature",
+             "units": "C"},
+            {"value": "",
+             "label": "humidity",
+             "units": "%"},
+            {"value": "",
+             "label": "bat_stat",
+             "units": "V"}],
+        "additional": {
+            "int_temperature": "",
+            "ext_temperature": "",
+            "humidity": "",
+            "bat_stat": ""
+            }
+     }
+
+    if config.sensor_type:
+        payload = payload_camera
+    else:
+        payload = payload_temp
+
     url_5 = url + 'observations/'
     obs = requests.post(url_5, headers=hdr, json=payload, timeout=20)
     print("Observation response: ", obs)

sageranger/sensor_class.py

@@ -0,0 +1,31 @@
+"""Sensor Info
+
+This class holds the values from
+the config file. Unpack_info
+handles the logic of mapping the
+variables to their values.
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class SensorInfo:
+    # pylint: disable=too-many-instance-attributes
+    """SensorInfo
+
+    This dataclass declares the
+    config datatypes. The variable
+    names must match those in the
+    config file.
+    """
+
+    path_csv: str
+    auth_token: str
+    source_type: str
+    provider: str
+    subject_subtype: str
+    subject_type: str
+    content_type: str
+    group_name: str
+    sensor_type: bool

sageranger/unpack_info.py

@@ -0,0 +1,59 @@
+"""Unpack Info
+
+This module contains the logice for
+unpacking the config file. It maps
+all the values from the config file
+to the corressponding values in the
+data class sensor_info. These functions
+are used in post_camera_er and in the repo
+cougarvision.
+
+"""
+import argparse
+from dataclasses import fields
+import yaml
+
+
+def parse_args():
+    """Creates parser for config yaml.
+
+    This function creates an arguement parser that creates an
+    args container with the arguement 'CONFIG'.
+
+    Returns:
+        argsparse.Namespace: An object containing all parsed arguement
+            values as attributes (e.g., args.CONFIG).
+    """
+    parser = argparse.ArgumentParser(description='Retrieves information from'
+                                     'config and posts events '
+                                     'and observations')
+    parser.add_argument('CONFIG', type=str, help='path to config file.')
+
+    return parser.parse_args()
+
+
+def get_config_info(class_type):
+    """Parses through config file.
+
+    This function maps values to the dataclasses
+    found in get_info.
+
+    Args:
+        class_type (str): get info has two dataclasses
+            config info and display info
+
+    Return:
+        dict: unpacked and mapped values to
+            class type
+    """
+    args = parse_args()
+    config_path = args.CONFIG
+
+    with open(config_path, 'r', encoding='utf-8') as file:
+        config_dict = yaml.safe_load(file)
+
+    # for direct mapping only use fields in the class fields
+    valid_keys = {f.name for f in fields(class_type)}
+    filtered_keys = {k: v for k, v in config_dict.items() if k in valid_keys}
+
+    return class_type(**filtered_keys)