From af38c81ad5a4ce86200dc2288ad6f794af3e2ae9 Mon Sep 17 00:00:00 2001 From: Maxime Liege Date: Sun, 17 Dec 2023 12:39:34 +0100 Subject: [PATCH] Some code optimization + adding maintenance endpoint --- georideapilib/api.py | 280 +++++++++++++------- georideapilib/objects/GeorideMaintenance.py | 92 +++++++ georideapilib/objects/__init__.py | 1 + 3 files changed, 276 insertions(+), 97 deletions(-) create mode 100644 georideapilib/objects/GeorideMaintenance.py diff --git a/georideapilib/api.py b/georideapilib/api.py index 0096a85..04e8ffb 100644 --- a/georideapilib/api.py +++ b/georideapilib/api.py @@ -5,22 +5,24 @@ Georide api lib import json import logging + import requests -from georideapilib.objects import ( - GeoRideTracker, - GeoRideAccount, - GeoRideUser, - GeoRideTrackerTrip, - GeoRideTrackerPosition, - GeoRideSharedTrip, - GeoRideTrackerBeacon -) from georideapilib.exception import ( LoginException, SeverException, UnauthorizedException ) +from georideapilib.objects import ( + GeoRideTracker, + GeoRideAccount, + GeoRideUser, + GeoRideTrackerTrip, + GeoRideTrackerPosition, + GeoRideSharedTrip, + GeoRideTrackerBeacon, + GeorideMaintenance +) GEORIDE_API_HOST = "https://api.georide.com" GEORIDE_SOCKET_HOST = "https://socket.georide.com" @@ -41,20 +43,25 @@ GEORIDE_API_ENDPOINT_TRACKER_ALARM_OFF = "/tracker/{trackerId}/sonor-alarm/off" GEORIDE_API_ENDPOINT_TRACKER_ALARM_ON = "/tracker/{trackerId}/sonor-alarm/on" GEORIDE_API_ENDPOINT_TRACKER_ECO_MODE_OFF = "/tracker/{trackerId}/eco-mode/off" GEORIDE_API_ENDPOINT_TRACKER_ECO_MODE_ON = "/tracker/{trackerId}/eco-mode/on" +GEORIDE_API_ENDPOINT_TRACKER_DELETE_MAINTENANCE = "/tracker/{trackerId}/maintenance/{maintenanceId}" +GEORIDE_API_ENDPOINT_TRACKER_MAINTENANCE = "/tracker/{trackerId}/maintenance" _SESSION = requests.Session() _LOGGER = logging.getLogger(__name__) + def get_authorisation_token(email, password): """ return an authorization token or no""" data = {"email": email, "password": password} encoded_data = json.dumps(data).encode('utf-8') - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_LOGIN, + response = _send_request( + "POST", + GEORIDE_API_ENDPOINT_LOGIN, + token=None, data=encoded_data, - headers={'Content-Type': 'application/json'}) - + ) + if response.status_code == 200: _LOGGER.debug("Login success") response_data = response.json() @@ -64,17 +71,18 @@ def get_authorisation_token(email, password): raise LoginException(get_authorisation_token, "Login failed") else: _LOGGER.error("Georide login, http error code: %s", response.status_code) - raise SeverException(get_authorisation_token, + raise SeverException(get_authorisation_token, "Georide login, http error code: {}".format(response.status_code)) return account def renew_token(token): """ renew the authorization token """ - headers = {"Authorization": "Bearer " + token} - response = _SESSION.get( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_NEW_TOKEN, - headers=headers) + response = _send_request( + "GET", + GEORIDE_API_ENDPOINT_NEW_TOKEN, + token + ) if response.status_code == 200: _LOGGER.debug("Token updated") response_data = response.json() @@ -84,16 +92,18 @@ def renew_token(token): raise UnauthorizedException(renew_token, "Renew token refused") else: _LOGGER.error("Georide login, http error code: %s", response.status_code) - raise SeverException(renew_token, + raise SeverException(renew_token, "Georide login, http error code: {}".format(response.status_code)) return new_token + def revoke_token(token): """ invalidate the authorization token """ - headers = {"Authorization": "Bearer " + token} - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_LOGOUT, - headers=headers) + response = _send_request( + "POST", + GEORIDE_API_ENDPOINT_LOGOUT, + token + ) if response.status_code == 401: _LOGGER.warnning("Token allready revoked") raise UnauthorizedException(revoke_token, "Token allready revoked") @@ -105,22 +115,24 @@ def revoke_token(token): def get_user(token): """ get the georide user info """ - headers = {"Authorization": "Bearer " + token} - response = _SESSION.get( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_USER, - headers=headers) + response = _send_request( + "GET", + GEORIDE_API_ENDPOINT_USER, + token + ) response_data = response.json() _LOGGER.debug(response_data) account = GeoRideUser.from_json(response_data) - return account + return account + def get_trackers(token): """ get user trackers """ - - headers = {"Authorization": "Bearer " + token} - response = _SESSION.get( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_TRACKERS, - headers=headers) + response = _send_request( + "GET", + GEORIDE_API_ENDPOINT_TRACKERS, + token + ) response_data = response.json() trackers = [] @@ -129,13 +141,14 @@ def get_trackers(token): trackers.append(GeoRideTracker.from_json(json_tracker)) return trackers + def get_tracker_beacons(token, tracker_id): """ get user trackers """ - - headers = {"Authorization": "Bearer " + token} - response = _SESSION.get( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_TRACKER_BEACON.format(trackerId=str(tracker_id)), - headers=headers) + response = _send_request( + "GET", + GEORIDE_API_ENDPOINT_TRACKER_BEACON.format(trackerId=str(tracker_id)), + token + ) response_data = response.json() trackers_beacons = [] @@ -146,13 +159,15 @@ def get_tracker_beacons(token, tracker_id): trackers_beacons.append(GeoRideTrackerBeacon.from_json(json_tracker_beacon)) return trackers_beacons + def get_trips(token, tracker_id, from_date, to_date): """ return all trips between two dates """ - headers = {"Authorization": "Bearer " + token} - response = _SESSION.get( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_TRIPS.format(trackerId=str(tracker_id)), - params={'from': from_date, 'to': to_date}, - headers=headers) + response = _send_request( + "GET", + GEORIDE_API_ENDPOINT_TRIPS.format(trackerId=str(tracker_id)), + token, + {'from': from_date, 'to': to_date} + ) response_data = response.json() trips = [] @@ -161,13 +176,15 @@ def get_trips(token, tracker_id, from_date, to_date): trips.append(GeoRideTrackerTrip.from_json(json_trip)) return trips + def get_positions(token, tracker_id, from_date, to_date): - """ return all trips between two dates """ - headers = {"Authorization": "Bearer " + token} - response = _SESSION.get( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_POSITIONS.format(trackerId=str(tracker_id)), - params={'from': from_date, 'to': to_date}, - headers=headers) + """ return positions from dates """ + response = _send_request( + "GET", + GEORIDE_API_ENDPOINT_POSITIONS.format(trackerId=str(tracker_id)), + token, + {'from': from_date, 'to': to_date} + ) response_data = response.json() positions = [] @@ -176,20 +193,24 @@ def get_positions(token, tracker_id, from_date, to_date): positions.append(GeoRideTrackerPosition.from_json(json_position)) return positions + def share_a_trip_by_trip_id(token, tracker_id, trip_id): """ share trip by trip id """ return _share_a_trip(token, tracker_id, trip_id=trip_id) + def share_a_trip_by_date(token, tracker_id, from_date, to_date): """ share trips between two dates """ return _share_a_trip(token, tracker_id, from_date=from_date, to_date=to_date) + def share_a_trip_by_trip_merge_id(token, tracker_id, trip_merged_id): """ share trip by trip merged id """ return _share_a_trip(token, tracker_id, trip_merged_id=trip_merged_id) -def _share_a_trip(token, tracker_id, trip_id=None, from_date=None, # pylint: disable= R0913 - to_date=None, trip_merged_id=None): + +def _share_a_trip(token, tracker_id, trip_id=None, from_date=None, # pylint: disable= R0913 + to_date=None, trip_merged_id=None): """ share trip by trib_id or between two dates or trip_merged_id """ data = None if trip_id is not None: @@ -201,91 +222,156 @@ def _share_a_trip(token, tracker_id, trip_id=None, from_date=None, # pylint: dis encoded_data = json.dumps(data).encode('utf-8') - headers = { - "Authorization": "Bearer " + token, - 'Content-Type': 'application/json' - } - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_TRIP_SHARE.format(trackerId=str(tracker_id)), - data=encoded_data, - headers=headers) + response = _send_request( + "POST", + GEORIDE_API_ENDPOINT_TRIP_SHARE.format(trackerId=str(tracker_id)), + token, + data=encoded_data + ) response_data = response.json() if response.status_code == 200: return GeoRideSharedTrip.from_json(response_data) return None + def lock_tracker(token, tracker_id): """ used to lock a tracker """ - headers = {"Authorization": "Bearer " + token} - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_LOCK.format(trackerId=str(tracker_id)), - headers=headers) + response = _send_request( + "POST", + GEORIDE_API_ENDPOINT_LOCK.format(trackerId=str(tracker_id)), + token + ) if response.status_code != 204: return False return True + def unlock_tracker(token, tracker_id): """ used to unlock a tracker """ - headers = {"Authorization": "Bearer " + token} - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_UNLOCK.format(trackerId=str(tracker_id)), - headers=headers) + response = _send_request( + "POST", + GEORIDE_API_ENDPOINT_UNLOCK.format(trackerId=str(tracker_id)), + token + ) if response.status_code != 204: return False return True + def toogle_lock_tracker(token, tracker_id): """ used to toggle lock a tracker """ - headers = {"Authorization": "Bearer " + token} - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_TOGGLE_LOCK.format(trackerId=str(tracker_id)), - headers=headers) + response = _send_request( + "POST", + GEORIDE_API_ENDPOINT_TOGGLE_LOCK.format(trackerId=str(tracker_id)), + token + ) response_data = response.json() return response_data['locked'] + def change_tracker_siren_state(token, tracker_id, state: bool): - """ used to toggle lock a tracker """ - headers = {"Authorization": "Bearer " + token} - response = None - if state == True: - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_TRACKER_ALARM_ON.format(trackerId=str(tracker_id)), - headers=headers) - else: - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_TRACKER_ALARM_OFF.format(trackerId=str(tracker_id)), - headers=headers) + """ used to change siren state """ + endpoint = (GEORIDE_API_ENDPOINT_TRACKER_ALARM_ON.format(trackerId=str(tracker_id))) if state \ + else (GEORIDE_API_ENDPOINT_TRACKER_ALARM_OFF.format(trackerId=str(tracker_id))) + response = _send_request( + "POST", + endpoint, + token + ) if response.status_code != 204: return False return True + def change_tracker_eco_mode_state(token, tracker_id, state: bool): - """ used to toggle lock a tracker """ - headers = {"Authorization": "Bearer " + token} - response = None - if state == True: - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_TRACKER_ECO_MODE_ON.format(trackerId=str(tracker_id)), - headers=headers) - else: - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_TRACKER_ECO_MODE_OFF.format(trackerId=str(tracker_id)), - headers=headers) + """ used to toggle eco mode state """ + endpoint = (GEORIDE_API_ENDPOINT_TRACKER_ECO_MODE_ON.format(trackerId=str(tracker_id))) if state \ + else (GEORIDE_API_ENDPOINT_TRACKER_ECO_MODE_OFF.format(trackerId=str(tracker_id))) + response = _send_request( + "POST", + endpoint, + token + ) + if response.status_code != 204: return False return True + +def add_or_update_maintenance(token, tracker_id, maintenance: GeorideMaintenance): + """ used to add or update a maintenance """ + response = _send_request( + "POST", + GEORIDE_API_ENDPOINT_TRACKER_MAINTENANCE.format(trackerId=str(tracker_id)), + token, + data=maintenance + ) + + if response.status_code == 200: + return True + return False + + +def delete_maintenance(token, tracker_id, maintenance_id): + """ used to add or update a maintenance """ + response = _send_request( + "DELETE", + GEORIDE_API_ENDPOINT_TRACKER_DELETE_MAINTENANCE.format( + trackerId=str(tracker_id), + maintenanceId=str(maintenance_id) + ), + token + ) + + if response.status_code == 200: + return True + return False + + +def get_maintenances_for_tracker(token, tracker_id): + """ used to retrieve maintenances """ + response = _send_request( + "GET", + GEORIDE_API_ENDPOINT_TRACKER_MAINTENANCE.format(trackerId=str(tracker_id)), + token + ) + + response_data = response.json() + maintenances = [] + if response.status_code == 200: + for maint in response_data['maintenanceList']: + maintenances.append(GeorideMaintenance.from_json(maint)) + return maintenances + else: + return None + + def shutdown_tracker(token, tracker_id): """ used to toggle lock a tracker """ - headers = {"Authorization": "Bearer " + token} - response = _SESSION.post( - GEORIDE_API_HOST + GEORIDE_API_ENDPOINT_SHUTDOWN_TRACKER.format(trackerId=str(tracker_id)), - headers=headers) + response = _send_request( + "POST", + GEORIDE_API_ENDPOINT_SHUTDOWN_TRACKER.format(trackerId=str(tracker_id)), + token + ) response_data = response.json() if response.status_code == 200: return response_data['locked'] return None + +def _send_request(method, endpoint, token, params=None, data=None): + headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"} if token else { + "Content-Type": "application/json"} + endpoint_url = GEORIDE_API_HOST + endpoint + + if method == "POST": + return _SESSION.post(endpoint_url, data=data, headers=headers) + elif method == "GET": + return _SESSION.get(endpoint_url, params=params, headers=headers) + elif method == "DELETE": + return _SESSION.delete(endpoint_url, headers=headers) + + if __name__ == '__main__': print("Not a main module") diff --git a/georideapilib/objects/GeorideMaintenance.py b/georideapilib/objects/GeorideMaintenance.py new file mode 100644 index 0000000..b3edc59 --- /dev/null +++ b/georideapilib/objects/GeorideMaintenance.py @@ -0,0 +1,92 @@ +import json +import logging + +_LOGGER = logging.getLogger(__name__) + + +class GeorideMaintenance: + + def __init__(self, + name, + lastMaintenanceDistance, + everyMaintenance, + lastMaintenanceDate=None, + id=None, + todo=None, + trackerId=None, + dateUnitType=None, + createdAt=None, + updatedAt=None): + self.id = id + self.name = name + self.todo = todo + self.trackerId = trackerId + self.lastMaintenanceDistance = lastMaintenanceDistance + self.lastMaintenanceDate = lastMaintenanceDate + self.dateUnitType = dateUnitType + self.everyMaintenance = everyMaintenance + self.createdAt = createdAt + self.updatedAt = updatedAt + + @property + def get_id(self): + return self.id + + @property + def get_name(self): + return self.name + + @property + def get_todo(self): + return self.todo + + @property + def tracker_id(self): + return self.trackerId + + @property + def last_maintenance_distance(self): + return self.lastMaintenanceDistance + + @property + def last_maintenance_date(self): + return self.lastMaintenanceDate + + @property + def date_unit_type(self): + return self.dateUnitType + + @property + def every_maintenance(self): + return self.everyMaintenance + + @property + def created_at(self): + return self.createdAt + + @property + def updated_at(self): + return self.updatedAt + + def to_distance_maintenance_json(self): + return json.dumps({ + "name": self.name, + "everyMaintenance": self.everyMaintenance, + "lastMaintenanceDistance": self.lastMaintenanceDistance + }).encode("utf-8") + + @staticmethod + def from_json(json): + """return new object fromjson""" + return GeorideMaintenance( + name=json['name'], + id=json['id'], + todo=json['todo'], + trackerId=json['trackerId'], + lastMaintenanceDistance=json['lastMaintenanceDistance'], + lastMaintenanceDate=json['lastMaintenanceDate'], + dateUnitType=json['dateUnitType'], + everyMaintenance=json['everyMaintenance'], + createdAt=json['createdAt'], + updatedAt=json['updatedAt'] + ) diff --git a/georideapilib/objects/__init__.py b/georideapilib/objects/__init__.py index a19acb5..1dc36ae 100644 --- a/georideapilib/objects/__init__.py +++ b/georideapilib/objects/__init__.py @@ -8,4 +8,5 @@ from .GeoRideTrackerBeacon import * from .GeoRideTrackerPosition import * from .GeoRideTrackerTrip import * from .GeoRideUser import * +from .GeorideMaintenance import *