Add type field
This commit is contained in:
@@ -23,8 +23,14 @@ class TelemetryController:
|
||||
try:
|
||||
odata_filter = request.args.get("$filter")
|
||||
start_timestamp, end_timestamp = self._extract_dates_from_odata(odata_filter)
|
||||
result = self._database_service.find_by_date("telemetry", start_timestamp, end_timestamp)
|
||||
return jsonify({"success" : True, "data" : result}), 200
|
||||
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
|
||||
except Exception as e:
|
||||
logger.log_info(e)
|
||||
return jsonify({"success": False, "data": None}), 500
|
||||
@@ -33,8 +39,14 @@ class TelemetryController:
|
||||
try:
|
||||
odata_filter = request.args.get("$filter")
|
||||
start_timestamp, end_timestamp = self._extract_dates_from_odata(odata_filter)
|
||||
result = self._database_service.find_by_date("telemetry", start_timestamp, end_timestamp)
|
||||
return jsonify({"success" : True, "data" : result}), 200
|
||||
results = self._database_service.find_by_date(
|
||||
collection="telemetry",
|
||||
filters={"type": "light"},
|
||||
timestamp_field="timestamp",
|
||||
start_time=start_timestamp,
|
||||
end_time=end_timestamp
|
||||
)
|
||||
return jsonify({"success" : True, "data" : results}), 200
|
||||
except Exception as e:
|
||||
logger.log_info(e)
|
||||
return jsonify({"success": False, "data": None}), 500
|
||||
@@ -44,8 +56,14 @@ class TelemetryController:
|
||||
try:
|
||||
odata_filter = request.args.get("$filter")
|
||||
start_timestamp, end_timestamp = self._extract_dates_from_odata(odata_filter)
|
||||
result = self._database_service.find_by_date("telemetry", start_timestamp, end_timestamp)
|
||||
return jsonify({"success" : True, "data" : result}), 200
|
||||
results = self._database_service.find_by_date(
|
||||
collection="telemetry",
|
||||
filters={"type": "gps"},
|
||||
timestamp_field="timestamp",
|
||||
start_time=start_timestamp,
|
||||
end_time=end_timestamp
|
||||
)
|
||||
return jsonify({"success" : True, "data" : results}), 200
|
||||
except Exception as e:
|
||||
logger.log_info(e)
|
||||
return jsonify({"success": False, "data": None}), 500
|
||||
|
||||
@@ -35,7 +35,8 @@ class MongoService:
|
||||
except Exception as e:
|
||||
logger.log_info(e)
|
||||
|
||||
def find_by_date(self, collection: str, filters: dict = None, timestamp_field: str = "created_at",
|
||||
def find_by_date(self, collection: str,
|
||||
filters: dict = None, timestamp_field: str = "created_at",
|
||||
start_time: int = None, end_time: int = None):
|
||||
try:
|
||||
collection = self._db[collection]
|
||||
|
||||
@@ -17,22 +17,23 @@ class MQTTForwarder:
|
||||
self._central_broker = dst_mqtt
|
||||
self._qos = qos
|
||||
|
||||
def start(self, src_topic: str, dst_topic: str):
|
||||
def start(self, src_topic: str, dst_topic: str, sensor_type: str):
|
||||
try:
|
||||
def forward_handler(topic: str, msg: str):
|
||||
forwarded_msg = self.__wrap_data(msg)
|
||||
forwarded_msg = self.__wrap_data(msg, sensor_type)
|
||||
self._central_broker.publish(dst_topic, forwarded_msg, self._qos)
|
||||
|
||||
self._local_broker.subscribe(src_topic, forward_handler)
|
||||
except Exception as e:
|
||||
print(f"An error occurred while forwarding from {src_topic} to {dst_topic}: {e}")
|
||||
|
||||
def __wrap_data(self, msg : str):
|
||||
def __wrap_data(self, msg : str, sensor_type: str):
|
||||
print(repr(msg))
|
||||
result = {}
|
||||
data = json.loads(msg)
|
||||
result["timestamp"] = int(time.time())
|
||||
result["systemId"] = self._client_id
|
||||
result["type"] = sensor_type
|
||||
result["data"] = {}
|
||||
for keys in data:
|
||||
result["data"][keys] = data[keys]
|
||||
|
||||
@@ -10,28 +10,28 @@ class ForwarderService:
|
||||
|
||||
src_broker_service: MQTTService
|
||||
dst_broker_service: MQTTService
|
||||
forwards: dict[MQTTForwarder, tuple[str, str]]
|
||||
forwards: dict[MQTTForwarder, tuple[str, str, str]]
|
||||
|
||||
def __init__(self, src_broker: MQTTService, dst_broker: MQTTService):
|
||||
self.src_broker_service = src_broker
|
||||
self.dst_broker_service = dst_broker
|
||||
self.forwards = {}
|
||||
|
||||
def register_forwarder(self, client_id: str, src_topic: str, dst_topic: str, qos : int) -> None:
|
||||
def register_forwarder(self, client_id: str, sensor_type : str | None, src_topic: str, dst_topic: str, qos : int) -> None:
|
||||
forwarder = MQTTForwarder(
|
||||
client_id,
|
||||
self.src_broker_service,
|
||||
self.dst_broker_service,
|
||||
qos
|
||||
)
|
||||
self.forwards[forwarder] = (src_topic, dst_topic)
|
||||
self.forwards[forwarder] = (src_topic, dst_topic, sensor_type)
|
||||
|
||||
def start_all(self):
|
||||
self.src_broker_service.connect()
|
||||
self.dst_broker_service.connect()
|
||||
|
||||
for forwarder, (src_topic, dst_topic) in self.forwards.items():
|
||||
forwarder.start(src_topic, dst_topic)
|
||||
for forwarder, (src_topic, dst_topic, sensor_type) in self.forwards.items():
|
||||
forwarder.start(src_topic, dst_topic, sensor_type)
|
||||
|
||||
def stop_all(self):
|
||||
self.src_broker_service.disconnect()
|
||||
|
||||
Reference in New Issue
Block a user