Naaturel/board-mate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add rfid queries

Browse Source
This commit is contained in:
Laurent
2026-01-04 16:13:46 +01:00
parent 8da40d70af
commit c9342c9891
1 changed files with 27 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
api-customer/src/controllers/telemetryController.py
View File

@@ -15,21 +15,15 @@ class TelemetryController:
self._database_service = database_service self._database_service = database_service
def _register_routes(self, app): def _register_routes(self, app):
app.add_url_rule("/telemetry/history/rfid", view_func=self.rfid, methods=['GET'])
app.add_url_rule("/telemetry/history/sound", view_func=self.sound, methods=['GET']) app.add_url_rule("/telemetry/history/sound", view_func=self.sound, methods=['GET'])
app.add_url_rule("/telemetry/history/light", view_func=self.light, methods=['GET']) app.add_url_rule("/telemetry/history/light", view_func=self.light, methods=['GET'])
app.add_url_rule("/telemetry/history/gps", view_func=self.gps, methods=['GET']) app.add_url_rule("/telemetry/history/gps", view_func=self.gps, methods=['GET'])
def sound(self): def rfid(self):
try: try:
odata_filter = request.args.get("$filter") odata_filter = request.args.get("$filter")
start_timestamp, end_timestamp = self._extract_dates_from_odata(odata_filter) results = self._query("rfid", odata_filter)
results = self._database_service.find_by_date(
collection="telemetry",
filters={"type": "sound"},
timestamp_field="timestamp",
start_time=start_timestamp,
end_time=end_timestamp
)
return jsonify({"success" : True, "data" : results}), 200 return jsonify({"success" : True, "data" : results}), 200
except Exception as e: except Exception as e:
logger.log_info(e) logger.log_info(e)
@@ -38,17 +32,16 @@ class TelemetryController:
def light(self): def light(self):
try: try:
odata_filter = request.args.get("$filter") odata_filter = request.args.get("$filter")
start_timestamp, end_timestamp = self._extract_dates_from_odata(odata_filter) results = self._query("light", odata_filter)
logger.log_info(start_timestamp) return jsonify({"success" : True, "data" : results}), 200
logger.log_info(end_timestamp) except Exception as e:
results = self._database_service.find_by_date( logger.log_info(e)
collection="telemetry", return jsonify({"success": False, "data": None}), 500
filters={"type": "light"},
timestamp_field="timestamp", def sound(self):
start_time=start_timestamp, try:
end_time=end_timestamp odata_filter = request.args.get("$filter")
) results = self._query("sound", odata_filter)
logger.log_info(f"Returned {len(results)} results ")
return jsonify({"success" : True, "data" : results}), 200 return jsonify({"success" : True, "data" : results}), 200
except Exception as e: except Exception as e:
logger.log_info(e) logger.log_info(e)
@@ -58,23 +51,25 @@ class TelemetryController:
def gps(self): def gps(self):
try: try:
odata_filter = request.args.get("$filter") odata_filter = request.args.get("$filter")
logger.log_info(odata_filter) results = self._query("gps", odata_filter)
start_timestamp, end_timestamp = self._extract_dates_from_odata(odata_filter)
logger.log_info(start_timestamp)
logger.log_info(end_timestamp)
results = self._database_service.find_by_date(
collection="telemetry",
filters={"type": "gps"},
timestamp_field="timestamp",
start_time=start_timestamp,
end_time=end_timestamp
)
logger.log_info(f"Returned {len(results)} results ")
return jsonify({"success" : True, "data" : results}), 200 return jsonify({"success" : True, "data" : results}), 200
except Exception as e: except Exception as e:
logger.log_info(e) logger.log_info(e)
return jsonify({"success": False, "data": None}), 500 return jsonify({"success": False, "data": None}), 500
def _query(self, sensor_type : str, query_filter : str):
start_timestamp, end_timestamp = self._extract_dates_from_odata(query_filter)
logger.log_info(start_timestamp)
logger.log_info(end_timestamp)
results = self._database_service.find_by_date(
collection="telemetry",
filters={"type": "gps"},
timestamp_field="timestamp",
start_time=start_timestamp,
end_time=end_timestamp
)
logger.log_info(f"Returned {len(results)} results ")
return results
def _extract_dates_from_odata(self, filter_str): def _extract_dates_from_odata(self, filter_str):